From 86fdf9a4d3d8ca7f03d1fb4d9662c98b5261ab24 Mon Sep 17 00:00:00 2001 From: Franzz Date: Mon, 14 Sep 2026 21:14:21 +0200 Subject: [PATCH] Add the original heos_bridge.py before merging it into the panel Recording the bridge as it stood so the rewrite that follows has something to be a diff against, rather than appearing from nowhere. --- heos_bridge.py | 488 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 488 insertions(+) create mode 100644 heos_bridge.py diff --git a/heos_bridge.py b/heos_bridge.py new file mode 100644 index 0000000..4373534 --- /dev/null +++ b/heos_bridge.py @@ -0,0 +1,488 @@ +#!/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