diff --git a/.gitignore b/.gitignore index 2309cc8..fa6cbd4 100644 --- a/.gitignore +++ b/.gitignore @@ -136,3 +136,12 @@ dist .yarn/install-state.gz .pnp.* + +# ---> Python / this project +__pycache__/ +*.py[cod] +.venv/ +venv/ + +# Learned HEOS group membership, written at runtime +members.json diff --git a/README.md b/README.md index d82103a..66232d9 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,180 @@ -# heos +# HEOS panel -Fix Denon Heos interface \ No newline at end of file +A phone-sized web remote for a Denon HEOS system, meant to be added to the +iOS home screen and used instead of the HEOS app. One Flask process serves +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) +- **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 + +## The kit it assumes + +| Room | Device | How HEOS addresses it | +| --- | --- | --- | +| Living Room | 2× Denon Home 200 as an In-Room Group | a **group** (`gid`) | +| Lego Room | Denon Home 400 | a **player** (`pid`) | +| Home Cinema | Denon AVR-X3800H | a **player**, plus Telnet on port 23 | + +Any other mix works — it is all in `config.py`. + +## Install + +```bash +pip3 install -r requirements.txt +python3 app.py +``` + +Then open `http://:5005/`. + +## Configure + +Open `http://:5005/api/targets` and copy the exact `name` HEOS reports +for each device into `TARGETS` in `config.py`. The names come from whatever +you typed in the HEOS app, so they rarely match the model names. + +```python +TARGETS = { + "avr": {"label": "Home Cinema", "heos_name": "Home Cinema"}, + "home400": {"label": "Lego Room", "heos_name": "Lego Room"}, + "living_room_group": {"label": "Living Room", "heos_name": "Denon Home 200 L"}, +} +HOST_KEY = "avr" # always the group host +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. + +## Add it to the iOS home screen + +Open the page in Safari → Share → **Add to Home Screen**. It then launches +full-screen with no browser chrome, which is the point of the exercise. + +Safari will only offer that over plain HTTP on the LAN, which is fine here; +if you ever put it behind a domain name, give it HTTPS. + +## Run it as a service + +```ini +# /etc/systemd/system/heos-panel.service +[Unit] +Description=HEOS panel +After=network-online.target + +[Service] +ExecStart=/usr/bin/python3 /home/pi/heos/app.py +WorkingDirectory=/home/pi/heos +Restart=always +User=pi + +[Install] +WantedBy=multi-user.target +``` + +```bash +sudo systemctl enable --now heos-panel +``` + +## How the grouping actually works + +Worth knowing, because HEOS makes two things easy to get wrong. + +**`set_group` replaces a group wholesale.** There is no "add this player". +Joining a second room therefore re-sends every member of the group, and the +host's `pid` has to come first — that is what makes the AVR the leader whose +content everyone plays. + +**Your Home 200 pair is a group, not a speaker.** Merging it into the AVR +means sending *both* speakers' pids; sending only the leader would leave the +second Home 200 playing on its own. And once merged, the pair's own `gid` +stops existing, so: + +- unmerging re-issues `set_group` with the pair's two pids, rebuilding it +- volume falls back to setting both players directly, since there is no + group volume to set any more + +The panel learns the pair's members the first time it sees them un-merged and +remembers them in `members.json`, which is what lets it rebuild the pair after +a restart. If you would rather pin them down, list them in `config.py`: + +```python +"living_room_group": { + "label": "Living Room", + "heos_name": "Denon Home 200 L", + "players": ["Denon Home 200 L", "Denon Home 200 R"], # leader first +}, +``` + +Leaving a room deliberately does *not* rewrite the AVR's group, so the other +room's music does not restart. + +## Two protocols, not one + +| | HEOS CLI (port 1255) | Denon Telnet (port 23) | +| --- | --- | --- | +| Speaks | JSON, `heos://player/...` | plain text, `SIGAME`, `SSFUN ?` | +| Used for | players, groups, volume | the AVR's **renamed** input list | + +HEOS only knows generic input ids like `inputs/hdmi_in_1`; the names you gave +your sources live in the AVR's own protocol, which is why both are here. + +The Telnet connection is held open, so input changes made with the physical +remote show up in the panel too. Some Denon models only accept **one** Telnet +connection at a time — if another integration (Home Assistant, say) already +holds it, the AVR card will read `offline` while the HEOS half keeps working. + +## HTTP API + +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/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"}` | + +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}`, +`/inputs`, `/input/{set,relay}`, `/avr/{raw,input,inputs}`, `/raw/`. + +Two of them are worth keeping for troubleshooting: + +``` +GET /raw/browse/browse?sid=1027 # any heos:// command, raw reply +GET /avr/raw?cmd=SSFUN ? # any Telnet command, every line back +``` + +## Working on it + +```bash +python3 app.py --demo # fake speakers, real interface +python3 -m unittest discover -s tests -t . # runs against a fake HEOS network +python3 tools/make_icons.py # redraw the home-screen icon +``` + +## Layout + +``` +app.py Flask: the UI, the API, and the old bridge's routes +controller.py what a room is, what grouping means, volume +heos.py HEOS CLI client (persistent socket, reconnects itself) +avr.py Denon Telnet client + the renamed input list +config.py your devices and preferences +demo.py fake speakers for --demo +templates/ static/ the interface +tests/ fake HEOS + AVR servers, and tests against them +``` diff --git a/app.py b/app.py new file mode 100644 index 0000000..07213dd --- /dev/null +++ b/app.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +"""HEOS panel: a phone-sized web remote plus the HTTP bridge it runs on. + + pip3 install -r requirements.txt + python3 app.py # http://:5005/ + +Everything the UI does goes through /api/*. The flatter, query-string +endpoints from the original heos_bridge.py (/volume/up?target=..., and +friends) are still here so existing Shortcuts and scripts keep working. +""" + +import argparse +from functools import wraps + +from flask import Flask, jsonify, render_template, request + +import config +from avr import AvrError +from controller import Controller, TargetError +from heos import HeosError + +app = Flask(__name__) +controller: Controller = None + + +def handle_errors(view): + """One place to turn our three failure modes into sensible JSON.""" + @wraps(view) + def wrapped(*args, **kwargs): + try: + return view(*args, **kwargs) + except (TargetError, ValueError) as exc: + return jsonify({"error": str(exc)}), 400 + except (HeosError, AvrError) as exc: + return jsonify({"error": str(exc)}), 502 + return wrapped + + +@app.after_request +def no_store(response): + """It is a LAN remote: never let a phone show a cached volume, and + never let iOS pin an old copy of the UI to the home screen.""" + response.headers["Cache-Control"] = "no-store" + return response + + +def _payload() -> dict: + return request.get_json(silent=True) or request.form.to_dict() or request.args.to_dict() + + +def _target_from(data: dict, field: str = "target") -> str: + key = data.get(field) + if not key: + raise ValueError(f"Missing '{field}'. Valid rooms: {', '.join(config.TARGETS)}") + if key not in config.TARGETS: + raise ValueError(f"Unknown room '{key}'. Valid rooms: {', '.join(config.TARGETS)}") + return key + + +# --- The UI ----------------------------------------------------------- +@app.route("/") +def index(): + return render_template( + "index.html", + app_name=config.APP_NAME, + host=config.TARGETS[config.HOST_KEY], + rooms=[{"key": key, **config.TARGETS[key]} for key in config.ROOM_KEYS], + step=config.VOLUME_STEP, + ) + + +@app.get("/manifest.webmanifest") +def manifest(): + """Rendered rather than static, so APP_NAME is only written down once.""" + return app.response_class( + render_template("manifest.webmanifest", app_name=config.APP_NAME), + mimetype="application/manifest+json", + ) + + +# --- API the UI talks to ---------------------------------------------- +@app.get("/api/state") +@handle_errors +def api_state(): + return jsonify(controller.state()) + + +@app.get("/api/targets") +@handle_errors +def api_targets(): + """Diagnostic: every player and group HEOS can see, with its exact + name -- this is what you copy into config.py.""" + found = controller.scan() + return jsonify({ + "players": [ + {"kind": "player", "name": p.get("name"), "pid": p.get("pid"), "model": p.get("model", "")} + for p in found["players"] + ], + "groups": [ + {"kind": "group", "name": g.get("name"), "gid": g.get("gid"), + "players": [{"name": m.get("name"), "pid": m.get("pid"), "role": m.get("role")} + for m in g.get("players", [])]} + for g in found["groups"] + ], + }) + + +@app.post("/api/volume") +@handle_errors +def api_volume(): + data = _payload() + 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("delta") is not None: + return jsonify({"target": key, "level": controller.nudge_volume(key, int(data["delta"]))}) + raise ValueError("Provide either 'delta' or 'level'") + + +@app.post("/api/mute") +@handle_errors +def api_mute(): + key = _target_from(_payload()) + controller.toggle_mute(key) + return jsonify({"target": key, "ok": True}) + + +@app.post("/api/group") +@handle_errors +def api_group(): + """Join or leave the AVR's group. The AVR is always the host, so its + content is what the joined rooms start playing.""" + data = _payload() + key = _target_from(data) + joined = data.get("joined") + if isinstance(joined, str): + joined = joined.lower() in ("1", "true", "yes", "on") + if joined is None: + raise ValueError("Provide 'joined': true to merge with the AVR, false to split off") + return jsonify({"joined": controller.join(key) if joined else controller.leave(key)}) + + +@app.post("/api/group/none") +@handle_errors +def api_group_none(): + """Every room back on its own.""" + return jsonify({"joined": controller.set_membership([])}) + + +@app.get("/api/avr/inputs") +@handle_errors +def api_avr_inputs(): + refresh = request.args.get("refresh") in ("1", "true", "yes") + return jsonify(controller.avr.inputs(refresh=refresh)) + + +@app.post("/api/avr/input") +@handle_errors +def api_avr_set_input(): + code = _payload().get("code") + if not code: + raise ValueError("Provide 'code', e.g. GAME -- see GET /api/avr/inputs") + return jsonify(controller.avr.select_input(code)) + + +# --- The original bridge's API, unchanged ----------------------------- +@app.get("/targets") +@handle_errors +def legacy_targets(): + found = controller.scan() + return jsonify( + [{"kind": "player", "name": p.get("name"), "id": p.get("pid"), "model": p.get("model", "")} + for p in found["players"]] + + [{"kind": "group", "name": g.get("name"), "id": g.get("gid"), + "members": [m.get("name") for m in g.get("players", [])]} + for g in found["groups"]] + ) + + +@app.get("/volume") +@handle_errors +def legacy_get_volume(): + return jsonify({"level": controller.volume(_target_from(request.args))}) + + +@app.post("/volume/set") +@handle_errors +def legacy_set_volume(): + level = request.args.get("level", type=int) + if level is None or not 0 <= level <= 100: + raise ValueError("provide integer 'level' between 0 and 100") + return jsonify({"level": controller.set_volume(_target_from(request.args), level)}) + + +@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)}) + + +@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)}) + + +@app.post("/volume/mute") +@handle_errors +def legacy_mute(): + controller.toggle_mute(_target_from(request.args)) + return jsonify({"ok": True}) + + +@app.post("/playback/") +@handle_errors +def legacy_playback(state): + key = _target_from(request.args) + if state in ("play", "pause", "stop"): + controller.play_state(key, state) + return jsonify({"state": state}) + if state in ("next", "previous"): + controller.skip(key, state) + return jsonify({"ok": True}) + raise ValueError(f"Unknown playback action '{state}'") + + +@app.post("/group/create") +@handle_errors +def legacy_group_create(): + host = _target_from(request.args, "host") + members = [m.strip() for m in request.args.get("members", "").split(",") if m.strip()] + if not members: + raise ValueError("provide '?host=&members='") + for key in members: + _target_from({"target": key}) + controller.group_targets(host, members) + return jsonify({"host": host, "members": members}) + + +@app.post("/group/remove") +@handle_errors +def legacy_group_remove(): + controller.ungroup(_target_from(request.args)) + return jsonify({"ok": True}) + + +@app.get("/inputs") +@handle_errors +def legacy_inputs(): + return jsonify(controller.heos_inputs(_target_from(request.args))) + + +@app.post("/input/set") +@handle_errors +def legacy_input_set(): + input_id = request.args.get("input") + if not input_id: + raise ValueError("provide '?input=' -- see GET /inputs for valid values") + controller.play_heos_input(_target_from(request.args), input_id) + return jsonify({"input": input_id}) + + +@app.post("/input/relay") +@handle_errors +def legacy_input_relay(): + input_id = request.args.get("input") + source = request.args.get("from") + if not input_id or not source: + raise ValueError("provide '?from=&input='") + controller.play_heos_input(_target_from(request.args), input_id, _target_from(request.args, "from")) + return jsonify({"input": input_id, "from": source}) + + +@app.get("/raw/") +@handle_errors +def legacy_raw(subpath): + """Diagnostic: forward any heos:// command as-is. + e.g. GET /raw/browse/browse?sid=1027""" + return jsonify(controller.heos.command(subpath, **request.args.to_dict())) + + +@app.get("/avr/raw") +@handle_errors +def legacy_avr_raw(): + cmd = request.args.get("cmd") + if not cmd: + raise ValueError("provide '?cmd='") + return jsonify({"lines": controller.avr.telnet.request(cmd, timeout=3.0)}) + + +@app.get("/avr/inputs") +@handle_errors +def legacy_avr_inputs(): + return jsonify(controller.avr.inputs(refresh=True)) + + +@app.route("/avr/input", methods=["GET", "POST"]) +@handle_errors +def legacy_avr_input(): + if request.method == "GET": + return jsonify(controller.avr.current_input() or {"error": "AVR not reachable"}) + code = request.args.get("input") + if not code: + raise ValueError("provide '?input=', e.g. GAME, TV, CD, AUX1") + return jsonify(controller.avr.select_input(code)) + + +def main(): + global controller + parser = argparse.ArgumentParser(description="HEOS panel") + parser.add_argument("--port", type=int, default=config.WEB_PORT) + parser.add_argument("--demo", action="store_true", + help="run with fake speakers, for working on the UI away from the kit") + args = parser.parse_args() + + if args.demo: + from demo import DemoController + controller = DemoController(config) + else: + controller = Controller(config) + + # 0.0.0.0 so your phone can reach it over the LAN. + app.run(host="0.0.0.0", port=args.port, threaded=True) + + +if __name__ == "__main__": + main() diff --git a/avr.py b/avr.py new file mode 100644 index 0000000..81ba5a5 --- /dev/null +++ b/avr.py @@ -0,0 +1,219 @@ +"""Denon AVR control over the classic Telnet protocol (TCP port 23). + +Nothing to do with HEOS. Commands are short plain-text strings ending in +a bare \\r: "SI?" asks which input is selected, "SIGAME" selects GAME, +"SSFUN ?" lists the sources *under the names you gave them* -- which is +the only reason we bother with this protocol at all, since HEOS only +ever reports generic identifiers like "inputs/hdmi_in_1". + +The AVR also pushes a line at us whenever anything changes, including +changes made from the physical remote. So instead of polling, we hold +the connection open, read continuously, and keep the last value of each +status prefix. Asking for the current input is then free. +""" + +import socket +import threading +import time + +# Status prefixes worth remembering from the AVR's chatter. +_TRACKED = ("SI", "PW", "MV", "MU") + +_LOG_LIMIT = 200 + + +class AvrError(RuntimeError): + pass + + +class DenonTelnet: + """Persistent listener + request/response on one Telnet connection.""" + + def __init__(self, host: str, port: int = 23, connect_timeout: float = 3.0): + self.host = host + self.port = port + self.connect_timeout = connect_timeout + self.status = {} # "SI" -> "MPLAY" + self.connected = False + self.last_error = None + self._sock = None + self._send_lock = threading.Lock() + self._cv = threading.Condition() + self._log = [] # [(seq, line)], newest last + self._seq = 0 + threading.Thread(target=self._listen_forever, daemon=True).start() + + # -- background reader --------------------------------------------- + def _listen_forever(self): + backoff = 1.0 + while True: + try: + self._open() + backoff = 1.0 + self._read_forever() + except OSError as exc: + self._drop(exc) + time.sleep(backoff) + backoff = min(30.0, backoff * 2) + + def _open(self): + sock = socket.create_connection((self.host, self.port), timeout=self.connect_timeout) + sock.settimeout(60.0) + self._sock = sock + self.connected = True + self.last_error = None + # Prime the status cache so the first page load knows the input. + for probe in ("PW?", "SI?"): + self.send(probe) + + def _read_forever(self): + buffer = b"" + while True: + try: + chunk = self._sock.recv(1024) + except socket.timeout: + continue # the AVR is simply quiet; nothing has changed + if not chunk: + raise ConnectionError("AVR closed the connection") + buffer += chunk + while b"\r" in buffer: + raw, buffer = buffer.split(b"\r", 1) + self._ingest(raw.decode("utf-8", "replace").strip()) + + def _drop(self, exc): + self.connected = False + self.last_error = str(exc) + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + self._sock = None + + def _ingest(self, line: str): + if not line: + return + with self._cv: + self._seq += 1 + self._log.append((self._seq, line)) + del self._log[:-_LOG_LIMIT] + for prefix in _TRACKED: + # SSFUN* also starts with 'SS', never with a tracked prefix, + # so a plain startswith is safe here. + if line.startswith(prefix) and len(line) > len(prefix): + self.status[prefix] = line[len(prefix):] + break + self._cv.notify_all() + + # -- sending ------------------------------------------------------- + def send(self, command: str): + sock = self._sock + if sock is None: + raise AvrError(f"AVR at {self.host} is not connected ({self.last_error or 'no connection'})") + with self._send_lock: + sock.sendall(command.encode("utf-8") + b"\r") + time.sleep(0.05) # the AVR wants a beat between commands + + def request(self, command: str, prefix: str = None, until=None, timeout: float = 2.5) -> list: + """Send a command and collect the reply lines it triggers. + + Returns as soon as a matching line arrives (or, with `until`, as + soon as that terminator line does), so a query costs milliseconds + rather than a fixed timeout. + """ + with self._cv: + cursor = self._seq + self.send(command) + + deadline = time.monotonic() + timeout + with self._cv: + while True: + lines = [ + line for seq, line in self._log + if seq > cursor and (prefix is None or line.startswith(prefix)) + ] + if lines and (until is None or any(until(line) for line in lines)): + return lines + remaining = deadline - time.monotonic() + if remaining <= 0: + return lines + self._cv.wait(remaining) + + def recent_lines(self) -> list: + with self._cv: + return [line for _, line in self._log] + + +def parse_ssfun(lines: list) -> list: + """Parse `SSFUN ?` output -- 'SSFUNBD Blu-ray ' and friends -- + into [{"code": "BD", "name": "Blu-ray"}, ...].""" + sources = [] + for line in lines: + if not line.startswith("SSFUN"): + continue + rest = line[len("SSFUN"):] + if rest.strip() in ("END", ""): + continue + parts = rest.split(" ", 1) + if len(parts) != 2: + continue + code, name = parts[0].strip(), parts[1].strip() + if code and name: + sources.append({"code": code, "name": name}) + return sources + + +class AvrControl: + """The input list and the current input, in the names you chose.""" + + def __init__(self, host: str, port: int = 23, allowed_codes=()): + self.telnet = DenonTelnet(host, port) + self.allowed_codes = list(allowed_codes or []) + self._inputs = 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.""" + if self._inputs is None or refresh: + lines = self.telnet.request( + "SSFUN ?", prefix="SSFUN", + until=lambda line: line.strip() == "SSFUN END", + timeout=3.0, + ) + sources = parse_ssfun(lines) + if sources: + self._inputs = sources + sources = self._inputs or [] + if self.allowed_codes: + order = {code: i for i, code in enumerate(self.allowed_codes)} + sources = sorted( + (s for s in sources if s["code"] in order), + key=lambda s: order[s["code"]], + ) + return sources + + def current_input(self) -> dict: + """{"code": "MPLAY", "name": "Apple TV"} -- the name comes from the + cached source list, the code from the AVR's own push messages.""" + code = self.telnet.status.get("SI") + if code is None: + lines = self.telnet.request("SI?", prefix="SI") + code = lines[0][2:] if lines else None + if code is None: + return None + return {"code": code, "name": self.name_for(code)} + + def name_for(self, code: str) -> str: + for source in self.inputs(): + if source["code"] == code: + return source["name"] + return code + + def select_input(self, code: str) -> dict: + self.telnet.request(f"SI{code}", prefix="SI", timeout=1.5) + self.telnet.status["SI"] = code # trust our own command immediately + return {"code": code, "name": self.name_for(code)} diff --git a/config.py b/config.py new file mode 100644 index 0000000..d3b3d7b --- /dev/null +++ b/config.py @@ -0,0 +1,66 @@ +"""Everything you might want to change lives here. + +Copy the values out of `GET /api/targets` (or the old `/targets`) the +first time you run this, so the names below match exactly what HEOS +reports for your own devices. +""" + +# --- Network ---------------------------------------------------------- +# Any ONE HEOS device's IP is enough: HEOS is a distributed system, so +# whichever unit you connect to can see and control every player and +# group on the network. +HEOS_HOST = "192.168.0.10" +HEOS_PORT = 1255 + +# The AVR's classic Denon Telnet control port. Completely separate from +# HEOS, and the only place your *renamed* input list actually lives -- +# HEOS itself only knows a fixed set of generic input identifiers. +AVR_HOST = "192.168.0.10" +AVR_PORT = 23 + +# Port the panel itself listens on. +WEB_PORT = 5005 + +# --- Rooms ------------------------------------------------------------ +# key -> how to find it on the network, and how to label it in the UI. +# +# heos_name : the EXACT name HEOS reports for that player or group. +# players : only for a target that is a HEOS *group* (a stereo pair +# or an In-Room Group). List its member players, leader +# first. Leave it out and the panel learns the members the +# first time it sees the group un-merged, then remembers +# them in members.json -- which is what lets it rebuild the +# pair after you unmerge it from the AVR. +TARGETS = { + "avr": { + "label": "Home Cinema", + "heos_name": "Home Cinema", # AVR-X3800H + }, + "home400": { + "label": "Lego Room", + "heos_name": "Lego Room", # Denon Home 400 + }, + "living_room_group": { + "label": "Living Room", + "heos_name": "Denon Home 200 L", # 2x Denon Home 200, In-Room Group + # "players": ["Denon Home 200 L", "Denon Home 200 R"], + }, +} + +# The AVR is always the group host: its content takes over every room +# that joins, which is the whole point of the merge buttons. +HOST_KEY = "avr" + +# The rooms that get a card with volume + a join/leave button, in order. +ROOM_KEYS = ["home400", "living_room_group"] + +# --- Behaviour -------------------------------------------------------- +# How many HEOS volume points one tap of +/- moves. +VOLUME_STEP = 2 + +# 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. +AVR_INPUT_CODES = [] + +# Shown as the app's name on the iOS home screen. +APP_NAME = "HEOS" diff --git a/controller.py b/controller.py new file mode 100644 index 0000000..046501b --- /dev/null +++ b/controller.py @@ -0,0 +1,394 @@ +"""The actual behaviour of the panel, on top of the two protocol clients. + +The interesting part is grouping. A HEOS "In-Room Group" (your pair of +Home 200s) is addressed by a gid and behaves like one speaker -- until +you merge it into the AVR's group, at which point that gid stops +existing and its two players are just two members of the AVR's group. +Two consequences drive most of the code below: + + * Merging must send EVERY member pid of the pair, not just its leader, + or the second Home 200 gets left behind. + * Unmerging must re-issue set_group with the pair's own pids to put + the pair back together, so we have to remember what they were. +""" + +import json +import threading +import time +from pathlib import Path + +from avr import AvrControl, AvrError +from heos import HeosClient, HeosError, parse_message + +MEMBERS_FILE = Path(__file__).with_name("members.json") + + +class TargetError(ValueError): + """We cannot find that room on the network right now.""" + + +class Controller: + def __init__(self, cfg): + self.cfg = cfg + self.heos = HeosClient(cfg.HEOS_HOST, cfg.HEOS_PORT) + self.avr = AvrControl(cfg.AVR_HOST, cfg.AVR_PORT, cfg.AVR_INPUT_CODES) + self._lock = threading.RLock() + self._members_file = Path(getattr(cfg, "MEMBERS_FILE", MEMBERS_FILE)) + self._learned = _load_learned(self._members_file) + self._players = [] + self._groups = [] + self._scanned_at = 0.0 + threading.Thread(target=self._keep_warm, daemon=True).start() + + def _keep_warm(self): + """HEOS hangs up on idle connections; a heartbeat keeps the first + button press of the evening as quick as the second.""" + while True: + time.sleep(240) + try: + self.heos.heart_beat() + except HeosError: + pass + + # -- network picture ----------------------------------------------- + def scan(self) -> dict: + """Re-read every player and group, and remember what the rooms + are made of while we can see them.""" + with self._lock: + self._players = self.heos.command("player/get_players").get("payload", []) + self._groups = self.heos.command("group/get_groups").get("payload", []) + self._scanned_at = time.monotonic() + self._learn_members() + return {"players": self._players, "groups": self._groups} + + def _fresh(self, max_age: float = 2.0): + if time.monotonic() - self._scanned_at > max_age: + self.scan() + + def _player_pid(self, name: str): + for player in self._players: + if player.get("name") == name: + return player.get("pid") + return None + + def _group_named(self, name: str): + for group in self._groups: + if group.get("name") == name: + return group + return None + + @staticmethod + def _ordered_pids(group: dict) -> list: + """Member pids with the leader first -- HEOS makes the first pid + in a set_group call the leader, so the order is not cosmetic.""" + players = group.get("players", []) + leaders = [p for p in players if p.get("role") == "leader"] + others = [p for p in players if p.get("role") != "leader"] + return [p.get("pid") for p in leaders + others] + + def _learn_members(self): + """Record what each room's group is made of whenever we catch it + standing on its own, so we can rebuild it after a merge.""" + host_pid = self._host_pid() + changed = False + for key in self.cfg.ROOM_KEYS: + if "players" in self.cfg.TARGETS[key]: + continue # configured by hand, nothing to learn + group = self._group_named(self.cfg.TARGETS[key]["heos_name"]) + if not group: + continue + pids = self._ordered_pids(group) + if host_pid in pids: + continue # currently merged with the AVR: not its own shape + if self._learned.get(key) != pids: + self._learned[key] = pids + changed = True + if changed: + _save_learned(self._members_file, self._learned) + + # -- resolving rooms to pids --------------------------------------- + def _host_pid(self): + """The AVR is always a plain player. Resolving it by player name + matters: once it leads a group, HEOS also reports a *group* under + the very same name.""" + target = self.cfg.TARGETS[self.cfg.HOST_KEY] + pid = self._player_pid(target["heos_name"]) + if pid is None: + raise TargetError(f"No HEOS player named '{target['heos_name']}' (the AVR) was found") + return pid + + def member_pids(self, key: str) -> list: + """Every player that makes up a room, leader first.""" + if key not in self.cfg.TARGETS: + raise TargetError(f"Unknown room '{key}'") + if key == self.cfg.HOST_KEY: + return [self._host_pid()] + + target = self.cfg.TARGETS[key] + name = target["heos_name"] + + if "players" in target: + pids = [] + for player_name in target["players"]: + pid = self._player_pid(player_name) + if pid is None: + raise TargetError(f"No HEOS player named '{player_name}' was found") + pids.append(pid) + return pids + + # A group under this name wins over a player under the same name: + # a stereo pair is usually named after its left-hand speaker. + host_pid = self._host_pid() + group = self._group_named(name) + if group: + pids = self._ordered_pids(group) + if host_pid not in pids: + return pids + + known = self._learned.get(key) + if known: + live = {p.get("pid") for p in self._players} + if all(pid in live for pid in known): + return known + + pid = self._player_pid(name) + if pid is not None: + return [pid] + + raise TargetError( + f"No HEOS player or group named '{name}' was found. " + f"Check GET /api/targets for the names HEOS actually reports." + ) + + # -- volume --------------------------------------------------------- + def _volume_handles(self, key: str) -> list: + """Where volume for this room lives right now, as (scope, id). + + A room that is its own HEOS group has a single group volume. Once + it is merged into the AVR's group that gid is gone, and the only + knobs left are the member players' own volumes. + """ + pids = self.member_pids(key) + if len(pids) > 1: + wanted = set(pids) + for group in self._groups: + if {p.get("pid") for p in group.get("players", [])} == wanted: + return [("group", group["gid"])] + return [("player", pid) for pid in pids] + + @staticmethod + def _id_param(scope: str) -> str: + return "pid" if scope == "player" else "gid" + + def _read_volume(self, scope, obj_id) -> int: + reply = self.heos.command(f"{scope}/get_volume", **{self._id_param(scope): obj_id}) + return int(parse_message(reply["heos"]["message"]).get("level", -1)) + + def volume(self, key: str) -> int: + with self._lock: + self._fresh() + scope, obj_id = self._volume_handles(key)[0] + return self._read_volume(scope, obj_id) + + def set_volume(self, key: str, level: int) -> int: + level = max(0, min(100, int(level))) + with self._lock: + self._fresh() + for scope, obj_id in self._volume_handles(key): + self.heos.command( + f"{scope}/set_volume", **{self._id_param(scope): obj_id}, level=level + ) + return level + + def nudge_volume(self, key: str, delta: int) -> int: + """Move volume by delta and report where it landed. + + Absolute rather than HEOS's own volume_up/down, because the UI + coalesces a fast burst of taps into one call and volume_up caps + its step at 10. + """ + with self._lock: + self._fresh() + handles = self._volume_handles(key) + current = self._read_volume(*handles[0]) + level = max(0, min(100, current + int(delta))) + 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() + for scope, obj_id in self._volume_handles(key): + self.heos.command(f"{scope}/toggle_mute", **{self._id_param(scope): obj_id}) + + # -- grouping -------------------------------------------------------- + def _host_group_pids(self) -> set: + """Every pid currently in the AVR's group (empty if it is alone).""" + host_pid = self._host_pid() + for group in self._groups: + pids = {p.get("pid") for p in group.get("players", [])} + if host_pid in pids and len(pids) > 1: + return pids + return set() + + def joined_keys(self) -> list: + joined, host_pids = [], self._host_group_pids() + for key in self.cfg.ROOM_KEYS: + try: + if host_pids & set(self.member_pids(key)): + joined.append(key) + except TargetError: + continue + return joined + + def group_targets(self, host_key: str, member_keys: list) -> list: + """One set_group call: the host's pid first -- that is what makes it + the leader whose content everyone else plays -- then every member + pid of every room listed.""" + with self._lock: + self._fresh() + pids = list(self.member_pids(host_key)) + for key in member_keys: + pids.extend(self.member_pids(key)) + self.heos.command("group/set_group", pid=",".join(str(p) for p in pids)) + return pids + + def ungroup(self, key: str) -> list: + """Stand a room back up on its own. For the Home 200 pair this + re-forms the pair rather than leaving two lone speakers behind.""" + with self._lock: + self._fresh() + pids = self.member_pids(key) + self.heos.command("group/set_group", pid=",".join(str(p) for p in pids)) + return pids + + def join(self, key: str) -> list: + """Add a room to the AVR's group. set_group replaces a group + wholesale, so the call has to name everyone who is already in it + as well as the newcomer.""" + with self._lock: + self.scan() + wanted = set(self.joined_keys()) | {key} + self.group_targets(self.cfg.HOST_KEY, [k for k in self.cfg.ROOM_KEYS if k in wanted]) + self.scan() + return self.joined_keys() + + def leave(self, key: str) -> list: + """Remove one room. Deliberately does not rewrite the AVR's group: + whatever is still joined keeps playing without a hiccup.""" + with self._lock: + self.scan() + if key in self.joined_keys(): + self.ungroup(key) + self.scan() + return self.joined_keys() + + def set_membership(self, joined: list) -> list: + """Make the AVR's group contain exactly these rooms and no others.""" + with self._lock: + self.scan() + wanted = [k for k in self.cfg.ROOM_KEYS if k in joined] + for key in list(self.joined_keys()): + if key not in wanted: + self.ungroup(key) + if wanted: + self.group_targets(self.cfg.HOST_KEY, wanted) + self.scan() + return self.joined_keys() + + # -- playback (kept for the original bridge's API) -------------------- + def playback_pid(self, key: str): + return self.member_pids(key)[0] + + def play_state(self, key: str, state: str): + with self._lock: + self._fresh() + self.heos.command("player/set_play_state", pid=self.playback_pid(key), state=state) + + def skip(self, key: str, direction: str): + with self._lock: + self._fresh() + command = "play_next" if direction == "next" else "play_previous" + self.heos.command(f"player/{command}", pid=self.playback_pid(key)) + + def heos_inputs(self, key: str) -> list: + """Physical inputs as HEOS sees them (generic ids, not your names).""" + with self._lock: + self._fresh() + reply = self.heos.command("browse/browse", sid=self.playback_pid(key)) + return [{"name": i.get("name"), "input_id": i.get("mid")} for i in reply.get("payload", [])] + + def play_heos_input(self, key: str, input_id: str, source_key: str = None): + with self._lock: + self._fresh() + params = {"pid": self.playback_pid(key), "input": input_id} + if source_key: + params["spid"] = self.playback_pid(source_key) + self.heos.command("browse/play_input", **params) + + # -- one snapshot for the UI ------------------------------------------ + def state(self) -> dict: + snapshot = { + "host": {"key": self.cfg.HOST_KEY, "label": self.cfg.TARGETS[self.cfg.HOST_KEY]["label"]}, + "rooms": [], + "avr": {"connected": False, "input": None, "inputs": []}, + "heos_ok": True, + "errors": [], + } + + with self._lock: + try: + self.scan() + joined = set(self.joined_keys()) + for key in self.cfg.ROOM_KEYS: + room = { + "key": key, + "label": self.cfg.TARGETS[key]["label"], + "available": True, + "grouped": key in joined, + "volume": None, + "error": None, + } + try: + scope, obj_id = self._volume_handles(key)[0] + room["volume"] = self._read_volume(scope, obj_id) + except (TargetError, HeosError, KeyError) as exc: + room["available"] = False + room["error"] = str(exc) + snapshot["rooms"].append(room) + except (HeosError, TargetError) as exc: + snapshot["heos_ok"] = False + snapshot["errors"].append(str(exc)) + snapshot["rooms"] = [ + {"key": key, "label": self.cfg.TARGETS[key]["label"], "available": False, + "grouped": False, "volume": None, "error": str(exc)} + for key in self.cfg.ROOM_KEYS + ] + + try: + snapshot["avr"] = { + "connected": self.avr.connected, + "inputs": self.avr.inputs(), + "input": self.avr.current_input(), + } + except (AvrError, OSError) as exc: + snapshot["errors"].append(str(exc)) + + return snapshot + + +def _load_learned(path: Path) -> dict: + try: + return json.loads(path.read_text()) + except (OSError, ValueError): + return {} + + +def _save_learned(path: Path, data: dict): + try: + path.write_text(json.dumps(data, indent=2)) + except OSError: + pass # a read-only checkout just means we re-learn next time diff --git a/demo.py b/demo.py new file mode 100644 index 0000000..85dab1a --- /dev/null +++ b/demo.py @@ -0,0 +1,119 @@ +"""A pretend HEOS network, for `python3 app.py --demo`. + +Lets you work on the interface on a laptop, with no speakers on the +network -- and lets the panel be tested without waking the house up. +""" + +import time + + +class _FakeAvr: + INPUTS = [ + {"code": "MPLAY", "name": "Apple TV"}, + {"code": "GAME", "name": "PlayStation"}, + {"code": "SAT/CBL", "name": "TV Box"}, + {"code": "BD", "name": "Blu-ray"}, + {"code": "TUNER", "name": "Radio"}, + {"code": "PHONO", "name": "Turntable"}, + ] + + def __init__(self, allowed_codes=()): + self.allowed_codes = list(allowed_codes or []) + self.connected = True + self._code = "MPLAY" + + def inputs(self, refresh=False): + sources = list(self.INPUTS) + if self.allowed_codes: + order = {code: i for i, code in enumerate(self.allowed_codes)} + sources = sorted((s for s in sources if s["code"] in order), key=lambda s: order[s["code"]]) + return sources + + def name_for(self, code): + return next((s["name"] for s in self.INPUTS if s["code"] == code), code) + + def current_input(self): + return {"code": self._code, "name": self.name_for(self._code)} + + def select_input(self, code): + time.sleep(0.15) # the real AVR is not instant either + self._code = code + return self.current_input() + + +class DemoController: + def __init__(self, cfg): + self.cfg = cfg + self.avr = _FakeAvr(cfg.AVR_INPUT_CODES) + self.heos = None + self._volume = {key: 22 + 7 * i for i, key in enumerate(cfg.TARGETS)} + self._joined = set() + + # -- what the UI uses --------------------------------------------- + def state(self): + return { + "host": {"key": self.cfg.HOST_KEY, "label": self.cfg.TARGETS[self.cfg.HOST_KEY]["label"]}, + "rooms": [ + {"key": key, "label": self.cfg.TARGETS[key]["label"], "available": True, + "grouped": key in self._joined, "volume": self._volume[key], "error": None} + for key in self.cfg.ROOM_KEYS + ], + "avr": {"connected": True, "inputs": self.avr.inputs(), "input": self.avr.current_input()}, + "heos_ok": True, + "errors": [], + "demo": True, + } + + def volume(self, key): + return self._volume[key] + + def set_volume(self, key, level): + self._volume[key] = max(0, min(100, int(level))) + return self._volume[key] + + def nudge_volume(self, key, delta): + return self.set_volume(key, self._volume[key] + int(delta)) + + def toggle_mute(self, key): + return None + + def join(self, key): + self._joined.add(key) + return self.joined_keys() + + def leave(self, key): + self._joined.discard(key) + return self.joined_keys() + + def set_membership(self, joined): + self._joined = {k for k in self.cfg.ROOM_KEYS if k in joined} + return self.joined_keys() + + def joined_keys(self): + return [k for k in self.cfg.ROOM_KEYS if k in self._joined] + + # -- enough of the rest to keep the legacy routes answering -------- + def scan(self): + return { + "players": [{"name": t["heos_name"], "pid": 1000 + i, "model": "Demo"} + for i, t in enumerate(self.cfg.TARGETS.values())], + "groups": [], + } + + def group_targets(self, host_key, member_keys): + return self.set_membership(member_keys) + + def ungroup(self, key): + return self.leave(key) + + def play_state(self, key, state): + return None + + def skip(self, key, direction): + return None + + def heos_inputs(self, key): + return [{"name": s["name"], "input_id": f"inputs/{s['code'].lower()}"} for s in self.avr.inputs()] + + def play_heos_input(self, key, input_id, source_key=None): + return None diff --git a/heos.py b/heos.py new file mode 100644 index 0000000..0212d83 --- /dev/null +++ b/heos.py @@ -0,0 +1,126 @@ +"""HEOS CLI client (TCP port 1255). + +HEOS is not JSON-RPC or anything else standard: you send one +`heos://group/command?a=1&b=2` line and read back one JSON line. This +keeps the socket open between commands -- the panel fires a command on +every button press, and a fresh TCP handshake per press is what made +the original bridge feel sluggish. +""" + +import json +import socket +import threading +import time +from urllib.parse import unquote_plus + + +class HeosError(RuntimeError): + """The device answered, but said no (or never answered at all).""" + + +def parse_message(message: str) -> dict: + """Turn a HEOS `heos.message` string into a dict. + + e.g. "pid=12345&level=23" -> {"pid": "12345", "level": "23"} + """ + result = {} + for part in message.split("&"): + if "=" in part: + key, value = part.split("=", 1) + result[key] = unquote_plus(value) + elif part: + result[part] = "" + return result + + +class HeosClient: + """One persistent, lock-guarded, self-healing HEOS connection.""" + + def __init__(self, host: str, port: int = 1255, timeout: float = 5.0): + self.host = host + self.port = port + self.timeout = timeout + self._sock = None + self._buffer = b"" + self._lock = threading.Lock() + + # -- connection management ----------------------------------------- + def _connect(self): + self._close() + sock = socket.create_connection((self.host, self.port), timeout=self.timeout) + sock.settimeout(self.timeout) + self._sock = sock + self._buffer = b"" + + def _close(self): + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + self._sock = None + self._buffer = b"" + + def _read_line(self, deadline: float) -> bytes: + while b"\r\n" not in self._buffer: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("no reply from HEOS in time") + self._sock.settimeout(remaining) + chunk = self._sock.recv(4096) + if not chunk: + raise ConnectionError("HEOS closed the connection") + self._buffer += chunk + line, self._buffer = self._buffer.split(b"\r\n", 1) + return line + + # -- the one method everything else goes through ------------------- + def command(self, path: str, **params) -> dict: + """Send `heos://?` and return the parsed reply. + + Retries once on a socket-level problem, because HEOS quietly + drops connections that have been idle for a few minutes. + """ + with self._lock: + for attempt in (1, 2): + try: + if self._sock is None: + self._connect() + return self._exchange(path, params) + except (OSError, ValueError) as exc: + # ValueError covers a desynced stream (bad JSON): in + # both cases the fix is the same, start a fresh socket. + self._close() + if attempt == 2: + raise HeosError(f"cannot reach HEOS at {self.host}: {exc}") from exc + + def _exchange(self, path: str, params: dict) -> dict: + query = "&".join(f"{k}={v}" for k, v in params.items() if v is not None) + command = f"heos://{path}" + (f"?{query}" if query else "") + self._sock.sendall(command.encode("utf-8") + b"\r\n") + + # Browse calls can take a while, hence the generous overall budget. + deadline = time.monotonic() + self.timeout * 3 + while True: + reply = json.loads(self._read_line(deadline).decode("utf-8")) + heos = reply.get("heos", {}) + + if heos.get("command") != path: + continue # an event or a late reply to something else + if "under process" in heos.get("message", ""): + continue # placeholder ack; the real payload follows + if heos.get("result") == "fail": + raise HeosError(_failure_text(heos)) + return reply + + def heart_beat(self): + """Keep the socket warm so the first press after an idle spell is + as fast as the rest.""" + self.command("system/heart_beat") + + +def _failure_text(heos: dict) -> str: + message = parse_message(heos.get("message", "")) + text = message.get("text") or "unknown error" + eid = message.get("eid") + return f"HEOS refused '{heos.get('command')}': {text}" + (f" (eid {eid})" if eid else "") diff --git a/heos_bridge.py b/heos_bridge.py deleted file mode 100644 index 4373534..0000000 --- a/heos_bridge.py +++ /dev/null @@ -1,488 +0,0 @@ -#!/usr/bin/env python3 -""" -HEOS <-> HTTP bridge for a mixed set of HEOS targets: - - 2x Denon Home 200, combined as a HEOS "In-Room Group" -- this is a - GROUP (addressed by "gid"), not a single player, and uses the - heos://group/... commands rather than heos://player/... - - 1x Denon Home 400 -- a normal player (addressed by "pid") - - 1x Denon AVR-X3800H -- also a normal player (its HEOS module has - its own pid) - -You only need to connect to ONE of these devices' CLI port (1255) -- -HEOS is a distributed system, so connecting to any single unit lets -you see and control every player/group on the network. The IP below -just needs to point at *one* of your four devices. - -Runs a tiny Flask server on the Raspberry Pi. Translates simple HTTP -calls into the correct HEOS CLI commands over a raw TCP socket. - -Setup: - pip3 install flask - python3 heos_bridge.py - -Then, to see what HEOS actually calls each of your devices/groups -(needed to fill in TARGETS below correctly): - GET http://:5005/targets - -Once TARGETS is filled in: - GET http://:5005/volume?target=living_room_group - POST http://:5005/volume/set?target=kitchen&level=25 - POST http://:5005/volume/up?target=avr&step=5 - POST http://:5005/volume/down?target=living_room_group&step=5 - POST http://:5005/volume/mute?target=kitchen - POST http://:5005/playback/play?target=avr - POST http://:5005/playback/pause?target=avr - POST http://:5005/playback/stop?target=avr - POST http://:5005/playback/next?target=avr - POST http://:5005/playback/previous?target=avr - GET http://:5005/inputs?target=avr - POST http://:5005/input/set?target=avr&input=inputs/hdmi_in_1 - POST http://:5005/input/relay?target=home400&from=avr&input=inputs/tv - POST http://:5005/group/create?host=avr&members=home400,living_room_group - POST http://:5005/group/remove?target=home400 - -The AVR's classic Telnet control protocol (port 23) is separate from HEOS -(port 1255) and is where its *renamed* input list actually lives -- HEOS -itself only knows a fixed generic set of input identifiers, not your -custom names. These endpoints talk to the AVR directly over Telnet: - GET http://:5005/avr/raw?cmd=SSFUN ? (explore renamed sources) - GET http://:5005/avr/input (current input, via SI?) - POST http://:5005/avr/input?input=GAME (select input, via SIGAME) -""" - -import json -import socket -from flask import Flask, request, jsonify - -# --- Configuration --------------------------------------------------- -# Any one of your devices' IPs works as the connection point. -SPEAKER_IP = "192.168.0.10" # <-- set to any one HEOS device's IP -HEOS_PORT = 1255 - -AVR_IP = "192.168.0.10" # <-- your AVR's IP (same device SPEAKER_IP points to) -AVR_TELNET_PORT = 23 - -# Map a short friendly key (used in the ?target= query param) to the -# EXACT name HEOS shows for that player or group (whatever you named -# it in the HEOS app). -# -# Run GET /targets first, copy the "name" values you see there, and -# paste them in on the right-hand side below. -TARGETS = { - "living_room_group": "Denon Home 200 L", # <-- Denon Home 200 In-Room Group - "home400": "Lego Room", # <-- Denon Home 400 - "avr": "Home Cinema", # <-- AVR-X3800H -} -# ---------------------------------------------------------------------- - -app = Flask(__name__) - -# name -> ("player", pid) or ("group", gid) -_id_cache = {} - -# gid -> a member pid to target for playback commands (groups have no -# play/pause/stop command of their own -- only players do) -_group_leader_pid = {} - - -def _heos_request(command: str, timeout: float = 4.0) -> dict: - """Open a TCP connection, send one HEOS CLI command, read the JSON reply. - - Some commands (notably deeper 'browse' calls) reply immediately with a - placeholder ack ("command under process") and send the real payload as - a second message on the same connection shortly after. We wait for that - follow-up instead of returning the placeholder. - """ - with socket.create_connection((SPEAKER_IP, HEOS_PORT), timeout=timeout) as sock: - sock.sendall((command + "\r\n").encode("utf-8")) - sock.settimeout(timeout) - - def read_one_line(sock_timeout): - sock.settimeout(sock_timeout) - buf = b"" - while b"\r\n" not in buf: - chunk = sock.recv(4096) - if not chunk: - break - buf += chunk - return buf.split(b"\r\n", 1)[0] - - line = read_one_line(timeout) - reply = json.loads(line.decode("utf-8")) - - if "under process" in reply.get("heos", {}).get("message", ""): - line = read_one_line(timeout * 3) # the real payload can take a bit longer - reply = json.loads(line.decode("utf-8")) - - return reply - - -def _refresh_targets() -> list: - """Fetch every player AND every group currently on the HEOS network.""" - _id_cache.clear() - _group_leader_pid.clear() - discovered = [] - - players_reply = _heos_request("heos://player/get_players") - for p in players_reply.get("payload", []): - _id_cache[p["name"]] = ("player", p["pid"]) - discovered.append({"kind": "player", "name": p["name"], "id": p["pid"], - "model": p.get("model", "")}) - - groups_reply = _heos_request("heos://group/get_groups") - for g in groups_reply.get("payload", []): - _id_cache[g["name"]] = ("group", g["gid"]) - members = g.get("players", []) - if members: - # prefer the member with role "leader" if present, else just the first - leader = next((m for m in members if m.get("role") == "leader"), members[0]) - _group_leader_pid[g["gid"]] = leader.get("pid") - discovered.append({"kind": "group", "name": g["name"], "id": g["gid"], - "members": [m.get("name") for m in members]}) - - return discovered - - -def _resolve(target_key: str): - """Resolve a friendly key (from TARGETS) to ('player'|'group', id).""" - if target_key not in TARGETS: - raise ValueError(f"Unknown target '{target_key}'. Valid keys: {', '.join(TARGETS)}") - heos_name = TARGETS[target_key] - - if heos_name not in _id_cache: - _refresh_targets() # id may have changed, e.g. group re-created after a reboot - - if heos_name not in _id_cache: - raise ValueError( - f"No HEOS player or group named '{heos_name}' found. " - f"Check GET /targets for the exact current names." - ) - return _id_cache[heos_name] - - -def _cmd_group(kind: str) -> str: - return "player" if kind == "player" else "group" - - -def _id_param(kind: str) -> str: - return "pid" if kind == "player" else "gid" - - -def _volume_status(kind: str, obj_id) -> dict: - cmd = f"heos://{_cmd_group(kind)}/get_volume?{_id_param(kind)}={obj_id}" - reply = _heos_request(cmd) - message = dict( - part.split("=", 1) for part in reply["heos"]["message"].split("&") if "=" in part - ) - return {"level": int(message.get("level", -1))} - - -@app.route("/targets", methods=["GET"]) -def list_targets(): - """Diagnostic: see every HEOS player/group on the network and its exact name.""" - return jsonify(_refresh_targets()) - - -def _require_target_key(): - key = request.args.get("target") - if not key: - raise ValueError(f"Missing '?target=' query param. Valid keys: {', '.join(TARGETS)}") - return key - - -@app.route("/volume", methods=["GET"]) -def get_volume(): - try: - kind, obj_id = _resolve(_require_target_key()) - return jsonify(_volume_status(kind, obj_id)) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - - -@app.route("/volume/set", methods=["POST"]) -def set_volume(): - level = request.args.get("level", type=int) - if level is None or not (0 <= level <= 100): - return jsonify({"error": "provide integer 'level' between 0 and 100"}), 400 - try: - kind, obj_id = _resolve(_require_target_key()) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://{_cmd_group(kind)}/set_volume?{_id_param(kind)}={obj_id}&level={level}") - return jsonify({"level": level}) - - -@app.route("/volume/up", methods=["POST"]) -def volume_up(): - step = request.args.get("step", default=5, type=int) - try: - kind, obj_id = _resolve(_require_target_key()) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://{_cmd_group(kind)}/volume_up?{_id_param(kind)}={obj_id}&step={step}") - return jsonify(_volume_status(kind, obj_id)) - - -@app.route("/volume/down", methods=["POST"]) -def volume_down(): - step = request.args.get("step", default=5, type=int) - try: - kind, obj_id = _resolve(_require_target_key()) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://{_cmd_group(kind)}/volume_down?{_id_param(kind)}={obj_id}&step={step}") - return jsonify(_volume_status(kind, obj_id)) - - -@app.route("/group/create", methods=["POST"]) -def create_group(): - """Group rooms together. The HOST's currently playing content takes over - the whole group; every other member's own playback is replaced by it. - Order matters: POST /group/create?host=avr&members=home400,living_room_group - makes the AVR the host -- Lego Room and the Living Room pair will start - playing whatever the AVR is playing. Swap host/members to merge the - other way.""" - host_key = request.args.get("host") - members_param = request.args.get("members", "") - member_keys = [m.strip() for m in members_param.split(",") if m.strip()] - - if not host_key or not member_keys: - return jsonify({"error": "provide '?host=&members='"}), 400 - - try: - host_kind, host_id = _resolve(host_key) - host_pid = _playback_pid(host_kind, host_id) - - member_pids = [] - for key in member_keys: - kind, obj_id = _resolve(key) - member_pids.append(_playback_pid(kind, obj_id)) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - - # Host's pid MUST come first -- that's what makes it the leader whose - # content the whole group plays. - pid_list = ",".join(str(p) for p in [host_pid] + member_pids) - _heos_request(f"heos://group/set_group?pid={pid_list}") - return jsonify({"host": host_key, "members": member_keys}) - - -@app.route("/group/remove", methods=["POST"]) -def remove_from_group(): - """Take a room back out of whatever dynamic group it's currently in. - POST /group/remove?target=home400""" - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - # A set_group call with a single pid removes that player from any group. - _heos_request(f"heos://group/set_group?pid={pid}") - return jsonify({"ok": True}) - - -@app.route("/volume/mute", methods=["POST"]) -def toggle_mute(): - try: - kind, obj_id = _resolve(_require_target_key()) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://{_cmd_group(kind)}/toggle_mute?{_id_param(kind)}={obj_id}") - return jsonify({"ok": True}) - - -def _playback_pid(kind: str, obj_id): - """Playback commands only exist under 'player', so a group needs to be - translated to one of its member pids.""" - if kind == "player": - return obj_id - pid = _group_leader_pid.get(obj_id) - if pid is None: - raise ValueError("Could not determine a playable member for this group") - return pid - - -def _set_play_state(state: str): - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://player/set_play_state?pid={pid}&state={state}") - return jsonify({"state": state}) - - -@app.route("/playback/play", methods=["POST"]) -def play(): - return _set_play_state("play") - - -@app.route("/playback/pause", methods=["POST"]) -def pause(): - return _set_play_state("pause") - - -@app.route("/playback/stop", methods=["POST"]) -def stop(): - return _set_play_state("stop") - - -@app.route("/playback/next", methods=["POST"]) -def next_track(): - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://player/play_next?pid={pid}") - return jsonify({"ok": True}) - - -@app.route("/playback/previous", methods=["POST"]) -def previous_track(): - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://player/play_previous?pid={pid}") - return jsonify({"ok": True}) - - -@app.route("/raw/", methods=["GET"]) -def raw(subpath): - """Diagnostic: forward any heos:// command as-is and return the raw reply. - e.g. GET /raw/browse/browse?sid=1027""" - query = request.query_string.decode() - cmd = f"heos://{subpath}" + (f"?{query}" if query else "") - return jsonify(_heos_request(cmd)) - - -@app.route("/inputs", methods=["GET"]) -def list_inputs(): - """List the physical inputs available on a target (mainly useful for the AVR). - Speakers with no physical inputs will just return an empty list.""" - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - # Each device's own sid (as listed under the fixed "HEOS aux inputs" - # source, sid=1027) equals its pid -- browsing that lists its inputs. - reply = _heos_request(f"heos://browse/browse?sid={pid}") - items = reply.get("payload", []) - return jsonify([{"name": i.get("name"), "input_id": i.get("mid")} for i in items]) - - -@app.route("/input/set", methods=["POST"]) -def set_input(): - input_id = request.args.get("input") - if not input_id: - return jsonify({"error": "provide '?input=' -- see GET /inputs for valid values"}), 400 - try: - kind, obj_id = _resolve(_require_target_key()) - pid = _playback_pid(kind, obj_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request(f"heos://browse/play_input?pid={pid}&input={input_id}") - return jsonify({"input": input_id}) - - -@app.route("/input/relay", methods=["POST"]) -def relay_input(): - """Push one device's input onto another, e.g. play the AVR's 'TV' input - on the Home 400: POST /input/relay?target=home400&from=avr&input=inputs/tv""" - input_id = request.args.get("input") - from_key = request.args.get("from") - if not input_id or not from_key: - return jsonify({"error": "provide '?from=&input='"}), 400 - try: - dest_kind, dest_id = _resolve(_require_target_key()) - dest_pid = _playback_pid(dest_kind, dest_id) - src_kind, src_id = _resolve(from_key) - src_pid = _playback_pid(src_kind, src_id) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - _heos_request( - f"heos://browse/play_input?pid={dest_pid}&spid={src_pid}&input={input_id}" - ) - return jsonify({"input": input_id, "from": from_key}) - - -# --- AVR Telnet control (port 23) ------------------------------------- -# Completely separate protocol from HEOS. Commands are short plain-text -# strings terminated by \r (not \r\n), e.g. "SI?" (query input), -# "SIGAME" (select the GAME input), "SSFUN ?" (list renamed sources). -# The AVR sends back one or more lines; we collect everything that -# arrives within a short window since queries can return multiple lines. - -def _denon_telnet_request(command: str, timeout: float = 3.0) -> list: - with socket.create_connection((AVR_IP, AVR_TELNET_PORT), timeout=timeout) as sock: - sock.sendall((command + "\r").encode("utf-8")) - sock.settimeout(timeout) - buf = b"" - try: - while True: - chunk = sock.recv(4096) - if not chunk: - break - buf += chunk - except socket.timeout: - pass # normal: we just stop once nothing more arrives in time - return [line for line in buf.decode("utf-8", errors="replace").split("\r") if line] - - -def _parse_ssfun(lines: list) -> list: - """Parse SSFUN ? output like 'SSFUNBD Blu-ray ' into - [{"code": "BD", "name": "Blu-ray"}, ...], skipping the 'SSFUN END' terminator.""" - result = [] - for line in lines: - if not line.startswith("SSFUN"): - continue - rest = line[len("SSFUN"):] - if rest.strip() == "END": - continue - parts = rest.split(" ", 1) - if len(parts) != 2: - continue - code, name = parts[0], parts[1].strip() - result.append({"code": code, "name": name}) - return result - - -@app.route("/avr/raw", methods=["GET"]) -def avr_raw(): - """Diagnostic: send any raw Telnet command to the AVR and see every - line it sends back, e.g. GET /avr/raw?cmd=SSFUN ?""" - cmd = request.args.get("cmd") - if not cmd: - return jsonify({"error": "provide '?cmd='"}), 400 - return jsonify({"lines": _denon_telnet_request(cmd)}) - - -@app.route("/avr/input", methods=["GET"]) -def avr_get_input(): - lines = _denon_telnet_request("SI?") - return jsonify({"lines": lines}) - - -@app.route("/avr/inputs", methods=["GET"]) -def avr_list_inputs(): - """Friendly parsed version of /avr/raw?cmd=SSFUN ? -- your actual - renamed input list, with the SI codes to use with /avr/input.""" - lines = _denon_telnet_request("SSFUN ?") - return jsonify(_parse_ssfun(lines)) - - -@app.route("/avr/input", methods=["POST"]) -def avr_set_input(): - input_code = request.args.get("input") - if not input_code: - return jsonify({"error": "provide '?input=', e.g. GAME, TV, CD, AUX1"}), 400 - lines = _denon_telnet_request(f"SI{input_code}") - return jsonify({"input": input_code, "lines": lines}) - - -if __name__ == "__main__": - # 0.0.0.0 so your phone can reach it over the LAN - app.run(host="0.0.0.0", port=5005) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..001e7c4 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +flask>=3.0 diff --git a/static/app.js b/static/app.js new file mode 100644 index 0000000..886d9e0 --- /dev/null +++ b/static/app.js @@ -0,0 +1,275 @@ +/* The panel. Every control acts immediately and reconciles with what the + speakers report a moment later, because a remote that waits for a + network round trip before it looks like it did anything feels broken. */ + +const STEP = Number(document.documentElement.dataset.step) || 2; +const POLL_MS = 5000; + +const el = (sel, root = document) => root.querySelector(sel); +const els = (sel, root = document) => Array.from(root.querySelectorAll(sel)); + +const ui = { + foot: el('[data-role="foot"]'), + toast: el('[data-role="toast"]'), + refresh: el('[data-role="refresh"]'), + splitAll: el('[data-role="split-all"]'), + avrStatus: el('[data-role="avr-status"]'), + inputButton: el('[data-role="input-button"]'), + inputName: el('[data-role="input-name"]'), + sheet: el('[data-role="sheet"]'), + options: el('[data-role="options"]'), +}; + +const rooms = {}; +let inputs = []; +let currentInput = null; + +/* --- transport -------------------------------------------------------- */ +async function api(path, body) { + const options = body === undefined + ? {} + : { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) }; + const response = await fetch(path, options); + let data = {}; + try { data = await response.json(); } catch (_) { /* empty or not JSON */ } + if (!response.ok) throw new Error(data.error || `${response.status} ${response.statusText}`); + return data; +} + +let toastTimer; +function toast(message) { + ui.toast.textContent = message; + ui.toast.hidden = false; + clearTimeout(toastTimer); + toastTimer = setTimeout(() => { ui.toast.hidden = true; }, 4000); +} + +/* --- rooms ------------------------------------------------------------ */ +els('.room').forEach((node) => { + const key = node.dataset.room; + const room = { + key, + node, + level: el('[data-role="level"]', node), + bar: el('[data-role="bar"]', node), + toggle: el('[data-role="group"]', node), + toggleLabel: el('[data-role="group-label"]', node), + steps: els('.step', node), + volume: null, + grouped: false, + available: false, + pending: 0, // taps not yet sent + inflight: false, + busy: false, // a grouping change is in flight + }; + rooms[key] = room; + + room.steps.forEach((button) => { + const direction = Number(button.dataset.delta); + holdable(button, () => nudge(key, direction)); + }); + + room.toggle.addEventListener('click', () => setGrouped(key, !room.grouped)); +}); + +function paintRoom(room) { + const known = room.volume !== null && room.volume !== undefined; + room.level.textContent = known ? room.volume : '—'; + room.bar.style.width = `${known ? room.volume : 0}%`; + room.node.classList.toggle('offline', !room.available); + room.steps.forEach((button) => { button.disabled = !room.available; }); + room.toggle.disabled = !room.available; + room.toggle.classList.toggle('busy', room.busy); + room.toggle.setAttribute('aria-pressed', String(room.grouped)); + room.toggleLabel.textContent = room.grouped ? 'Grouped with AVR' : 'Separate'; +} + +/* Press and hold to keep moving, accelerating as you hold. */ +function holdable(node, action) { + let timer = null; + let delay = 420; + + const stop = () => { clearTimeout(timer); timer = null; delay = 420; }; + const tick = () => { + action(); + delay = Math.max(130, delay * 0.72); + timer = setTimeout(tick, delay); + }; + + node.addEventListener('pointerdown', (event) => { + if (event.button > 0 || node.disabled) return; + event.preventDefault(); // also suppresses the click that would double-fire + action(); + timer = setTimeout(tick, delay); + }); + ['pointerup', 'pointercancel', 'pointerleave'].forEach((type) => node.addEventListener(type, stop)); +} + +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; + paintRoom(room); + 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; + room.inflight = true; + try { + const data = await api('/api/volume', { target: key, delta }); + if (!room.pending) { + room.volume = data.level; + paintRoom(room); + } + } catch (error) { + toast(error.message); + refresh(); + } finally { + room.inflight = false; + if (room.pending) flushVolume(key); + } +} + +async function setGrouped(key, joined) { + const room = rooms[key]; + if (room.busy || !room.available) return; + room.busy = true; + room.grouped = joined; // show the new state while the speakers catch up + paintRoom(room); + try { + const data = await api('/api/group', { target: key, joined }); + applyJoined(data.joined || []); + } catch (error) { + toast(error.message); + } finally { + room.busy = false; + refresh(); + } +} + +function applyJoined(joined) { + Object.values(rooms).forEach((room) => { + room.grouped = joined.includes(room.key); + paintRoom(room); + }); +} + +ui.splitAll.addEventListener('click', async () => { + try { + applyJoined([]); + const data = await api('/api/group/none'); + applyJoined(data.joined || []); + } catch (error) { + toast(error.message); + } finally { + refresh(); + } +}); + +/* --- the input picker -------------------------------------------------- */ +ui.inputButton.addEventListener('click', openSheet); +el('[data-role="scrim"]').addEventListener('click', closeSheet); +el('[data-role="sheet-close"]').addEventListener('click', closeSheet); + +function sheetOpen() { return !ui.sheet.hidden; } + +function openSheet() { + if (!inputs.length) { + toast('No inputs reported by the AVR yet'); + return; + } + ui.options.innerHTML = ''; + inputs.forEach((source) => { + const option = document.createElement('button'); + option.className = 'option'; + option.setAttribute('aria-current', String(currentInput && currentInput.code === source.code)); + option.innerHTML = ''; + option.firstChild.textContent = source.name; + option.lastChild.textContent = source.code; + option.addEventListener('click', () => chooseInput(source)); + ui.options.appendChild(option); + }); + ui.sheet.hidden = false; +} + +function closeSheet() { ui.sheet.hidden = true; } + +async function chooseInput(source) { + closeSheet(); + currentInput = source; + ui.inputName.textContent = source.name; + try { + const data = await api('/api/avr/input', { code: source.code }); + currentInput = data; + ui.inputName.textContent = data.name; + } catch (error) { + toast(error.message); + refresh(); + } +} + +/* --- state ------------------------------------------------------------- */ +function render(state) { + (state.rooms || []).forEach((incoming) => { + const room = rooms[incoming.key]; + if (!room) return; + 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; + paintRoom(room); + }); + + const avr = state.avr || {}; + inputs = avr.inputs || []; + currentInput = avr.input || null; + ui.inputName.textContent = currentInput ? currentInput.name : '—'; + ui.avrStatus.textContent = avr.connected ? 'ready' : 'offline'; + ui.avrStatus.classList.toggle('on', Boolean(avr.connected)); + + const problems = state.errors || []; + ui.foot.textContent = problems.length ? problems[0] : (state.demo ? 'demo mode — no real speakers' : ''); + ui.foot.classList.toggle('bad', problems.length > 0); +} + +let refreshing = false; +async function refresh() { + if (refreshing) return; + refreshing = true; + try { + render(await api('/api/state')); + } catch (error) { + ui.foot.textContent = error.message; + ui.foot.classList.add('bad'); + } finally { + refreshing = false; + } +} + +function busy() { + return sheetOpen() || Object.values(rooms).some((r) => r.pending || r.inflight || r.busy); +} + +ui.refresh.addEventListener('click', () => { + ui.refresh.classList.add('spin'); + setTimeout(() => ui.refresh.classList.remove('spin'), 700); + refresh(); +}); + +setInterval(() => { + if (document.visibilityState === 'visible' && !busy()) refresh(); +}, POLL_MS); + +document.addEventListener('visibilitychange', () => { + if (document.visibilityState === 'visible') refresh(); +}); + +Object.values(rooms).forEach(paintRoom); +refresh(); diff --git a/static/icon-180.png b/static/icon-180.png new file mode 100644 index 0000000..b374d6d Binary files /dev/null and b/static/icon-180.png differ diff --git a/static/icon-512.png b/static/icon-512.png new file mode 100644 index 0000000..c3ee4a2 Binary files /dev/null and b/static/icon-512.png differ diff --git a/static/panel.css b/static/panel.css new file mode 100644 index 0000000..2c13d75 --- /dev/null +++ b/static/panel.css @@ -0,0 +1,195 @@ +/* A remote lives in the dark next to a TV, so the panel is dark too -- + and every control is sized for a thumb, not a cursor. */ + +:root { + color-scheme: dark; + --bg: #0a0d14; + --card: #151a25; + --edge: #232b3b; + --raised: #1e2634; + --ink: #eef2f9; + --muted: #8d98ad; + --accent: #5b8def; + --live: #3ddc97; + --warn: #ff7a6b; + --radius: 22px; +} + +* { box-sizing: border-box; } + +html { background: var(--bg); } + +body { + margin: 0; + background: var(--bg); + color: var(--ink); + font: 16px/1.4 -apple-system, BlinkMacSystemFont, "SF Pro Text", "Segoe UI", system-ui, sans-serif; + -webkit-font-smoothing: antialiased; + -webkit-tap-highlight-color: transparent; + -webkit-touch-callout: none; + user-select: none; + overscroll-behavior-y: contain; +} + +button { + font: inherit; + color: inherit; + border: 0; + background: none; + cursor: pointer; + touch-action: manipulation; +} + +svg { width: 22px; height: 22px; fill: none; stroke: currentColor; stroke-width: 2; stroke-linecap: round; stroke-linejoin: round; } + +.app { + max-width: 520px; + margin: 0 auto; + padding: max(12px, env(safe-area-inset-top)) max(16px, env(safe-area-inset-right)) + max(24px, env(safe-area-inset-bottom)) max(16px, env(safe-area-inset-left)); + display: flex; + flex-direction: column; + gap: 14px; +} + +/* --- header ---------------------------------------------------------- */ +.top { display: flex; align-items: center; justify-content: space-between; padding: 6px 4px 0; } +.top h1 { margin: 0; font-size: 22px; font-weight: 650; letter-spacing: .02em; } + +.icon-button { + width: 44px; height: 44px; border-radius: 50%; + display: grid; place-items: center; + color: var(--muted); background: var(--card); +} +.icon-button:active { background: var(--raised); color: var(--ink); } +.icon-button.spin svg { animation: spin .7s linear; } +@keyframes spin { to { transform: rotate(360deg); } } + +/* --- cards ----------------------------------------------------------- */ +.card { + background: var(--card); + border: 1px solid var(--edge); + border-radius: var(--radius); + padding: 16px; + display: flex; + flex-direction: column; + gap: 14px; +} +.card.offline { opacity: .5; } + +.card-head { display: flex; align-items: center; justify-content: space-between; gap: 12px; } +.card-head h2 { margin: 0; font-size: 17px; font-weight: 600; } + +.level { color: var(--muted); font-size: 13px; } +.level b { + color: var(--ink); + font-size: 26px; + font-weight: 640; + font-variant-numeric: tabular-nums; +} + +.pill { + font-size: 11px; text-transform: uppercase; letter-spacing: .08em; + color: var(--muted); background: var(--raised); + padding: 4px 9px; border-radius: 999px; +} +.pill.on { color: var(--live); } + +/* --- volume ---------------------------------------------------------- */ +.volume { display: flex; align-items: center; gap: 14px; } + +.step { + flex: 0 0 auto; + width: 78px; height: 62px; + border-radius: 18px; + background: var(--raised); + font-size: 30px; font-weight: 500; line-height: 1; + display: grid; place-items: center; +} +.step:active { background: var(--accent); transform: scale(.96); } +.step:disabled { opacity: .4; } + +.meter { flex: 1; height: 8px; border-radius: 999px; background: #0c111b; overflow: hidden; } +.meter i { display: block; height: 100%; width: 0; border-radius: 999px; background: var(--accent); transition: width .12s ease-out; } + +/* --- join / leave the AVR -------------------------------------------- */ +.toggle { + height: 52px; border-radius: 16px; + background: var(--raised); + display: flex; align-items: center; justify-content: center; gap: 10px; + font-size: 15px; font-weight: 550; + color: var(--muted); +} +.toggle .dot { width: 9px; height: 9px; border-radius: 50%; background: currentColor; opacity: .6; } +.toggle[aria-pressed="true"] { background: var(--accent); color: #fff; } +.toggle[aria-pressed="true"] .dot { background: #fff; opacity: 1; } +.toggle:active { transform: scale(.985); } +.toggle:disabled { opacity: .5; } +.toggle.busy { opacity: .6; } + +/* --- source card ------------------------------------------------------ */ +.source-button { + display: flex; align-items: center; gap: 12px; + width: 100%; min-height: 64px; + padding: 10px 14px; + border-radius: 16px; + background: var(--raised); + text-align: left; +} +.source-button:active { transform: scale(.985); } +.source-button .eyebrow { font-size: 11px; text-transform: uppercase; letter-spacing: .08em; color: var(--muted); } +.source-button .value { flex: 1; font-size: 19px; font-weight: 600; } +.source-button .chevron { color: var(--muted); } + +/* --- misc ------------------------------------------------------------- */ +.wide { width: 100%; height: 50px; border-radius: 16px; font-size: 15px; } +.ghost { background: transparent; border: 1px solid var(--edge); color: var(--muted); } +.ghost:active { background: var(--card); color: var(--ink); } + +.foot { margin: 2px 4px 0; min-height: 18px; font-size: 12px; color: var(--muted); text-align: center; } +.foot.bad { color: var(--warn); } + +/* --- input sheet ------------------------------------------------------ */ +.sheet-wrap { position: fixed; inset: 0; z-index: 10; display: flex; flex-direction: column; justify-content: flex-end; } +.scrim { position: absolute; inset: 0; background: rgba(4, 6, 11, .6); backdrop-filter: blur(3px); } + +.sheet { + position: relative; + background: var(--card); + border: 1px solid var(--edge); + border-radius: 28px 28px 0 0; + padding: 10px 16px calc(16px + env(safe-area-inset-bottom)); + max-height: 82vh; + display: flex; flex-direction: column; gap: 12px; + animation: rise .22s cubic-bezier(.22, .68, .3, 1); +} +@keyframes rise { from { transform: translateY(14%); opacity: .4; } } + +.grabber { width: 38px; height: 4px; border-radius: 999px; background: var(--edge); margin: 2px auto 4px; } +.sheet h3 { margin: 0 4px; font-size: 14px; font-weight: 600; color: var(--muted); } + +.options { display: flex; flex-direction: column; gap: 8px; overflow-y: auto; -webkit-overflow-scrolling: touch; } +.option { + display: flex; align-items: center; justify-content: space-between; gap: 10px; + min-height: 58px; padding: 0 16px; + border-radius: 16px; background: var(--raised); + font-size: 17px; text-align: left; +} +.option:active { transform: scale(.985); } +.option[aria-current="true"] { background: var(--accent); color: #fff; } +.option .code { font-size: 12px; color: var(--muted); } +.option[aria-current="true"] .code { color: rgba(255, 255, 255, .8); } + +/* --- toast ------------------------------------------------------------ */ +.toast { + position: fixed; left: 50%; transform: translateX(-50%); + bottom: calc(22px + env(safe-area-inset-bottom)); + z-index: 20; max-width: 90vw; + padding: 12px 16px; border-radius: 14px; + background: #2b1f24; border: 1px solid #52303a; color: #ffd9d4; + font-size: 14px; box-shadow: 0 10px 30px rgba(0, 0, 0, .45); +} + +@media (prefers-reduced-motion: reduce) { + * { animation: none !important; transition: none !important; } +} diff --git a/templates/index.html b/templates/index.html new file mode 100644 index 0000000..210ac8b --- /dev/null +++ b/templates/index.html @@ -0,0 +1,80 @@ + + + + + +{{ app_name }} + + + + + + + + + + + + + + + +
+
+

