heos/demo.py

"""A pretend HEOS network, for `python3 app.py --demo`.

Lets you work on the interface on a laptop, with no speakers on the
network -- and lets the panel be tested without waking the house up.
"""

import time

from controller import stepped_level


class _FakeAvr:
    INPUTS = [
        {"code": "MPLAY", "name": "Apple TV"},
        {"code": "GAME", "name": "PlayStation"},
        {"code": "SAT/CBL", "name": "TV Box"},
        {"code": "BD", "name": "Blu-ray"},
        {"code": "TUNER", "name": "Radio"},
        {"code": "PHONO", "name": "Turntable"},
    ]

    def __init__(self, allowed_codes=()):
        self.allowed_codes = list(allowed_codes or [])
        self.connected = True
        self._code = "MPLAY"

    def inputs(self, refresh=False):
        sources = list(self.INPUTS)
        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 name_for(self, code):
        return next((s["name"] for s in self.INPUTS if s["code"] == code), code)

    def current_input(self):
        return {"code": self._code, "name": self.name_for(self._code)}

    def select_input(self, code):
        time.sleep(0.15)  # the real AVR is not instant either
        self._code = code
        return self.current_input()


class DemoController:
    def __init__(self, cfg):
        self.cfg = cfg
        self.avr = _FakeAvr(cfg.AVR_INPUT_CODES)
        self.heos = None
        self._volume = {key: 22 + 7 * i for i, key in enumerate(cfg.TARGETS)}
        self._joined = set()

    # -- what the UI uses ---------------------------------------------
    def state(self):
        return {
            "host": {"key": self.cfg.HOST_KEY, "label": self.cfg.TARGETS[self.cfg.HOST_KEY]["label"]},
            "rooms": [
                {"key": key, "label": self.cfg.TARGETS[key]["label"], "available": True,
                 "grouped": key in self._joined, "volume": self._volume[key], "error": None}
                for key in self.cfg.ROOM_KEYS
            ],
            "avr": {"connected": True, "inputs": self.avr.inputs(), "input": self.avr.current_input()},
            "heos_ok": True,
            "errors": [],
            "demo": True,
        }

    def volume(self, key):
        return self._volume[key]

    def set_volume(self, key, level):
        self._volume[key] = max(0, min(100, int(level)))
        return self._volume[key]

    def nudge_volume(self, key, delta):
        return self.set_volume(key, self._volume[key] + int(delta))

    def step_volume(self, key, steps):
        return self.set_volume(key, stepped_level(self._volume[key], int(steps), self.cfg.VOLUME_STEP))

    def toggle_mute(self, key):
        return None

    def join(self, key):
        self._joined.add(key)
        return self.joined_keys()

    def leave(self, key):
        self._joined.discard(key)
        return self.joined_keys()

    def set_membership(self, joined):
        self._joined = {k for k in self.cfg.ROOM_KEYS if k in joined}
        return self.joined_keys()

    def joined_keys(self):
        return [k for k in self.cfg.ROOM_KEYS if k in self._joined]

    # -- enough of the rest to keep the legacy routes answering --------
    def scan(self):
        return {
            "players": [{"name": t["heos_name"], "pid": 1000 + i, "model": "Demo"}
                        for i, t in enumerate(self.cfg.TARGETS.values())],
            "groups": [],
        }

    def group_targets(self, host_key, member_keys):
        return self.set_membership(member_keys)

    def ungroup(self, key):
        return self.leave(key)

    def play_state(self, key, state):
        return None

    def skip(self, key, direction):
        return None

    def heos_inputs(self, key):
        return [{"name": s["name"], "input_id": f"inputs/{s['code'].lower()}"} for s in self.avr.inputs()]

    def play_heos_input(self, key, input_id, source_key=None):
        return None