Add IR remote control (LIRC) with a number-key content mapping
Adds a TCP client for lircd's classic protocol: play/pause/next/prev/ volume/mute map to the same intents every other front-end already emits, and number keys 0-9 play an assigned album/audiobook from the start or a podcast show's newest episode, resolved fresh on every press. The mapping is configured in config.yml and editable from the frontend: a small "Taste zuweisen" button on the play screen (or the A+digit keyboard shortcut) opens a 10-key picker to assign whatever is currently playing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
342
python-backend/tests/test_lirc.py
Normal file
342
python-backend/tests/test_lirc.py
Normal file
@@ -0,0 +1,342 @@
|
||||
"""The IR remote: line parsing, and the service against a scripted lircd."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
from collections.abc import AsyncIterator
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from musicmouse.clock import FakeClock
|
||||
from musicmouse.config import LircConfig, RemoteSlotConfig, load_config
|
||||
from musicmouse.events import (
|
||||
ConnectionChanged,
|
||||
Event,
|
||||
IntentEvent,
|
||||
NextTrackRequested,
|
||||
PauseRequested,
|
||||
PlayAlbumRequested,
|
||||
PlayRequested,
|
||||
PlaySeriesLatestRequested,
|
||||
PrevTrackRequested,
|
||||
VolumeChangeRequested,
|
||||
)
|
||||
from musicmouse.services.lirc import LircService
|
||||
from musicmouse.services.lirc.protocol import LircButtonEvent, parse_line
|
||||
from musicmouse.simulator.harness import Simulation, build_simulation
|
||||
from tests.conftest import VALID_CONFIG, write_config
|
||||
|
||||
# ---------------------------------------------------------------------- parse_line
|
||||
|
||||
|
||||
def test_a_valid_line_is_parsed() -> None:
|
||||
assert parse_line("0000000000001781 00 BTN_1 Hauppauge") == LircButtonEvent(
|
||||
code="0000000000001781", repeat=0, button="BTN_1", remote="Hauppauge"
|
||||
)
|
||||
|
||||
|
||||
def test_a_repeat_field_is_hexadecimal() -> None:
|
||||
event = parse_line("0000000000001781 0a BTN_1 Hauppauge")
|
||||
assert event is not None
|
||||
assert event.repeat == 10
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"line",
|
||||
[
|
||||
"",
|
||||
"0000000000001781 00 BTN_1", # missing remote
|
||||
"0000000000001781 00 BTN_1 Hauppauge extra", # extra token
|
||||
"0000000000001781 zz BTN_1 Hauppauge", # non-hex repeat
|
||||
],
|
||||
)
|
||||
def test_malformed_lines_are_rejected(line: str) -> None:
|
||||
assert parse_line(line) is None
|
||||
|
||||
|
||||
def test_surrounding_whitespace_is_ignored() -> None:
|
||||
assert parse_line(" 0000000000001781 00 BTN_1 Hauppauge \n") is not None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- service
|
||||
|
||||
|
||||
class ScriptedLircd:
|
||||
"""A tiny stand-in for lircd: accepts one connection at a time and lets the test
|
||||
push lines to whoever is currently connected."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._writer: asyncio.StreamWriter | None = None
|
||||
self._connected = asyncio.Event()
|
||||
self.connection_count = 0
|
||||
|
||||
async def _handle(self, _reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
self._writer = writer
|
||||
self.connection_count += 1
|
||||
self._connected.set()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await asyncio.Event().wait() # held open until the test drops it
|
||||
|
||||
async def wait_connected(self) -> None:
|
||||
await self._connected.wait()
|
||||
|
||||
async def send(self, line: str) -> None:
|
||||
await self._connected.wait()
|
||||
assert self._writer is not None
|
||||
self._writer.write(f"{line}\n".encode())
|
||||
await self._writer.drain()
|
||||
# Real socket I/O, not the fake clock: give the reader a moment to actually
|
||||
# see the bytes before the caller checks what happened.
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
def drop(self) -> None:
|
||||
"""Simulate the link dying: close the socket, and wait for a new connect."""
|
||||
assert self._writer is not None
|
||||
self._writer.close()
|
||||
self._writer = None
|
||||
self._connected = asyncio.Event()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def lircd() -> AsyncIterator[tuple[ScriptedLircd, asyncio.base_events.Server]]:
|
||||
station = ScriptedLircd()
|
||||
server = await asyncio.start_server(station._handle, "127.0.0.1", 0)
|
||||
try:
|
||||
yield station, server
|
||||
finally:
|
||||
server.close()
|
||||
# The scripted connection handler holds its connection open forever (it does
|
||||
# not know the test is done), so `wait_closed()` alone would hang here.
|
||||
server.close_clients()
|
||||
await server.wait_closed()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def sim(config_dir: Path) -> AsyncIterator[Simulation]:
|
||||
config = load_config(write_config(config_dir, VALID_CONFIG))
|
||||
# Real time for the socket I/O itself, virtual time for the reconnect sleep - tests
|
||||
# drive that explicitly via `sim.clock.advance()` and drain the bus after.
|
||||
simulation = await build_simulation(config, clock=FakeClock())
|
||||
try:
|
||||
yield simulation
|
||||
finally:
|
||||
await simulation.aclose()
|
||||
|
||||
|
||||
def _lirc_config(port: int, *, reconnect_interval: float = 5.0) -> LircConfig:
|
||||
return LircConfig(host="127.0.0.1", port=port, reconnect_interval=reconnect_interval)
|
||||
|
||||
|
||||
async def _run_service(sim: Simulation, config: LircConfig) -> asyncio.Task[None]:
|
||||
service = LircService(sim.app, config, clock=sim.clock)
|
||||
return asyncio.create_task(service.run())
|
||||
|
||||
|
||||
async def test_transport_and_volume_buttons_emit_intents(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[Event] = []
|
||||
sim.bus.subscribe_all(seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 KEY_PLAY Hauppauge")
|
||||
await station.send("0 00 KEY_PAUSE Hauppauge")
|
||||
await station.send("0 00 KEY_STOP Hauppauge")
|
||||
await station.send("0 00 KEY_PREVIOUS Hauppauge")
|
||||
await station.send("0 00 KEY_FORWARD Hauppauge")
|
||||
await station.send("0 00 KEY_VOLUMEUP Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
kinds = [type(event) for event in seen if isinstance(event, IntentEvent)]
|
||||
assert kinds == [
|
||||
PlayRequested,
|
||||
PauseRequested,
|
||||
PauseRequested,
|
||||
PrevTrackRequested,
|
||||
NextTrackRequested,
|
||||
VolumeChangeRequested,
|
||||
]
|
||||
|
||||
|
||||
async def test_events_from_another_remote_are_ignored(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[Event] = []
|
||||
sim.bus.subscribe_all(seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 KEY_PLAY small_led_remote")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert not any(isinstance(event, IntentEvent) for event in seen)
|
||||
|
||||
|
||||
async def test_transport_buttons_only_act_on_the_first_press(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[Event] = []
|
||||
sim.bus.subscribe(NextTrackRequested, seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 KEY_FORWARD Hauppauge")
|
||||
await station.send("0 01 KEY_FORWARD Hauppauge")
|
||||
await station.send("0 02 KEY_FORWARD Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert len(seen) == 1
|
||||
|
||||
|
||||
async def test_volume_buttons_act_on_every_repeat(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[VolumeChangeRequested] = []
|
||||
sim.bus.subscribe(VolumeChangeRequested, seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 KEY_VOLUMEDOWN Hauppauge")
|
||||
await station.send("0 01 KEY_VOLUMEDOWN Hauppauge")
|
||||
await station.send("0 02 KEY_VOLUMEDOWN Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert len(seen) == 3
|
||||
assert all(event.delta < 0 for event in seen)
|
||||
|
||||
|
||||
async def test_mute_toggles_and_restores(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
sim.player.set_volume(37)
|
||||
await sim.bus.drain()
|
||||
|
||||
await station.send("0 00 KEY_MUTE Hauppauge")
|
||||
await sim.bus.drain()
|
||||
assert sim.player.volume == 0
|
||||
|
||||
await station.send("0 00 KEY_MUTE Hauppauge")
|
||||
await sim.bus.drain()
|
||||
assert sim.player.volume == 37
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
|
||||
async def test_a_digit_with_no_assignment_does_nothing(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[Event] = []
|
||||
sim.bus.subscribe(PlayAlbumRequested, seen.append)
|
||||
sim.bus.subscribe(PlaySeriesLatestRequested, seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 BTN_5 Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert seen == []
|
||||
|
||||
|
||||
async def test_an_album_digit_plays_from_the_beginning(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
album = next(a for a in sim.app.library.albums if a.title == "Kinderparty Lieder")
|
||||
sim.app.config.remote = {"3": RemoteSlotConfig(target_kind="album", target=album.id)}
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 BTN_3 Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert sim.player.playlist is not None
|
||||
assert sim.player.playlist.album_id == album.id
|
||||
assert sim.player.track_index == 0
|
||||
|
||||
|
||||
async def test_a_series_digit_plays_the_latest_episode(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
sim.app.config.remote = {"7": RemoteSlotConfig(target_kind="series", target="Wissen macht Ah")}
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await station.send("0 00 BTN_7 Hauppauge")
|
||||
await sim.bus.drain()
|
||||
finally:
|
||||
task.cancel()
|
||||
|
||||
assert sim.player.playlist is not None
|
||||
played = sim.app.library.get(sim.player.playlist.album_id)
|
||||
assert played is not None
|
||||
assert played.title == "Neu"
|
||||
|
||||
|
||||
async def test_a_dropped_link_reconnects(
|
||||
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
||||
) -> None:
|
||||
station, server = lircd
|
||||
port = server.sockets[0].getsockname()[1]
|
||||
seen: list[ConnectionChanged] = []
|
||||
sim.bus.subscribe(ConnectionChanged, seen.append)
|
||||
|
||||
task = await _run_service(sim, _lirc_config(port, reconnect_interval=10.0))
|
||||
try:
|
||||
await station.wait_connected()
|
||||
await sim.bus.drain()
|
||||
assert station.connection_count == 1
|
||||
|
||||
station.drop()
|
||||
await asyncio.sleep(0.05) # let the client notice EOF
|
||||
await sim.bus.drain()
|
||||
|
||||
assert isinstance(sim.clock, FakeClock)
|
||||
await sim.clock.advance(10.0)
|
||||
await station.wait_connected()
|
||||
await sim.bus.drain()
|
||||
|
||||
assert station.connection_count == 2
|
||||
assert [event.connected for event in seen] == [True, False, True]
|
||||
finally:
|
||||
task.cancel()
|
||||
Reference in New Issue
Block a user