141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
"""The IR remote: a TCP client for lircd, translated into the same intents every other
|
|
front-end emits.
|
|
|
|
Connect, read lines until the link drops, wait, repeat - the same reconnect shape as
|
|
:class:`~musicmouse.devices.serial_link.SerialLink`, over a plain socket instead of a
|
|
serial port because lircd speaks its classic protocol on a bare TCP connection.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Callable
|
|
|
|
from musicmouse.app import App
|
|
from musicmouse.clock import Clock, RealClock
|
|
from musicmouse.config import Digit, LircConfig
|
|
from musicmouse.events import (
|
|
ConnectionChanged,
|
|
IntentEvent,
|
|
NextTrackRequested,
|
|
PauseRequested,
|
|
PlayAlbumRequested,
|
|
PlayRequested,
|
|
PlaySeriesLatestRequested,
|
|
PreviousTrackRequested,
|
|
SetVolumeRequested,
|
|
VolumeChangeRequested,
|
|
)
|
|
from musicmouse.services.lirc.protocol import LircButtonEvent, parse_line
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["LircService"]
|
|
|
|
#: Acted on only at repeat 0 (the first press) - holding the button must not replay it.
|
|
_TRANSPORT: dict[str, Callable[[], IntentEvent]] = {
|
|
"KEY_PLAY": lambda: PlayRequested(source="lirc"),
|
|
# No separate "stop" concept exists in this player; the remote's stop button just
|
|
# pauses, like its pause button.
|
|
"KEY_PAUSE": lambda: PauseRequested(source="lirc"),
|
|
"KEY_STOP": lambda: PauseRequested(source="lirc"),
|
|
"KEY_PREVIOUS": lambda: PreviousTrackRequested(source="lirc"),
|
|
"KEY_REWIND": lambda: PreviousTrackRequested(source="lirc"),
|
|
"KEY_FORWARD": lambda: NextTrackRequested(source="lirc"),
|
|
}
|
|
|
|
#: Acted on every repeat, for a continuous ramp while held - same feel as the rotary
|
|
#: encoder (see ``reactions.playback.rotary_turned``).
|
|
_VOLUME: dict[str, int] = {"KEY_VOLUMEUP": 1, "KEY_VOLUMEDOWN": -1}
|
|
|
|
_DIGITS: dict[str, Digit] = {
|
|
"BTN_0": "0",
|
|
"BTN_1": "1",
|
|
"BTN_2": "2",
|
|
"BTN_3": "3",
|
|
"BTN_4": "4",
|
|
"BTN_5": "5",
|
|
"BTN_6": "6",
|
|
"BTN_7": "7",
|
|
"BTN_8": "8",
|
|
"BTN_9": "9",
|
|
}
|
|
|
|
|
|
class LircService:
|
|
name = "lirc"
|
|
|
|
def __init__(self, app: App, config: LircConfig, *, clock: Clock | None = None) -> None:
|
|
self.app = app
|
|
self.config = config
|
|
self._clock = clock or RealClock()
|
|
#: Volume to restore on the next mute press, remembered the way the frontend's
|
|
#: own mute toggle does (App.tsx) rather than via any new Player API.
|
|
self._pre_mute_volume: int | None = None
|
|
|
|
async def run(self) -> None:
|
|
"""Connect, read lines until the link drops, wait, repeat. Runs until cancelled."""
|
|
while True:
|
|
try:
|
|
await self._session()
|
|
except OSError as exc:
|
|
_log.warning(
|
|
"lircd link to %s:%d lost (%s); retrying in %gs",
|
|
self.config.host,
|
|
self.config.port,
|
|
exc,
|
|
self.config.reconnect_interval,
|
|
)
|
|
await self._clock.sleep(self.config.reconnect_interval)
|
|
|
|
async def _session(self) -> None:
|
|
reader, writer = await asyncio.open_connection(self.config.host, self.config.port)
|
|
_log.info("Connected to lircd at %s:%d", self.config.host, self.config.port)
|
|
self.app.bus.emit(ConnectionChanged(target="lirc", connected=True, source="lirc"))
|
|
try:
|
|
while True:
|
|
raw = await reader.readline()
|
|
if not raw:
|
|
return
|
|
event = parse_line(raw.decode(errors="replace"))
|
|
if event is None or event.remote != self.config.remote_name:
|
|
continue
|
|
self._handle(event)
|
|
finally:
|
|
writer.close()
|
|
self.app.bus.emit(ConnectionChanged(target="lirc", connected=False, source="lirc"))
|
|
|
|
def _handle(self, event: LircButtonEvent) -> None:
|
|
if event.button in _TRANSPORT:
|
|
if event.repeat == 0:
|
|
self.app.bus.emit(_TRANSPORT[event.button]())
|
|
elif event.button in _VOLUME:
|
|
step = self.app.config.general.volume_increment * _VOLUME[event.button]
|
|
self.app.bus.emit(VolumeChangeRequested(delta=step, source="lirc"))
|
|
elif event.button == "KEY_MUTE":
|
|
if event.repeat == 0:
|
|
self._toggle_mute()
|
|
elif (digit := _DIGITS.get(event.button)) is not None and event.repeat == 0:
|
|
self._play_digit(digit)
|
|
|
|
def _toggle_mute(self) -> None:
|
|
player = self.app.player
|
|
if player.volume > 0:
|
|
self._pre_mute_volume = player.volume
|
|
self.app.bus.emit(SetVolumeRequested(volume=0, source="lirc"))
|
|
else:
|
|
restore = self._pre_mute_volume or self.app.config.general.initial_volume
|
|
self.app.bus.emit(SetVolumeRequested(volume=restore, source="lirc"))
|
|
|
|
def _play_digit(self, digit: Digit) -> None:
|
|
slot = self.app.config.remote.get(digit)
|
|
if slot is None:
|
|
return
|
|
if slot.target_kind == "album":
|
|
self.app.bus.emit(
|
|
PlayAlbumRequested(album_id=slot.target, track_index=0, source="lirc")
|
|
)
|
|
else:
|
|
self.app.bus.emit(PlaySeriesLatestRequested(series=slot.target, source="lirc"))
|