heos/avr.py

"""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


def parse_sssod(lines: list) -> dict:
    """Parse `SSSOD ?` output -- 'SSSODTUNER DEL' and friends -- into
    {"TUNER": False, "CD": True, ...}, i.e. which sources you have left
    switched on in the AVR's own setup menu."""
    usage = {}
    for line in lines:
        if not line.startswith("SSSOD"):
            continue
        rest = line[len("SSSOD"):].strip()
        if rest in ("END", ""):
            continue
        code, _, value = rest.rpartition(" ")
        code = code.strip()
        if code:
            usage[code] = value.strip().upper() != "DEL"
    return usage


class AvrControl:
    """The input list and the current input, in the names you chose."""

    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
        self._usage = None

    @property
    def connected(self) -> bool:
        return self.telnet.connected

    def all_inputs(self, refresh: bool = False) -> list:
        """Every source the AVR knows, under your names, deleted ones
        included. Cached: it only changes when you edit the setup menu."""
        if self._inputs is None or refresh:
            lines = self.telnet.request(
                "SSFUN ?", prefix="SSFUN",
                until=lambda line: line.strip() == "SSFUN END",
                timeout=3.0,
            )
            sources = parse_ssfun(lines)
            if sources:
                self._inputs = sources
        if self._usage is None or refresh:
            lines = self.telnet.request(
                "SSSOD ?", prefix="SSSOD",
                until=lambda line: line.strip() == "SSSOD END",
                timeout=3.0,
            )
            self._usage = parse_sssod(lines)
        return self._inputs or []

    def inputs(self, refresh: bool = False) -> list:
        """What the picker offers: the sources you can actually select.

        Sources you deleted in the AVR's setup menu are left out -- they
        are exactly the ones you never want to land on. Anything SSSOD
        does not mention is kept, so a model that does not answer that
        command shows its whole list rather than nothing at all.
        """
        sources = [s for s in self.all_inputs(refresh) if self._usage.get(s["code"], True)]
        if self.allowed_codes:
            order = {code: i for i, code in enumerate(self.allowed_codes)}
            sources = sorted(
                (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.all_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)}