music-assistant-server

1"""
2Reusable bridge roles for in-process Sendspin integrations.
3
4Provides a BridgePlayerRole that receives audio from Sendspin's PushStream
5and forwards it to an external player via callbacks. This role can be used
6by any bridge implementation (AirPlay, etc.) to integrate external players
7with Sendspin's synchronization and timing.
8
9Also provides BridgeVisualizerRole and BridgeColorRole for in-process
10consumers of the visualization pipeline (e.g. Hue Entertainment): feature
11extraction runs against the group's audio and results are delivered via
12callbacks instead of a WebSocket connection.
13"""
14
15from __future__ import annotations
16
17from collections.abc import Callable
18from typing import TYPE_CHECKING
19
20from aiosendspin.models.core import ServerStateMessage
21from aiosendspin.models.visualizer import BeatAvailability, StreamStartVisualizer
22from aiosendspin.server import VolumeChangedEvent
23from aiosendspin.server.roles import AudioRequirements, Role
24from aiosendspin.server.roles.registry import register_role
25from aiosendspin.server.roles.visualizer.features import VisualizerFeatureExtractor
26
27from music_assistant.mass import LOGGER
28
29if TYPE_CHECKING:
30    from aiosendspin.models.core import ServerStatePayload
31    from aiosendspin.models.types import ServerMessage
32    from aiosendspin.models.visualizer import BeatTiming, ClientHelloVisualizerSupport
33    from aiosendspin.server import SendspinClient
34    from aiosendspin.server.roles import AudioChunk
35    from aiosendspin.server.roles.visualizer.features import ExtractedFrame
36
37# Audio format constants for bridge players.
38BRIDGE_SAMPLE_RATE = 44100
39BRIDGE_BIT_DEPTH = 16
40BRIDGE_CHANNELS = 2
41BRIDGE_BYTES_PER_SAMPLE = BRIDGE_BIT_DEPTH // 8
42
43BRIDGE_ROLE_ID = "player@_bridge"
44VISUALIZER_BRIDGE_ROLE_ID = "visualizer@_bridge"
45COLOR_BRIDGE_ROLE_ID = "color@_bridge"
46
47
48class BridgePlayerRole(Role):
49    """
50    Custom Sendspin player role for external player bridges.
51
52    This role receives audio from Sendspin's PushStream and forwards it
53    to an external player via callbacks. It bypasses the normal WebSocket
54    audio delivery since external players don't have a WebSocket connection.
55
56    Created by the role factory registry. After creation, the bridge must
57    call set_callbacks() to wire up audio/volume/stream callbacks.
58    """
59
60    def __init__(self, client: SendspinClient) -> None:
61        """
62        Initialize the bridge player role.
63
64        :param client: The Sendspin client this role belongs to.
65        """
66        self._client = client
67        self._on_audio_chunk_cb: Callable[[AudioChunk], None] | None = None
68        self._on_volume_change_cb: Callable[[int], None] | None = None
69        self._on_mute_change_cb: Callable[[bool], None] | None = None
70        self._on_stream_start_cb: Callable[[], None] | None = None
71        self._on_stream_end_cb: Callable[[], None] | None = None
72        self._audio_requirements: AudioRequirements | None = None
73        self._volume: int = 100
74        self._muted: bool = False
75        # Bridge-specific timing reported to PushStream so server scheduling
76        # accounts for the downstream player's startup and jitter requirements.
77        self._required_lead_time_ms: int = 0
78        self._min_buffer_ms: int = 0
79
80    def set_callbacks(
81        self,
82        *,
83        on_audio_chunk: Callable[[AudioChunk], None],
84        on_volume_change: Callable[[int], None],
85        on_mute_change: Callable[[bool], None],
86        on_stream_start: Callable[[], None],
87        on_stream_end: Callable[[], None],
88        initial_volume: int = 100,
89        initial_muted: bool = False,
90    ) -> None:
91        """
92        Wire up bridge callbacks after role creation.
93
94        :param on_audio_chunk: Callback to receive audio chunks.
95        :param on_volume_change: Callback when volume level changes.
96        :param on_mute_change: Callback when mute state changes.
97        :param on_stream_start: Callback when the stream starts.
98        :param on_stream_end: Callback when the stream ends.
99        :param initial_volume: Initial volume level (0-100).
100        :param initial_muted: Initial mute state.
101        """
102        self._on_audio_chunk_cb = on_audio_chunk
103        self._on_volume_change_cb = on_volume_change
104        self._on_mute_change_cb = on_mute_change
105        self._on_stream_start_cb = on_stream_start
106        self._on_stream_end_cb = on_stream_end
107        self.update_player_state(volume=initial_volume, muted=initial_muted)
108
109    @property
110    def role_id(self) -> str:
111        """Return role identifier."""
112        return BRIDGE_ROLE_ID
113
114    @property
115    def role_family(self) -> str:
116        """Return role family name."""
117        return "player"
118
119    def setup_audio_requirements(
120        self,
121        sample_rate: int = BRIDGE_SAMPLE_RATE,
122        bit_depth: int = BRIDGE_BIT_DEPTH,
123        channels: int = BRIDGE_CHANNELS,
124    ) -> None:
125        """
126        Set up audio requirements for bridge PCM format.
127
128        Call with the sink's native rate/depth so MA transcodes to the
129        correct format before delivering chunks. Defaults to the bridge
130        constants for callers that don't need format negotiation.
131        """
132        self._audio_requirements = AudioRequirements(
133            sample_rate=sample_rate,
134            bit_depth=bit_depth,
135            channels=channels,
136            transformer=None,
137        )
138
139    @property
140    def preferred_format(self) -> AudioRequirements | None:
141        """Return the audio format declared via setup_audio_requirements."""
142        return self._audio_requirements
143
144    def get_audio_requirements(self) -> AudioRequirements | None:
145        """Return audio requirements for PushStream."""
146        return self._audio_requirements
147
148    def set_timing(self, *, required_lead_time_ms: int, min_buffer_ms: int) -> None:
149        """
150        Configure timing values reported to PushStream for scheduling.
151
152        Call before or during a stream session to adjust the lead time and
153        ongoing buffer floor reported to the Sendspin send-ahead calculation.
154        Bridge implementations should reflect their downstream player's real
155        startup latency (CLI spawn, protocol handshake, device buffer fill)
156        and jitter requirements.
157        """
158        self._required_lead_time_ms = max(0, required_lead_time_ms)
159        self._min_buffer_ms = max(0, min_buffer_ms)
160
161    def get_required_lead_time_us(self) -> int:
162        """Return bridge startup lead time in microseconds."""
163        return self._required_lead_time_ms * 1_000
164
165    def get_min_buffer_us(self) -> int:
166        """Return bridge minimum ongoing buffer duration in microseconds."""
167        return self._min_buffer_ms * 1_000
168
169    def get_player_volume(self) -> int | None:
170        """Return current volume level."""
171        return self._volume
172
173    def get_player_muted(self) -> bool | None:
174        """Return current mute state."""
175        return self._muted
176
177    def set_player_volume(self, volume: int) -> None:
178        """Set volume and notify bridge."""
179        self._volume = volume
180        self._emit_volume_changed()
181        if self._on_volume_change_cb:
182            self._on_volume_change_cb(volume)
183
184    def set_player_mute(self, muted: bool) -> None:
185        """Set mute state and notify bridge."""
186        self._muted = muted
187        self._emit_volume_changed()
188        if self._on_mute_change_cb:
189            self._on_mute_change_cb(muted)
190
191    def update_player_state(self, *, volume: int | None = None, muted: bool | None = None) -> None:
192        """
193        Adopt volume/mute state the bridged player is already in.
194
195        Use this instead of set_player_volume/set_player_mute for state that came
196        from the player's own side (device feedback, a mute applied elsewhere): the
197        bridge callbacks are not invoked, so the value is not pushed back at the
198        player it was just read from.
199
200        :param volume: Volume level (0-100) the player is at, or None to leave it.
201        :param muted: Mute state the player is in, or None to leave it.
202        """
203        changed = False
204        if volume is not None and volume != self._volume:
205            self._volume = volume
206            changed = True
207        if muted is not None and muted != self._muted:
208            self._muted = muted
209            changed = True
210        if changed:
211            self._emit_volume_changed()
212
213    def on_audio_chunk(self, chunk: AudioChunk) -> None:
214        """Receive audio chunk from PushStream and forward to callback."""
215        if self._on_audio_chunk_cb:
216            self._on_audio_chunk_cb(chunk)
217
218    def on_connect(self) -> None:
219        """Subscribe to PlayerGroupRole on attach."""
220        self._subscribe_to_group_role()
221
222    def on_disconnect(self) -> None:
223        """Unsubscribe from PlayerGroupRole on detach."""
224        self._unsubscribe_from_group_role()
225
226    def has_connection(self) -> bool:
227        """Return True to indicate bridge is "connected" for audio purposes."""
228        return True
229
230    def supports_preconnect_audio(self) -> bool:
231        """Return True -- bridge can receive audio before the stream starts."""
232        return True
233
234    def on_stream_start(self) -> None:
235        """Log stream start and invoke callback."""
236        LOGGER.debug("BridgePlayerRole stream started for client %s", self._client.client_id)
237        if self._on_stream_start_cb:
238            self._on_stream_start_cb()
239
240    def on_stream_end(self) -> None:
241        """Log stream end and invoke the stream-end callback."""
242        LOGGER.debug("BridgePlayerRole stream ended for client %s", self._client.client_id)
243        if self._on_stream_end_cb:
244            self._on_stream_end_cb()
245
246    def _emit_volume_changed(self) -> None:
247        """Emit VolumeChangedEvent so the SendspinPlayer stays in sync."""
248        self.emit_client_event(VolumeChangedEvent(volume=self._volume, muted=self._muted))
249
250
251class BridgeVisualizerRole(Role):
252    """
253    Custom Sendspin visualizer role for in-process bridges.
254
255    Runs the server's visualizer feature extraction on the group's audio and
256    forwards extracted frames, beat schedules, and stream lifecycle events to
257    callbacks instead of sending binary frames over a WebSocket.
258
259    Created by the role factory registry. After creation, the bridge must call
260    set_callbacks() and setup_visualizer(), then attach the roles via the
261    client's attach_preinitialized_roles(): external clients never get a
262    transport attach, so the group-role subscription (which beat schedules
263    and lifecycle broadcasts flow through) has to be made explicitly.
264    """
265
266    def __init__(self, client: SendspinClient) -> None:
267        """
268        Initialize the bridge visualizer role.
269
270        :param client: The Sendspin client this role belongs to.
271        """
272        self._client = client
273        self._support: ClientHelloVisualizerSupport | None = None
274        self._extractor: VisualizerFeatureExtractor | None = None
275        self._beat_availability = BeatAvailability.PENDING
276        self._on_frame_cb: Callable[[ExtractedFrame], None] | None = None
277        self._on_beats_cb: Callable[[list[BeatTiming]], None] | None = None
278        self._on_beats_clear_cb: Callable[[], None] | None = None
279        self._on_stream_start_cb: Callable[[], None] | None = None
280        self._on_stream_clear_cb: Callable[[], None] | None = None
281        self._on_stream_end_cb: Callable[[], None] | None = None
282
283    def set_callbacks(
284        self,
285        *,
286        on_frame: Callable[[ExtractedFrame], None],
287        on_beats: Callable[[list[BeatTiming]], None],
288        on_beats_clear: Callable[[], None],
289        on_stream_start: Callable[[], None],
290        on_stream_clear: Callable[[], None],
291        on_stream_end: Callable[[], None],
292    ) -> None:
293        """
294        Wire up bridge callbacks after role creation.
295
296        :param on_frame: Callback receiving extracted visualizer frames.
297        :param on_beats: Callback receiving beat schedule segments.
298        :param on_beats_clear: Callback when the beat schedule is dropped.
299        :param on_stream_start: Callback when the stream starts.
300        :param on_stream_clear: Callback when buffered stream state is cleared (seek).
301        :param on_stream_end: Callback when the stream ends.
302        """
303        self._on_frame_cb = on_frame
304        self._on_beats_cb = on_beats
305        self._on_beats_clear_cb = on_beats_clear
306        self._on_stream_start_cb = on_stream_start
307        self._on_stream_clear_cb = on_stream_clear
308        self._on_stream_end_cb = on_stream_end
309
310    def setup_visualizer(self, support: ClientHelloVisualizerSupport) -> None:
311        """
312        Configure feature extraction from the registered hello's support object.
313
314        :param support: The visualizer support object from the client hello.
315        """
316        self._support = support
317
318    @property
319    def role_id(self) -> str:
320        """Return role identifier."""
321        return VISUALIZER_BRIDGE_ROLE_ID
322
323    @property
324    def role_family(self) -> str:
325        """Return role family name."""
326        return "visualizer"
327
328    def get_audio_requirements(self) -> AudioRequirements:
329        """Return audio requirements for visualizer analysis."""
330        return AudioRequirements(
331            sample_rate=48_000,
332            bit_depth=16,
333            channels=2,
334            frame_duration_us=25_000,
335        )
336
337    def has_connection(self) -> bool:
338        """Return True to indicate bridge is "connected" for audio purposes."""
339        return True
340
341    def supports_preconnect_audio(self) -> bool:
342        """Return True -- bridge receives audio without a transport attach."""
343        return True
344
345    def replay_from_pcm_cache(self) -> bool:
346        """Replay buffered PCM on late join (visualizer is analysis-only)."""
347        return True
348
349    @property
350    def wants_beats(self) -> bool:
351        """True if the bridge requested beats and beats are not unavailable."""
352        return (
353            self._support is not None
354            and "beat" in self._support.types
355            and self._beat_availability is not BeatAvailability.UNAVAILABLE
356        )
357
358    def append_beats(self, beats: list[BeatTiming]) -> None:
359        """Forward a beat schedule segment to the bridge."""
360        if self._on_beats_cb and beats:
361            self._on_beats_cb(list(beats))
362
363    def clear_beats(self) -> None:
364        """Forward a beat-schedule clear to the bridge."""
365        if self._on_beats_clear_cb:
366            self._on_beats_clear_cb()
367
368    def set_beat_availability(self, availability: BeatAvailability) -> None:
369        """Track beat availability; UNAVAILABLE drops the delivered schedule."""
370        self._beat_availability = availability
371        if availability is BeatAvailability.UNAVAILABLE:
372            self.clear_beats()
373
374    def on_connect(self) -> None:
375        """Subscribe to VisualizerGroupRole on attach."""
376        self._subscribe_to_group_role()
377
378    def on_disconnect(self) -> None:
379        """Unsubscribe from VisualizerGroupRole on detach."""
380        self._unsubscribe_from_group_role()
381
382    def on_stream_start(self) -> None:
383        """Build a fresh extractor for the new stream and notify the bridge."""
384        if self._support is not None:
385            req = self.get_audio_requirements()
386            self._extractor = VisualizerFeatureExtractor(
387                sample_rate=req.sample_rate,
388                channels=req.channels,
389                config=StreamStartVisualizer.from_support(self._support),
390            )
391        LOGGER.debug("BridgeVisualizerRole stream started for client %s", self._client.client_id)
392        if self._on_stream_start_cb:
393            self._on_stream_start_cb()
394
395    def on_audio_chunk(self, chunk: AudioChunk) -> None:
396        """Extract features from group audio and forward each frame."""
397        if self._extractor is None or self._on_frame_cb is None:
398            return
399        for frame in self._extractor.process_chunk(chunk.data, chunk.timestamp_us):
400            self._on_frame_cb(frame)
401
402    def on_stream_clear(self) -> None:
403        """Reset extractor state on seek/clear and notify the bridge."""
404        if self._extractor is not None:
405            self._extractor.reset()
406        if self._on_stream_clear_cb:
407            self._on_stream_clear_cb()
408
409    def on_stream_end(self) -> None:
410        """Tear down the extractor and notify the bridge."""
411        self._extractor = None
412        LOGGER.debug("BridgeVisualizerRole stream ended for client %s", self._client.client_id)
413        if self._on_stream_end_cb:
414            self._on_stream_end_cb()
415
416
417class BridgeColorRole(Role):
418    """
419    Custom Sendspin color role for in-process bridges.
420
421    Receives color palette state updates from ColorGroupRole and forwards
422    them to a callback instead of sending server/state over a WebSocket.
423    """
424
425    def __init__(self, client: SendspinClient) -> None:
426        """
427        Initialize the bridge color role.
428
429        :param client: The Sendspin client this role belongs to.
430        """
431        self._client = client
432        self._on_color_cb: Callable[[ServerStatePayload], None] | None = None
433
434    def set_callbacks(self, *, on_color: Callable[[ServerStatePayload], None]) -> None:
435        """
436        Wire up the color update callback after role creation.
437
438        :param on_color: Callback receiving server/state payloads with color updates.
439        """
440        self._on_color_cb = on_color
441
442    @property
443    def role_id(self) -> str:
444        """Return role identifier."""
445        return COLOR_BRIDGE_ROLE_ID
446
447    @property
448    def role_family(self) -> str:
449        """Return role family name."""
450        return "color"
451
452    def has_connection(self) -> bool:
453        """Return True to indicate bridge is "connected"."""
454        return True
455
456    def send_message(self, message: ServerMessage) -> None:
457        """Intercept outbound server/state color updates and forward to the bridge."""
458        if isinstance(message, ServerStateMessage) and self._on_color_cb:
459            self._on_color_cb(message.payload)
460
461    def on_connect(self) -> None:
462        """Subscribe to ColorGroupRole on attach."""
463        self._subscribe_to_group_role()
464
465    def on_disconnect(self) -> None:
466        """Unsubscribe from ColorGroupRole on detach."""
467        self._unsubscribe_from_group_role()
468
469
470register_role(BRIDGE_ROLE_ID, lambda client: BridgePlayerRole(client=client))
471register_role(VISUALIZER_BRIDGE_ROLE_ID, lambda client: BridgeVisualizerRole(client=client))
472register_role(COLOR_BRIDGE_ROLE_ID, lambda client: BridgeColorRole(client=client))
473