{{ app_name }}

+ +
+ +
+
+

{{ host.label }}

+ offline +
+ +
+ + {% for room in rooms %} +
+
+

{{ room.label }}

+ — +
+ +
+ +
+ +
+ + +
+ {% endfor %} + + +

+
+ + + + + + + + diff --git a/templates/manifest.webmanifest b/templates/manifest.webmanifest new file mode 100644 index 0000000..2050a53 --- /dev/null +++ b/templates/manifest.webmanifest @@ -0,0 +1,12 @@ +{ + "name": "{{ app_name }}", + "short_name": "{{ app_name }}", + "start_url": "/", + "display": "standalone", + "background_color": "#0a0d14", + "theme_color": "#0a0d14", + "icons": [ + { "src": "/static/icon-180.png", "sizes": "180x180", "type": "image/png" }, + { "src": "/static/icon-512.png", "sizes": "512x512", "type": "image/png", "purpose": "any maskable" } + ] +} diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..b33aea6 --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,170 @@ +"""Stand-in Denon hardware: just enough HEOS and Telnet to test against. + +The grouping rules are the part worth pinning down -- what a set_group +call does to a stereo pair is the kind of thing you do not want to find +out by experimenting on the speakers at eleven at night. +""" + +import json +import socket +import threading + + +class FakeHeos(threading.Thread): + """A HEOS CLI server on localhost, with four players and a pair.""" + + NAMES = {1: "Home Cinema", 2: "Lego Room", 3: "Denon Home 200 L", 4: "Denon Home 200 R"} + + def __init__(self): + super().__init__(daemon=True) + self.groups = {3: [3, 4]} # gid -> pids, leader first + self.volumes = {pid: 20 for pid in self.NAMES} + self.group_volumes = {3: 25} + self.commands = [] # everything we were asked to do + self.server = socket.socket() + self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server.bind(("127.0.0.1", 0)) + self.server.listen(8) + self.port = self.server.getsockname()[1] + self.start() + + def run(self): + while True: + try: + conn, _ = self.server.accept() + except OSError: + return + threading.Thread(target=self._serve, args=(conn,), daemon=True).start() + + def _serve(self, conn): + buffer = b"" + with conn: + while True: + try: + chunk = conn.recv(4096) + except OSError: + return + if not chunk: + return + buffer += chunk + while b"\r\n" in buffer: + line, buffer = buffer.split(b"\r\n", 1) + reply = self.handle(line.decode().strip()) + conn.sendall(json.dumps(reply).encode() + b"\r\n") + + # -- command handling ------------------------------------------------ + def handle(self, command: str) -> dict: + self.commands.append(command) + path, _, query = command[len("heos://"):].partition("?") + args = dict(part.split("=", 1) for part in query.split("&") if "=" in part) + + if path == "player/get_players": + return self._ok(path, payload=[ + {"name": name, "pid": pid, "model": "Fake"} for pid, name in self.NAMES.items() + ]) + + if path == "group/get_groups": + payload = [] + for gid, pids in self.groups.items(): + payload.append({ + "name": self.NAMES[gid], "gid": gid, + "players": [ + {"name": self.NAMES[pid], "pid": pid, + "role": "leader" if pid == gid else "member"} + for pid in pids + ], + }) + return self._ok(path, payload=payload) + + if path == "group/set_group": + pids = [int(p) for p in args["pid"].split(",")] + for gid in list(self.groups): + self.groups[gid] = [p for p in self.groups[gid] if p not in pids] + if len(self.groups[gid]) < 2: + del self.groups[gid] # HEOS dissolves a group of one + if len(pids) > 1: + self.groups[pids[0]] = pids + self.group_volumes.setdefault(pids[0], 25) + return self._ok(path, message=args["pid"]) + + if path.endswith("/get_volume"): + store, key = self._store(path, args) + return self._ok(path, message=f"{key[0]}={key[1]}&level={store[key[1]]}") + + if path.endswith("/set_volume"): + store, key = self._store(path, args) + store[key[1]] = int(args["level"]) + return self._ok(path, message=f"{key[0]}={key[1]}&level={args['level']}") + + if path.endswith("/toggle_mute") or path == "system/heart_beat": + return self._ok(path) + + return {"heos": {"command": path, "result": "fail", "message": "eid=2&text=Not+supported"}} + + def _store(self, path, args): + if path.startswith("group/"): + gid = int(args["gid"]) + if gid not in self.groups: + raise KeyError(gid) + return self.group_volumes, ("gid", gid) + return self.volumes, ("pid", int(args["pid"])) + + @staticmethod + def _ok(path, message="", payload=None): + reply = {"heos": {"command": path, "result": "success", "message": message}} + if payload is not None: + reply["payload"] = payload + return reply + + +class FakeAvr(threading.Thread): + """A Denon Telnet server that knows SI and SSFUN.""" + + SOURCES = [("MPLAY", "Apple TV"), ("GAME", "PlayStation"), ("SAT/CBL", "TV Box")] + + def __init__(self): + super().__init__(daemon=True) + self.input = "MPLAY" + self.server = socket.socket() + self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server.bind(("127.0.0.1", 0)) + self.server.listen(4) + self.port = self.server.getsockname()[1] + self.start() + + def run(self): + while True: + try: + conn, _ = self.server.accept() + except OSError: + return + threading.Thread(target=self._serve, args=(conn,), daemon=True).start() + + def _serve(self, conn): + buffer = b"" + with conn: + while True: + try: + chunk = conn.recv(1024) + except OSError: + return + if not chunk: + return + buffer += chunk + while b"\r" in buffer: + line, buffer = buffer.split(b"\r", 1) + for reply in self.handle(line.decode().strip()): + conn.sendall(reply.encode() + b"\r") + + def handle(self, command: str) -> list: + 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 == "SI?": + return [f"SI{self.input}"] + if command.startswith("SI"): + self.input = command[2:] + return [f"SI{self.input}"] + if command == "PW?": + return ["PWON"] + return [] diff --git a/tests/test_panel.py b/tests/test_panel.py new file mode 100644 index 0000000..883020c --- /dev/null +++ b/tests/test_panel.py @@ -0,0 +1,166 @@ +"""Run against the fake hardware in fakes.py: + + python3 -m unittest discover -s tests -t . +""" + +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from controller import Controller # noqa: E402 +from tests.fakes import FakeAvr, FakeHeos # noqa: E402 + +PAIR = {3, 4} # the two Home 200s +AVR_PID = 1 +HOME400_PID = 2 + + +def build(tmpdir): + heos, avr = FakeHeos(), FakeAvr() + cfg = SimpleNamespace( + HEOS_HOST="127.0.0.1", HEOS_PORT=heos.port, + AVR_HOST="127.0.0.1", AVR_PORT=avr.port, + HOST_KEY="avr", + ROOM_KEYS=["home400", "living_room_group"], + TARGETS={ + "avr": {"label": "Home Cinema", "heos_name": "Home Cinema"}, + "home400": {"label": "Lego Room", "heos_name": "Lego Room"}, + "living_room_group": {"label": "Living Room", "heos_name": "Denon Home 200 L"}, + }, + AVR_INPUT_CODES=[], + MEMBERS_FILE=str(Path(tmpdir) / "members.json"), + ) + return Controller(cfg), heos, avr + + +class PanelTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.panel, self.heos, self.avr = build(self.tmp.name) + + def group_pids(self): + return {gid: set(pids) for gid, pids in self.heos.groups.items()} + + # -- resolving ------------------------------------------------------ + def test_pair_resolves_to_both_speakers(self): + """The bug this replaces: grouping used only the pair's leader, + which left the second Home 200 behind.""" + self.panel.scan() + self.assertEqual(set(self.panel.member_pids("living_room_group")), PAIR) + self.assertEqual(self.panel.member_pids("home400"), [HOME400_PID]) + + def test_avr_resolves_to_a_player_even_while_it_leads_a_group(self): + self.panel.join("home400") + # HEOS now reports a *group* named "Home Cinema" as well as the player. + self.assertEqual(self.panel.member_pids("avr"), [AVR_PID]) + + # -- grouping ------------------------------------------------------- + def test_joining_takes_the_whole_pair(self): + self.panel.join("living_room_group") + self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR}) + self.assertEqual(self.panel.joined_keys(), ["living_room_group"]) + + def test_joining_keeps_whoever_is_already_grouped(self): + self.panel.join("home400") + self.panel.join("living_room_group") + self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, HOME400_PID} | PAIR}) + self.assertEqual(self.panel.joined_keys(), ["home400", "living_room_group"]) + + def test_leaving_rebuilds_the_stereo_pair(self): + self.panel.join("living_room_group") + self.panel.leave("living_room_group") + self.assertEqual(self.group_pids(), {3: PAIR}) # pair back, AVR alone + self.assertEqual(self.panel.joined_keys(), []) + + def test_leaving_one_room_does_not_disturb_the_other(self): + self.panel.join("home400") + self.panel.join("living_room_group") + before = [c for c in self.heos.commands if "set_group" in c] + self.panel.leave("home400") + after = [c for c in self.heos.commands if "set_group" in c] + self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR}) + # Exactly one new set_group: the remaining room is never regrouped, + # which is what stops its music restarting. + self.assertEqual(len(after) - len(before), 1) + + def test_separate_everything(self): + self.panel.set_membership(["home400", "living_room_group"]) + self.panel.set_membership([]) + self.assertEqual(self.group_pids(), {3: PAIR}) + self.assertEqual(self.panel.joined_keys(), []) + + def test_membership_survives_a_restart_while_merged(self): + """Once merged, the pair's own group is gone from HEOS, so a fresh + process has to fall back on what it learned earlier.""" + self.panel.join("living_room_group") + reborn = Controller(self.panel.cfg) + reborn.scan() + self.assertEqual(set(reborn.member_pids("living_room_group")), PAIR) + reborn.leave("living_room_group") + self.assertEqual(self.group_pids(), {3: PAIR}) + + # -- volume --------------------------------------------------------- + def test_pair_uses_group_volume_when_it_stands_alone(self): + self.panel.scan() + self.assertEqual(self.panel.set_volume("living_room_group", 42), 42) + self.assertEqual(self.heos.group_volumes[3], 42) + + def test_pair_uses_player_volume_once_merged(self): + """Its gid stops existing the moment it joins the AVR, so the old + bridge's get_volume?gid= call would simply fail here.""" + self.panel.join("living_room_group") + self.assertEqual(self.panel.set_volume("living_room_group", 31), 31) + self.assertEqual(self.heos.volumes[3], 31) + self.assertEqual(self.heos.volumes[4], 31) + self.assertEqual(self.panel.volume("living_room_group"), 31) + + def test_nudge_clamps_at_the_ends(self): + self.panel.set_volume("home400", 98) + self.assertEqual(self.panel.nudge_volume("home400", 5), 100) + self.panel.set_volume("home400", 1) + self.assertEqual(self.panel.nudge_volume("home400", -9), 0) + + # -- the AVR --------------------------------------------------------- + def test_renamed_inputs_and_selection(self): + deadline = time.time() + 5 + while not self.panel.avr.connected and time.time() < deadline: + time.sleep(0.05) + self.assertTrue(self.panel.avr.connected) + + self.assertEqual( + self.panel.avr.inputs(), + [{"code": "MPLAY", "name": "Apple TV"}, + {"code": "GAME", "name": "PlayStation"}, + {"code": "SAT/CBL", "name": "TV Box"}], + ) + self.assertEqual(self.panel.avr.current_input(), {"code": "MPLAY", "name": "Apple TV"}) + self.assertEqual(self.panel.avr.select_input("GAME"), {"code": "GAME", "name": "PlayStation"}) + self.assertEqual(self.avr.input, "GAME") + + # -- the whole snapshot the UI renders ------------------------------- + def test_state_snapshot(self): + self.panel.join("home400") + state = self.panel.state() + self.assertTrue(state["heos_ok"]) + self.assertEqual([r["key"] for r in state["rooms"]], ["home400", "living_room_group"]) + self.assertEqual([r["grouped"] for r in state["rooms"]], [True, False]) + self.assertTrue(all(isinstance(r["volume"], int) for r in state["rooms"])) + + def test_state_reports_trouble_instead_of_blowing_up(self): + self.panel.heos.host = "127.0.0.1" + self.panel.heos.port = 1 # nothing is listening there + self.panel.heos._close() + state = self.panel.state() + self.assertFalse(state["heos_ok"]) + self.assertTrue(state["errors"]) + self.assertTrue(all(r["available"] is False for r in state["rooms"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/make_icons.py b/tools/make_icons.py new file mode 100644 index 0000000..0d40a8a --- /dev/null +++ b/tools/make_icons.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +"""Draw the home-screen icon, with no image library to install. + + python3 tools/make_icons.py + +Writes static/icon-180.png (what iOS uses for the home screen) and +static/icon-512.png (Android / the web manifest). Re-run it if you want +different colours -- they are the two constants below. +""" + +import math +import struct +import zlib +from pathlib import Path + +TOP = (0x1D, 0x27, 0x3C) # background gradient, top +BOTTOM = (0x0A, 0x0D, 0x14) # background gradient, bottom +GLYPH = (0xEE, 0xF2, 0xF9) # the wave itself + +STATIC = Path(__file__).resolve().parent.parent / "static" +SAMPLES = 3 # supersampling per axis, for smooth edges + + +def coverage(x: float, y: float, size: float) -> float: + """How much of the glyph covers this point: a dot plus three arcs + opening to the right, i.e. the usual 'sound coming out' mark.""" + cx, cy = 0.33 * size, 0.5 * size + dx, dy = x - cx, y - cy + distance = math.hypot(dx, dy) + + if distance <= 0.075 * size: + return 1.0 + + angle = abs(math.degrees(math.atan2(dy, dx))) + if angle > 50: + return 0.0 + + half = 0.024 * size + for radius in (0.17, 0.27, 0.37): + if abs(distance - radius * size) <= half: + return 1.0 + return 0.0 + + +def render(size: int) -> bytes: + rows = bytearray() + step = 1.0 / SAMPLES + for py in range(size): + rows.append(0) # PNG filter type 0 for this scanline + mix = py / max(1, size - 1) + background = tuple( + round(TOP[i] + (BOTTOM[i] - TOP[i]) * mix) for i in range(3) + ) + for px in range(size): + hits = 0 + for sy in range(SAMPLES): + for sx in range(SAMPLES): + if coverage(px + (sx + 0.5) * step, py + (sy + 0.5) * step, size): + hits += 1 + alpha = hits / (SAMPLES * SAMPLES) + if alpha == 0: + rows.extend(background) + else: + rows.extend( + round(background[i] + (GLYPH[i] - background[i]) * alpha) for i in range(3) + ) + return bytes(rows) + + +def write_png(path: Path, size: int, raw: bytes): + def chunk(kind: bytes, data: bytes) -> bytes: + return (struct.pack(">I", len(data)) + kind + data + + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)) + + header = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0) # 8-bit truecolour + path.write_bytes( + b"\x89PNG\r\n\x1a\n" + + chunk(b"IHDR", header) + + chunk(b"IDAT", zlib.compress(raw, 9)) + + chunk(b"IEND", b"") + ) + + +if __name__ == "__main__": + for size in (180, 512): + target = STATIC / f"icon-{size}.png" + write_png(target, size, render(size)) + print(f"wrote {target} ({target.stat().st_size} bytes)")