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

Merge the bridge and a home-screen interface into one app

Browse Source 
The HEOS app is unpleasant to use for the four things that actually get
done in this house, so this is those four things on one screen: volume
for the Home 400 and the Living Room pair, joining either to the AVR,
splitting them off again, and picking the AVR's input.

The bridge's logic moves in mostly intact, split into a HEOS client, an
AVR client and a controller, with two grouping bugs fixed on the way:

  * Merging a room into the AVR sent only the pair's leader pid, which
    left the second Home 200 behind. Group targets now expand to every
    member pid, and unmerging re-forms the pair rather than leaving two
    lone speakers. The members are learned while the pair is visible and
    remembered in members.json so a restart can still rebuild it.

  * Volume for the pair used its gid, which stops existing the moment it
    joins the AVR's group. It now falls back to the member players.

Both sockets are kept open rather than reconnecting per command, which
is what made every button press feel slow; the AVR connection doubles as
a listener, so input changes made with the physical remote show up too.

The original query-string endpoints still answer, so anything already
pointed at the bridge keeps working.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
franzz
2026-09-14 21:15:43 +02:00
co-authored by Claude Opus 5
parent 86fdf9a4d3
commit 375bcfdd76
20 changed files with 2428 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files
avr.py
+219
View File
@@ -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)}