franzzandClaude Opus 5
375bcfdd76
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>
167 lines
7.0 KiB
Python
167 lines
7.0 KiB
Python
"""Run against the fake hardware in fakes.py:
|
|
|
|
python3 -m unittest discover -s tests -t .
|
|
"""
|
|
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from controller import Controller # noqa: E402
|
|
from tests.fakes import FakeAvr, FakeHeos # noqa: E402
|
|
|
|
PAIR = {3, 4} # the two Home 200s
|
|
AVR_PID = 1
|
|
HOME400_PID = 2
|
|
|
|
|
|
def build(tmpdir):
|
|
heos, avr = FakeHeos(), FakeAvr()
|
|
cfg = SimpleNamespace(
|
|
HEOS_HOST="127.0.0.1", HEOS_PORT=heos.port,
|
|
AVR_HOST="127.0.0.1", AVR_PORT=avr.port,
|
|
HOST_KEY="avr",
|
|
ROOM_KEYS=["home400", "living_room_group"],
|
|
TARGETS={
|
|
"avr": {"label": "Home Cinema", "heos_name": "Home Cinema"},
|
|
"home400": {"label": "Lego Room", "heos_name": "Lego Room"},
|
|
"living_room_group": {"label": "Living Room", "heos_name": "Denon Home 200 L"},
|
|
},
|
|
AVR_INPUT_CODES=[],
|
|
MEMBERS_FILE=str(Path(tmpdir) / "members.json"),
|
|
)
|
|
return Controller(cfg), heos, avr
|
|
|
|
|
|
class PanelTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.panel, self.heos, self.avr = build(self.tmp.name)
|
|
|
|
def group_pids(self):
|
|
return {gid: set(pids) for gid, pids in self.heos.groups.items()}
|
|
|
|
# -- resolving ------------------------------------------------------
|
|
def test_pair_resolves_to_both_speakers(self):
|
|
"""The bug this replaces: grouping used only the pair's leader,
|
|
which left the second Home 200 behind."""
|
|
self.panel.scan()
|
|
self.assertEqual(set(self.panel.member_pids("living_room_group")), PAIR)
|
|
self.assertEqual(self.panel.member_pids("home400"), [HOME400_PID])
|
|
|
|
def test_avr_resolves_to_a_player_even_while_it_leads_a_group(self):
|
|
self.panel.join("home400")
|
|
# HEOS now reports a *group* named "Home Cinema" as well as the player.
|
|
self.assertEqual(self.panel.member_pids("avr"), [AVR_PID])
|
|
|
|
# -- grouping -------------------------------------------------------
|
|
def test_joining_takes_the_whole_pair(self):
|
|
self.panel.join("living_room_group")
|
|
self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR})
|
|
self.assertEqual(self.panel.joined_keys(), ["living_room_group"])
|
|
|
|
def test_joining_keeps_whoever_is_already_grouped(self):
|
|
self.panel.join("home400")
|
|
self.panel.join("living_room_group")
|
|
self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, HOME400_PID} | PAIR})
|
|
self.assertEqual(self.panel.joined_keys(), ["home400", "living_room_group"])
|
|
|
|
def test_leaving_rebuilds_the_stereo_pair(self):
|
|
self.panel.join("living_room_group")
|
|
self.panel.leave("living_room_group")
|
|
self.assertEqual(self.group_pids(), {3: PAIR}) # pair back, AVR alone
|
|
self.assertEqual(self.panel.joined_keys(), [])
|
|
|
|
def test_leaving_one_room_does_not_disturb_the_other(self):
|
|
self.panel.join("home400")
|
|
self.panel.join("living_room_group")
|
|
before = [c for c in self.heos.commands if "set_group" in c]
|
|
self.panel.leave("home400")
|
|
after = [c for c in self.heos.commands if "set_group" in c]
|
|
self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR})
|
|
# Exactly one new set_group: the remaining room is never regrouped,
|
|
# which is what stops its music restarting.
|
|
self.assertEqual(len(after) - len(before), 1)
|
|
|
|
def test_separate_everything(self):
|
|
self.panel.set_membership(["home400", "living_room_group"])
|
|
self.panel.set_membership([])
|
|
self.assertEqual(self.group_pids(), {3: PAIR})
|
|
self.assertEqual(self.panel.joined_keys(), [])
|
|
|
|
def test_membership_survives_a_restart_while_merged(self):
|
|
"""Once merged, the pair's own group is gone from HEOS, so a fresh
|
|
process has to fall back on what it learned earlier."""
|
|
self.panel.join("living_room_group")
|
|
reborn = Controller(self.panel.cfg)
|
|
reborn.scan()
|
|
self.assertEqual(set(reborn.member_pids("living_room_group")), PAIR)
|
|
reborn.leave("living_room_group")
|
|
self.assertEqual(self.group_pids(), {3: PAIR})
|
|
|
|
# -- volume ---------------------------------------------------------
|
|
def test_pair_uses_group_volume_when_it_stands_alone(self):
|
|
self.panel.scan()
|
|
self.assertEqual(self.panel.set_volume("living_room_group", 42), 42)
|
|
self.assertEqual(self.heos.group_volumes[3], 42)
|
|
|
|
def test_pair_uses_player_volume_once_merged(self):
|
|
"""Its gid stops existing the moment it joins the AVR, so the old
|
|
bridge's get_volume?gid= call would simply fail here."""
|
|
self.panel.join("living_room_group")
|
|
self.assertEqual(self.panel.set_volume("living_room_group", 31), 31)
|
|
self.assertEqual(self.heos.volumes[3], 31)
|
|
self.assertEqual(self.heos.volumes[4], 31)
|
|
self.assertEqual(self.panel.volume("living_room_group"), 31)
|
|
|
|
def test_nudge_clamps_at_the_ends(self):
|
|
self.panel.set_volume("home400", 98)
|
|
self.assertEqual(self.panel.nudge_volume("home400", 5), 100)
|
|
self.panel.set_volume("home400", 1)
|
|
self.assertEqual(self.panel.nudge_volume("home400", -9), 0)
|
|
|
|
# -- the AVR ---------------------------------------------------------
|
|
def test_renamed_inputs_and_selection(self):
|
|
deadline = time.time() + 5
|
|
while not self.panel.avr.connected and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
self.assertTrue(self.panel.avr.connected)
|
|
|
|
self.assertEqual(
|
|
self.panel.avr.inputs(),
|
|
[{"code": "MPLAY", "name": "Apple TV"},
|
|
{"code": "GAME", "name": "PlayStation"},
|
|
{"code": "SAT/CBL", "name": "TV Box"}],
|
|
)
|
|
self.assertEqual(self.panel.avr.current_input(), {"code": "MPLAY", "name": "Apple TV"})
|
|
self.assertEqual(self.panel.avr.select_input("GAME"), {"code": "GAME", "name": "PlayStation"})
|
|
self.assertEqual(self.avr.input, "GAME")
|
|
|
|
# -- the whole snapshot the UI renders -------------------------------
|
|
def test_state_snapshot(self):
|
|
self.panel.join("home400")
|
|
state = self.panel.state()
|
|
self.assertTrue(state["heos_ok"])
|
|
self.assertEqual([r["key"] for r in state["rooms"]], ["home400", "living_room_group"])
|
|
self.assertEqual([r["grouped"] for r in state["rooms"]], [True, False])
|
|
self.assertTrue(all(isinstance(r["volume"], int) for r in state["rooms"]))
|
|
|
|
def test_state_reports_trouble_instead_of_blowing_up(self):
|
|
self.panel.heos.host = "127.0.0.1"
|
|
self.panel.heos.port = 1 # nothing is listening there
|
|
self.panel.heos._close()
|
|
state = self.panel.state()
|
|
self.assertFalse(state["heos_ok"])
|
|
self.assertTrue(state["errors"])
|
|
self.assertTrue(all(r["available"] is False for r in state["rooms"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|