diff --git a/README.md b/README.md index 1f09d99..515f4de 100644 --- a/README.md +++ b/README.md @@ -6,11 +6,14 @@ both the interface and the HTTP bridge behind it. Everything it does fits on one screen: -- **Volume** up/down for the Home 400 and the Living Room pair (hold to keep moving) +- **Volume** up/down for the Home 400 and the Living Room pair. A tap lands + on the next multiple of `VOLUME_STEP` — from 23 it goes to 25, not 28 — + so the levels stay round. Hold to keep moving. - **Group** either room with the AVR — the AVR is always the host, so its sound takes over whatever joins it - **Ungroup** either room again, or all of them at once -- **Change the AVR's input**, listed under the names you gave them +- **Change the AVR's input**, listed under the names you gave them, minus + the sources you deleted in the AVR's setup menu ## The kit it assumes @@ -25,7 +28,7 @@ Any other mix works — it is all in `config.py`. ## Install ```bash -cd /home/pi/heos +cd /var/www/html/heos python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt @@ -62,8 +65,13 @@ ROOM_KEYS = ["home400", "living_room_group"] # the cards, in order `HEOS_HOST` only needs to point at **one** device: HEOS is distributed, so any unit can see and control the whole network. `AVR_HOST` must be the AVR itself. -`AVR_INPUT_CODES` narrows the input picker to the sources you actually use -(and sets their order); leave it empty to list everything the AVR reports. +`VOLUME_STEP` is the grid the volume buttons snap to, not simply how much +they add: at 5, a tap moves 23 to 25 and 25 to 30. + +The input picker already leaves out sources switched off in the AVR's own +setup menu (it asks the AVR with `SSSOD ?`). `AVR_INPUT_CODES` narrows it +further to the sources you actually use, and sets their order; leave it empty +to list everything the AVR still has switched on. ## Add it to the iOS home screen @@ -151,13 +159,16 @@ Used by the interface: | --- | --- | | `GET /api/state` | everything the UI draws, in one call | | `GET /api/targets` | every player and group HEOS can see | -| `POST /api/volume` | `{"target": "home400", "delta": 2}` or `{"level": 25}` | +| `POST /api/volume` | `{"target": "home400", "steps": 1}` — taps, snapped to `VOLUME_STEP`. Also takes `delta` (raw points) or `level` (absolute) | | `POST /api/mute` | `{"target": "home400"}` | | `POST /api/group` | `{"target": "home400", "joined": true}` | | `POST /api/group/none` | every room back on its own | | `GET /api/avr/inputs` | your renamed sources | | `POST /api/avr/input` | `{"code": "GAME"}` | +`POST /volume/up` and `/volume/down` take one snapped tap by default; pass +`?step=3` and they move that many raw points instead, as they always did. + The original bridge's endpoints still answer, so existing Shortcuts and scripts keep working: `/targets`, `/volume`, `/volume/{set,up,down,mute}`, `/playback/{play,pause,stop,next,previous}`, `/group/{create,remove}`, diff --git a/app.py b/app.py index 07213dd..2b318be 100644 --- a/app.py +++ b/app.py @@ -112,9 +112,12 @@ def api_volume(): key = _target_from(data) if data.get("level") is not None: return jsonify({"target": key, "level": controller.set_volume(key, int(data["level"]))}) + if data.get("steps") is not None: + # Taps, not points: each one lands on the next multiple of VOLUME_STEP. + return jsonify({"target": key, "level": controller.step_volume(key, int(data["steps"]))}) if data.get("delta") is not None: return jsonify({"target": key, "level": controller.nudge_volume(key, int(data["delta"]))}) - raise ValueError("Provide either 'delta' or 'level'") + raise ValueError("Provide one of 'steps', 'delta' or 'level'") @app.post("/api/mute") @@ -195,15 +198,21 @@ def legacy_set_volume(): @app.post("/volume/up") @handle_errors def legacy_volume_up(): - step = request.args.get("step", default=config.VOLUME_STEP, type=int) - return jsonify({"level": controller.nudge_volume(_target_from(request.args), step)}) + key = _target_from(request.args) + step = request.args.get("step", type=int) + # No ?step= means one tap of the panel's own button, snapping to the + # next multiple; an explicit ?step= stays a raw number of points. + level = controller.step_volume(key, 1) if step is None else controller.nudge_volume(key, step) + return jsonify({"level": level}) @app.post("/volume/down") @handle_errors def legacy_volume_down(): - step = request.args.get("step", default=config.VOLUME_STEP, type=int) - return jsonify({"level": controller.nudge_volume(_target_from(request.args), -step)}) + key = _target_from(request.args) + step = request.args.get("step", type=int) + level = controller.step_volume(key, -1) if step is None else controller.nudge_volume(key, -step) + return jsonify({"level": level}) @app.post("/volume/mute") diff --git a/avr.py b/avr.py index 81ba5a5..6a68b72 100644 --- a/avr.py +++ b/avr.py @@ -163,6 +163,24 @@ def parse_ssfun(lines: list) -> list: return sources +def parse_sssod(lines: list) -> dict: + """Parse `SSSOD ?` output -- 'SSSODTUNER DEL' and friends -- into + {"TUNER": False, "CD": True, ...}, i.e. which sources you have left + switched on in the AVR's own setup menu.""" + usage = {} + for line in lines: + if not line.startswith("SSSOD"): + continue + rest = line[len("SSSOD"):].strip() + if rest in ("END", ""): + continue + code, _, value = rest.rpartition(" ") + code = code.strip() + if code: + usage[code] = value.strip().upper() != "DEL" + return usage + + class AvrControl: """The input list and the current input, in the names you chose.""" @@ -170,14 +188,15 @@ class AvrControl: self.telnet = DenonTelnet(host, port) self.allowed_codes = list(allowed_codes or []) self._inputs = None + self._usage = None @property def connected(self) -> bool: return self.telnet.connected - def inputs(self, refresh: bool = False) -> list: - """Your renamed source list. Cached: it only changes when you - rename something in the AVR's own setup menu.""" + def all_inputs(self, refresh: bool = False) -> list: + """Every source the AVR knows, under your names, deleted ones + included. Cached: it only changes when you edit the setup menu.""" if self._inputs is None or refresh: lines = self.telnet.request( "SSFUN ?", prefix="SSFUN", @@ -187,7 +206,24 @@ class AvrControl: sources = parse_ssfun(lines) if sources: self._inputs = sources - sources = self._inputs or [] + if self._usage is None or refresh: + lines = self.telnet.request( + "SSSOD ?", prefix="SSSOD", + until=lambda line: line.strip() == "SSSOD END", + timeout=3.0, + ) + self._usage = parse_sssod(lines) + return self._inputs or [] + + def inputs(self, refresh: bool = False) -> list: + """What the picker offers: the sources you can actually select. + + Sources you deleted in the AVR's setup menu are left out -- they + are exactly the ones you never want to land on. Anything SSSOD + does not mention is kept, so a model that does not answer that + command shows its whole list rather than nothing at all. + """ + sources = [s for s in self.all_inputs(refresh) if self._usage.get(s["code"], True)] if self.allowed_codes: order = {code: i for i, code in enumerate(self.allowed_codes)} sources = sorted( @@ -208,7 +244,7 @@ class AvrControl: return {"code": code, "name": self.name_for(code)} def name_for(self, code: str) -> str: - for source in self.inputs(): + for source in self.all_inputs(): if source["code"] == code: return source["name"] return code diff --git a/config.py b/config.py index 86de7fd..682cec2 100644 --- a/config.py +++ b/config.py @@ -55,11 +55,14 @@ HOST_KEY = "avr" ROOM_KEYS = ["home400", "living_room_group"] # --- Behaviour -------------------------------------------------------- -# How many HEOS volume points one tap of +/- moves. +# The grid the volume buttons snap to. A tap moves to the next multiple of +# this rather than adding it, so at 5 a level of 23 goes to 25, not 28. VOLUME_STEP = 5 -# Limit the input picker to the sources you actually use, by their SI -# code (see GET /api/avr/inputs for the full list). Empty = show all. +# Sources you deleted in the AVR's setup menu are hidden from the picker +# automatically. This narrows it further to the ones you actually use, by +# their SI code (see GET /api/avr/inputs), and sets their order in the +# list. Empty = every source the AVR still has switched on. AVR_INPUT_CODES = [] # Shown as the app's name on the iOS home screen. diff --git a/controller.py b/controller.py index 046501b..58df416 100644 --- a/controller.py +++ b/controller.py @@ -23,6 +23,25 @@ from heos import HeosClient, HeosError, parse_message MEMBERS_FILE = Path(__file__).with_name("members.json") +def stepped_level(current: int, steps: int, size: int) -> int: + """Where the volume lands after `steps` taps of +/-. + + A tap moves to the next multiple of `size` rather than adding `size`, + so a level of 23 with a step of 5 goes 23 -> 25 -> 30 on the way up + and 23 -> 20 -> 15 on the way down. Levels that are already on a + multiple simply move a whole step. + """ + if steps > 0: + landing = (current // size + steps) * size + elif steps < 0: + aligned = current // size * size + first = current - size if aligned == current else aligned + landing = first - (-steps - 1) * size + else: + return current + return max(0, min(100, landing)) + + class TargetError(ValueError): """We cannot find that room on the network right now.""" @@ -218,6 +237,25 @@ class Controller: ) return level + def step_volume(self, key: str, steps: int) -> int: + """Move by whole taps, landing on multiples of VOLUME_STEP. + + Counted in taps rather than points so the speakers' own level is + what gets rounded, even when the phone's copy of it is a few + seconds stale. + """ + with self._lock: + self._fresh() + handles = self._volume_handles(key) + level = stepped_level( + self._read_volume(*handles[0]), int(steps), self.cfg.VOLUME_STEP + ) + for scope, obj_id in handles: + self.heos.command( + f"{scope}/set_volume", **{self._id_param(scope): obj_id}, level=level + ) + return level + def toggle_mute(self, key: str): with self._lock: self._fresh() diff --git a/demo.py b/demo.py index 85dab1a..c0a00ae 100644 --- a/demo.py +++ b/demo.py @@ -6,6 +6,8 @@ network -- and lets the panel be tested without waking the house up. import time +from controller import stepped_level + class _FakeAvr: INPUTS = [ @@ -74,6 +76,9 @@ class DemoController: def nudge_volume(self, key, delta): return self.set_volume(key, self._volume[key] + int(delta)) + def step_volume(self, key, steps): + return self.set_volume(key, stepped_level(self._volume[key], int(steps), self.cfg.VOLUME_STEP)) + def toggle_mute(self, key): return None diff --git a/static/app.js b/static/app.js index 886d9e0..766cc26 100644 --- a/static/app.js +++ b/static/app.js @@ -58,7 +58,7 @@ els('.room').forEach((node) => { volume: null, grouped: false, available: false, - pending: 0, // taps not yet sent + taps: 0, // button presses not yet sent inflight: false, busy: false, // a grouping change is in flight }; @@ -105,26 +105,37 @@ function holdable(node, action) { ['pointerup', 'pointercancel', 'pointerleave'].forEach((type) => node.addEventListener(type, stop)); } +/* A tap lands on the next multiple of STEP rather than adding STEP, so a + level of 23 goes to 25 on the way up and 20 on the way down. Mirrors + stepped_level() in controller.py -- the panel guesses with it, the + speakers are the ones that decide. */ +function nextLevel(current, direction) { + if (direction > 0) return Math.min(100, (Math.floor(current / STEP) + 1) * STEP); + const aligned = Math.floor(current / STEP) * STEP; + return Math.max(0, aligned === current ? current - STEP : aligned); +} + function nudge(key, direction) { const room = rooms[key]; if (!room.available) return; - room.volume = Math.max(0, Math.min(100, (room.volume ?? 0) + direction * STEP)); - room.pending += direction * STEP; + room.volume = nextLevel(room.volume ?? 0, direction); + room.taps += direction; paintRoom(room); - flushVolume(key); + if (room.taps === 0) refresh(); // taps cancelled out; take the real level + else flushVolume(key); } /* A burst of taps collapses into one call, so holding + does not queue up thirty requests the speakers then have to chew through. */ async function flushVolume(key) { const room = rooms[key]; - if (room.inflight || !room.pending) return; - const delta = room.pending; - room.pending = 0; + if (room.inflight || !room.taps) return; + const steps = room.taps; + room.taps = 0; room.inflight = true; try { - const data = await api('/api/volume', { target: key, delta }); - if (!room.pending) { + const data = await api('/api/volume', { target: key, steps }); + if (!room.taps) { room.volume = data.level; paintRoom(room); } @@ -133,7 +144,7 @@ async function flushVolume(key) { refresh(); } finally { room.inflight = false; - if (room.pending) flushVolume(key); + if (room.taps) flushVolume(key); } } @@ -223,7 +234,7 @@ function render(state) { room.available = incoming.available; room.grouped = incoming.grouped; // Do not stomp on a volume the user is in the middle of changing. - if (!room.pending && !room.inflight) room.volume = incoming.volume; + if (!room.taps && !room.inflight) room.volume = incoming.volume; paintRoom(room); }); @@ -254,7 +265,7 @@ async function refresh() { } function busy() { - return sheetOpen() || Object.values(rooms).some((r) => r.pending || r.inflight || r.busy); + return sheetOpen() || Object.values(rooms).some((r) => r.taps || r.inflight || r.busy); } ui.refresh.addEventListener('click', () => { diff --git a/tests/fakes.py b/tests/fakes.py index b33aea6..a3f662b 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -120,7 +120,9 @@ class FakeHeos(threading.Thread): class FakeAvr(threading.Thread): """A Denon Telnet server that knows SI and SSFUN.""" - SOURCES = [("MPLAY", "Apple TV"), ("GAME", "PlayStation"), ("SAT/CBL", "TV Box")] + SOURCES = [("MPLAY", "Apple TV"), ("GAME", "PlayStation"), + ("SAT/CBL", "TV Box"), ("DVD", "Old DVD")] + DELETED = {"DVD"} # switched off in the AVR's setup menu def __init__(self): super().__init__(daemon=True) @@ -160,6 +162,9 @@ class FakeAvr(threading.Thread): if command == "SSFUN ?": # The real AVR pads the names out with spaces. return [f"SSFUN{code} {name} " for code, name in self.SOURCES] + ["SSFUN END"] + if command == "SSSOD ?": + return [f"SSSOD{code} {'DEL' if code in self.DELETED else 'USE'}" + for code, _ in self.SOURCES] + ["SSSOD END"] if command == "SI?": return [f"SI{self.input}"] if command.startswith("SI"): diff --git a/tests/test_panel.py b/tests/test_panel.py index 883020c..9c3494f 100644 --- a/tests/test_panel.py +++ b/tests/test_panel.py @@ -12,7 +12,7 @@ from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from controller import Controller # noqa: E402 +from controller import Controller, stepped_level # noqa: E402 from tests.fakes import FakeAvr, FakeHeos # noqa: E402 PAIR = {3, 4} # the two Home 200s @@ -33,6 +33,7 @@ def build(tmpdir): "living_room_group": {"label": "Living Room", "heos_name": "Denon Home 200 L"}, }, AVR_INPUT_CODES=[], + VOLUME_STEP=5, MEMBERS_FILE=str(Path(tmpdir) / "members.json"), ) return Controller(cfg), heos, avr @@ -126,6 +127,23 @@ class PanelTest(unittest.TestCase): self.panel.set_volume("home400", 1) self.assertEqual(self.panel.nudge_volume("home400", -9), 0) + # -- volume in whole steps ------------------------------------------- + def test_a_tap_lands_on_the_next_multiple(self): + self.panel.set_volume("home400", 23) + self.assertEqual(self.panel.step_volume("home400", 1), 25) + self.panel.set_volume("home400", 23) + self.assertEqual(self.panel.step_volume("home400", -1), 20) + + def test_a_level_already_on_a_multiple_moves_a_whole_step(self): + self.panel.set_volume("home400", 25) + self.assertEqual(self.panel.step_volume("home400", 1), 30) + self.panel.set_volume("home400", 25) + self.assertEqual(self.panel.step_volume("home400", -1), 20) + + def test_a_burst_of_taps_snaps_once_then_moves_whole_steps(self): + self.panel.set_volume("living_room_group", 23) + self.assertEqual(self.panel.step_volume("living_room_group", 3), 35) + # -- the AVR --------------------------------------------------------- def test_renamed_inputs_and_selection(self): deadline = time.time() + 5 @@ -140,6 +158,13 @@ class PanelTest(unittest.TestCase): {"code": "SAT/CBL", "name": "TV Box"}], ) self.assertEqual(self.panel.avr.current_input(), {"code": "MPLAY", "name": "Apple TV"}) + + # "Old DVD" is deleted in the AVR's setup menu, so the picker skips + # it -- but it keeps its name, in case the AVR is sitting on it. + self.assertNotIn("DVD", [s["code"] for s in self.panel.avr.inputs()]) + self.assertIn({"code": "DVD", "name": "Old DVD"}, self.panel.avr.all_inputs()) + self.assertEqual(self.panel.avr.name_for("DVD"), "Old DVD") + self.assertEqual(self.panel.avr.select_input("GAME"), {"code": "GAME", "name": "PlayStation"}) self.assertEqual(self.avr.input, "GAME") @@ -162,5 +187,27 @@ class PanelTest(unittest.TestCase): self.assertTrue(all(r["available"] is False for r in state["rooms"])) +class StepArithmeticTest(unittest.TestCase): + """The rule on its own -- no speakers involved.""" + + def test_up_from_between_multiples(self): + self.assertEqual([stepped_level(23, n, 5) for n in (1, 2, 3)], [25, 30, 35]) + + def test_down_from_between_multiples(self): + self.assertEqual([stepped_level(23, -n, 5) for n in (1, 2, 3)], [20, 15, 10]) + + def test_from_a_multiple(self): + self.assertEqual(stepped_level(25, 1, 5), 30) + self.assertEqual(stepped_level(25, -1, 5), 20) + + def test_clamped_to_the_ends(self): + self.assertEqual(stepped_level(98, 1, 5), 100) + self.assertEqual(stepped_level(2, -1, 5), 0) + self.assertEqual(stepped_level(0, -1, 5), 0) + + def test_no_taps_changes_nothing(self): + self.assertEqual(stepped_level(23, 0, 5), 23) + + if __name__ == "__main__": unittest.main()