franzz/heos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add the original heos_bridge.py before merging it into the panel

Browse Source 
Recording the bridge as it stood so the rewrite that follows has
something to be a diff against, rather than appearing from nowhere.
This commit is contained in:
franzz
2026-09-14 21:14:21 +02:00
parent d9cadb7e64
commit 86fdf9a4d3
1 changed files with 488 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
heos_bridge.py
+488
View File
@@ -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://<pi-ip>:5005/targets
Once TARGETS is filled in:
GET http://<pi-ip>:5005/volume?target=living_room_group
POST http://<pi-ip>:5005/volume/set?target=kitchen&level=25
POST http://<pi-ip>:5005/volume/up?target=avr&step=5
POST http://<pi-ip>:5005/volume/down?target=living_room_group&step=5
POST http://<pi-ip>:5005/volume/mute?target=kitchen
POST http://<pi-ip>:5005/playback/play?target=avr
POST http://<pi-ip>:5005/playback/pause?target=avr
POST http://<pi-ip>:5005/playback/stop?target=avr
POST http://<pi-ip>:5005/playback/next?target=avr
POST http://<pi-ip>:5005/playback/previous?target=avr
GET http://<pi-ip>:5005/inputs?target=avr
POST http://<pi-ip>:5005/input/set?target=avr&input=inputs/hdmi_in_1
POST http://<pi-ip>:5005/input/relay?target=home400&from=avr&input=inputs/tv
POST http://<pi-ip>:5005/group/create?host=avr&members=home400,living_room_group
POST http://<pi-ip>: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://<pi-ip>:5005/avr/raw?cmd=SSFUN ? (explore renamed sources)
GET http://<pi-ip>:5005/avr/input (current input, via SI?)
POST http://<pi-ip>: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=<target>&members=<comma-separated targets>'"}), 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/<path:subpath>", 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=<source target>&input=<input id>'"}), 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=<raw telnet command>'"}), 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=<code>', 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)