Refactor Cocoa stuff into a 'receiver'

The idea here is that there are other places that might want to know
what's playing, besides MPNowPlayingInfoCenter. For example, to expose
the now playing info to Übersicht efficiently, it needs to be available
from a web browser, ideally using WebSockets. So there could be a
receiver that runs a small WebSockets server and sends out now playing
info to anyone who connects.

Additionally, I hope to write receivers for MPRIS and for the System
Media Transport Controls on Windows, making mpd-now-playable equally
useful across all platforms.

None of this is implemented yet, of course, but I hope to get the
WebSockets receiver done pretty soon!

I'm going to keep the default behaviour unchanged. Unless you
explicitly configure different receivers in config.toml,
mpd-now-playable will just behave as an MPNowPlayingInfoCenter
integration as it's always done.
This commit is contained in:
Danielle McLean 2024-07-09 12:52:49 +10:00
parent 27d8c37139
commit 00ba34bd0b
Signed by: 00dani
GPG key ID: 6854781A0488421C
12 changed files with 214 additions and 38 deletions

View file

@ -0,0 +1,76 @@
from asyncio import AbstractEventLoop, new_event_loop
from dataclasses import dataclass
from importlib import import_module
from typing import Generic, Iterable, Literal, Protocol, TypeVar, cast
from .config.model import BaseReceiverConfig, Config
from .player import Player
from .song import Song
from .tools.types import not_none
T = TypeVar("T", bound=AbstractEventLoop, covariant=True)
class LoopFactory(Generic[T], Protocol):
@property
def is_replaceable(self) -> bool: ...
@classmethod
def make_loop(cls) -> T: ...
class Receiver(Protocol):
def __init__(self, player: Player, config: Config) -> None: ...
@classmethod
def loop_factory(cls) -> LoopFactory[AbstractEventLoop]: ...
def update(self, song: Song | None) -> None: ...
class ReceiverModule(Protocol):
receiver: type[Receiver]
class DefaultLoopFactory(LoopFactory[AbstractEventLoop]):
@property
def is_replaceable(self) -> Literal[True]:
return True
@classmethod
def make_loop(cls) -> AbstractEventLoop:
return new_event_loop()
@dataclass
class IncompatibleReceiverError(Exception):
a: type[Receiver]
b: type[Receiver]
def import_receiver(config: BaseReceiverConfig) -> type[Receiver]:
mod = cast(
ReceiverModule, import_module(f"mpd_now_playable.receivers.{config.kind}")
)
return mod.receiver
def choose_loop_factory(
receivers: Iterable[type[Receiver]],
) -> LoopFactory[AbstractEventLoop]:
"""Given the desired receivers, determine which asyncio event loop implementation will support all of them. Will raise an IncompatibleReceiverError if no such implementation exists."""
chosen_fac: LoopFactory[AbstractEventLoop] = DefaultLoopFactory()
chosen_rec: type[Receiver] | None = None
for rec in receivers:
fac = rec.loop_factory()
if fac.is_replaceable:
continue
if chosen_fac.is_replaceable:
chosen_fac = fac
elif type(fac) is not type(chosen_fac):
raise IncompatibleReceiverError(rec, not_none(chosen_rec))
chosen_rec = rec
return chosen_fac