heos/controller.py

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


def stepped_level(current: int, steps: int, size: int) -> int:
    """Where the volume lands after `steps` taps of +/-.

    A tap moves to the next multiple of `size` rather than adding `size`,
    so a level of 23 with a step of 5 goes 23 -> 25 -> 30 on the way up
    and 23 -> 20 -> 15 on the way down. Levels that are already on a
    multiple simply move a whole step.
    """
    if steps > 0:
        landing = (current // size + steps) * size
    elif steps < 0:
        aligned = current // size * size
        first = current - size if aligned == current else aligned
        landing = first - (-steps - 1) * size
    else:
        return current
    return max(0, min(100, landing))


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 step_volume(self, key: str, steps: int) -> int:
        """Move by whole taps, landing on multiples of VOLUME_STEP.

        Counted in taps rather than points so the speakers' own level is
        what gets rounded, even when the phone's copy of it is a few
        seconds stale.
        """
        with self._lock:
            self._fresh()
            handles = self._volume_handles(key)
            level = stepped_level(
                self._read_volume(*handles[0]), int(steps), self.cfg.VOLUME_STEP
            )
            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