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    ) -> None:
90        """
91        Wire up bridge callbacks after role creation.
92
93        :param on_audio_chunk: Callback to receive audio chunks.
94        :param on_volume_change: Callback when volume level changes.
95        :param on_mute_change: Callback when mute state changes.
96        :param on_stream_start: Callback when the stream starts.
97        :param on_stream_end: Callback when the stream ends.
98        :param initial_volume: Initial volume level (0-100).
99        """
100        self._on_audio_chunk_cb = on_audio_chunk
101        self._on_volume_change_cb = on_volume_change
102        self._on_mute_change_cb = on_mute_change
103        self._on_stream_start_cb = on_stream_start
104        self._on_stream_end_cb = on_stream_end
105        volume_changed = self._volume != initial_volume
106        self._volume = initial_volume
107        if volume_changed:
108            self._emit_volume_changed()
109
110    @property
111    def role_id(self) -> str:
112        """Return role identifier."""
113        return BRIDGE_ROLE_ID
114
115    @property
116    def role_family(self) -> str:
117        """Return role family name."""
118        return "player"
119
120    def setup_audio_requirements(
121        self,
122        sample_rate: int = BRIDGE_SAMPLE_RATE,
123        bit_depth: int = BRIDGE_BIT_DEPTH,
124        channels: int = BRIDGE_CHANNELS,
125    ) -> None:
126        """
127        Set up audio requirements for bridge PCM format.
128
129        Call with the sink's native rate/depth so MA transcodes to the
130        correct format before delivering chunks. Defaults to the bridge
131        constants for callers that don't need format negotiation.
132        """
133        self._audio_requirements = AudioRequirements(
134            sample_rate=sample_rate,
135            bit_depth=bit_depth,
136            channels=channels,
137            transformer=None,
138        )
139
140    @property
141    def preferred_format(self) -> AudioRequirements | None:
142        """Return the audio format declared via setup_audio_requirements."""
143        return self._audio_requirements
144
145    def get_audio_requirements(self) -> AudioRequirements | None:
146        """Return audio requirements for PushStream."""
147        return self._audio_requirements
148
149    def set_timing(self, *, required_lead_time_ms: int, min_buffer_ms: int) -> None:
150        """
151        Configure timing values reported to PushStream for scheduling.
152
153        Call before or during a stream session to adjust the lead time and
154        ongoing buffer floor reported to the Sendspin send-ahead calculation.
155        Bridge implementations should reflect their downstream player's real
156        startup latency (CLI spawn, protocol handshake, device buffer fill)
157        and jitter requirements.
158        """
159        self._required_lead_time_ms = max(0, required_lead_time_ms)
160        self._min_buffer_ms = max(0, min_buffer_ms)
161
162    def get_required_lead_time_us(self) -> int:
163        """Return bridge startup lead time in microseconds."""
164        return self._required_lead_time_ms * 1_000
165
166    def get_min_buffer_us(self) -> int:
167        """Return bridge minimum ongoing buffer duration in microseconds."""
168        return self._min_buffer_ms * 1_000
169
170    def get_player_volume(self) -> int | None:
171        """Return current volume level."""
172        return self._volume
173
174    def get_player_muted(self) -> bool | None:
175        """Return current mute state."""
176        return self._muted
177
178    def set_player_volume(self, volume: int) -> None:
179        """Set volume and notify bridge."""
180        self._volume = volume
181        self._emit_volume_changed()
182        if self._on_volume_change_cb:
183            self._on_volume_change_cb(volume)
184
185    def set_player_mute(self, muted: bool) -> None:
186        """Set mute state and notify bridge."""
187        self._muted = muted
188        self._emit_volume_changed()
189        if self._on_mute_change_cb:
190            self._on_mute_change_cb(muted)
191
192    def on_audio_chunk(self, chunk: AudioChunk) -> None:
193        """Receive audio chunk from PushStream and forward to callback."""
194        if self._on_audio_chunk_cb:
195            self._on_audio_chunk_cb(chunk)
196
197    def on_connect(self) -> None:
198        """Subscribe to PlayerGroupRole on attach."""
199        self._subscribe_to_group_role()
200
201    def on_disconnect(self) -> None:
202        """Unsubscribe from PlayerGroupRole on detach."""
203        self._unsubscribe_from_group_role()
204
205    def has_connection(self) -> bool:
206        """Return True to indicate bridge is "connected" for audio purposes."""
207        return True
208
209    def supports_preconnect_audio(self) -> bool:
210        """Return True -- bridge can receive audio before the stream starts."""
211        return True
212
213    def on_stream_start(self) -> None:
214        """Log stream start and invoke callback."""
215        LOGGER.debug("BridgePlayerRole stream started for client %s", self._client.client_id)
216        if self._on_stream_start_cb:
217            self._on_stream_start_cb()
218
219    def on_stream_end(self) -> None:
220        """Log stream end and invoke the stream-end callback."""
221        LOGGER.debug("BridgePlayerRole stream ended for client %s", self._client.client_id)
222        if self._on_stream_end_cb:
223            self._on_stream_end_cb()
224
225    def _emit_volume_changed(self) -> None:
226        """Emit VolumeChangedEvent so the SendspinPlayer stays in sync."""
227        self.emit_client_event(VolumeChangedEvent(volume=self._volume, muted=self._muted))
228
229
230class BridgeVisualizerRole(Role):
231    """
232    Custom Sendspin visualizer role for in-process bridges.
233
234    Runs the server's visualizer feature extraction on the group's audio and
235    forwards extracted frames, beat schedules, and stream lifecycle events to
236    callbacks instead of sending binary frames over a WebSocket.
237
238    Created by the role factory registry. After creation, the bridge must call
239    set_callbacks() and setup_visualizer(), then attach the roles via the
240    client's attach_preinitialized_roles(): external clients never get a
241    transport attach, so the group-role subscription (which beat schedules
242    and lifecycle broadcasts flow through) has to be made explicitly.
243    """
244
245    def __init__(self, client: SendspinClient) -> None:
246        """
247        Initialize the bridge visualizer role.
248
249        :param client: The Sendspin client this role belongs to.
250        """
251        self._client = client
252        self._support: ClientHelloVisualizerSupport | None = None
253        self._extractor: VisualizerFeatureExtractor | None = None
254        self._beat_availability = BeatAvailability.PENDING
255        self._on_frame_cb: Callable[[ExtractedFrame], None] | None = None
256        self._on_beats_cb: Callable[[list[BeatTiming]], None] | None = None
257        self._on_beats_clear_cb: Callable[[], None] | None = None
258        self._on_stream_start_cb: Callable[[], None] | None = None
259        self._on_stream_clear_cb: Callable[[], None] | None = None
260        self._on_stream_end_cb: Callable[[], None] | None = None
261
262    def set_callbacks(
263        self,
264        *,
265        on_frame: Callable[[ExtractedFrame], None],
266        on_beats: Callable[[list[BeatTiming]], None],
267        on_beats_clear: Callable[[], None],
268        on_stream_start: Callable[[], None],
269        on_stream_clear: Callable[[], None],
270        on_stream_end: Callable[[], None],
271    ) -> None:
272        """
273        Wire up bridge callbacks after role creation.
274
275        :param on_frame: Callback receiving extracted visualizer frames.
276        :param on_beats: Callback receiving beat schedule segments.
277        :param on_beats_clear: Callback when the beat schedule is dropped.
278        :param on_stream_start: Callback when the stream starts.
279        :param on_stream_clear: Callback when buffered stream state is cleared (seek).
280        :param on_stream_end: Callback when the stream ends.
281        """
282        self._on_frame_cb = on_frame
283        self._on_beats_cb = on_beats
284        self._on_beats_clear_cb = on_beats_clear
285        self._on_stream_start_cb = on_stream_start
286        self._on_stream_clear_cb = on_stream_clear
287        self._on_stream_end_cb = on_stream_end
288
289    def setup_visualizer(self, support: ClientHelloVisualizerSupport) -> None:
290        """
291        Configure feature extraction from the registered hello's support object.
292
293        :param support: The visualizer support object from the client hello.
294        """
295        self._support = support
296
297    @property
298    def role_id(self) -> str:
299        """Return role identifier."""
300        return VISUALIZER_BRIDGE_ROLE_ID
301
302    @property
303    def role_family(self) -> str:
304        """Return role family name."""
305        return "visualizer"
306
307    def get_audio_requirements(self) -> AudioRequirements:
308        """Return audio requirements for visualizer analysis."""
309        return AudioRequirements(
310            sample_rate=48_000,
311            bit_depth=16,
312            channels=2,
313            frame_duration_us=25_000,
314        )
315
316    def has_connection(self) -> bool:
317        """Return True to indicate bridge is "connected" for audio purposes."""
318        return True
319
320    def supports_preconnect_audio(self) -> bool:
321        """Return True -- bridge receives audio without a transport attach."""
322        return True
323
324    def replay_from_pcm_cache(self) -> bool:
325        """Replay buffered PCM on late join (visualizer is analysis-only)."""
326        return True
327
328    @property
329    def wants_beats(self) -> bool:
330        """True if the bridge requested beats and beats are not unavailable."""
331        return (
332            self._support is not None
333            and "beat" in self._support.types
334            and self._beat_availability is not BeatAvailability.UNAVAILABLE
335        )
336
337    def append_beats(self, beats: list[BeatTiming]) -> None:
338        """Forward a beat schedule segment to the bridge."""
339        if self._on_beats_cb and beats:
340            self._on_beats_cb(list(beats))
341
342    def clear_beats(self) -> None:
343        """Forward a beat-schedule clear to the bridge."""
344        if self._on_beats_clear_cb:
345            self._on_beats_clear_cb()
346
347    def set_beat_availability(self, availability: BeatAvailability) -> None:
348        """Track beat availability; UNAVAILABLE drops the delivered schedule."""
349        self._beat_availability = availability
350        if availability is BeatAvailability.UNAVAILABLE:
351            self.clear_beats()
352
353    def on_connect(self) -> None:
354        """Subscribe to VisualizerGroupRole on attach."""
355        self._subscribe_to_group_role()
356
357    def on_disconnect(self) -> None:
358        """Unsubscribe from VisualizerGroupRole on detach."""
359        self._unsubscribe_from_group_role()
360
361    def on_stream_start(self) -> None:
362        """Build a fresh extractor for the new stream and notify the bridge."""
363        if self._support is not None:
364            req = self.get_audio_requirements()
365            self._extractor = VisualizerFeatureExtractor(
366                sample_rate=req.sample_rate,
367                channels=req.channels,
368                config=StreamStartVisualizer.from_support(self._support),
369            )
370        LOGGER.debug("BridgeVisualizerRole stream started for client %s", self._client.client_id)
371        if self._on_stream_start_cb:
372            self._on_stream_start_cb()
373
374    def on_audio_chunk(self, chunk: AudioChunk) -> None:
375        """Extract features from group audio and forward each frame."""
376        if self._extractor is None or self._on_frame_cb is None:
377            return
378        for frame in self._extractor.process_chunk(chunk.data, chunk.timestamp_us):
379            self._on_frame_cb(frame)
380
381    def on_stream_clear(self) -> None:
382        """Reset extractor state on seek/clear and notify the bridge."""
383        if self._extractor is not None:
384            self._extractor.reset()
385        if self._on_stream_clear_cb:
386            self._on_stream_clear_cb()
387
388    def on_stream_end(self) -> None:
389        """Tear down the extractor and notify the bridge."""
390        self._extractor = None
391        LOGGER.debug("BridgeVisualizerRole stream ended for client %s", self._client.client_id)
392        if self._on_stream_end_cb:
393            self._on_stream_end_cb()
394
395
396class BridgeColorRole(Role):
397    """
398    Custom Sendspin color role for in-process bridges.
399
400    Receives color palette state updates from ColorGroupRole and forwards
401    them to a callback instead of sending server/state over a WebSocket.
402    """
403
404    def __init__(self, client: SendspinClient) -> None:
405        """
406        Initialize the bridge color role.
407
408        :param client: The Sendspin client this role belongs to.
409        """
410        self._client = client
411        self._on_color_cb: Callable[[ServerStatePayload], None] | None = None
412
413    def set_callbacks(self, *, on_color: Callable[[ServerStatePayload], None]) -> None:
414        """
415        Wire up the color update callback after role creation.
416
417        :param on_color: Callback receiving server/state payloads with color updates.
418        """
419        self._on_color_cb = on_color
420
421    @property
422    def role_id(self) -> str:
423        """Return role identifier."""
424        return COLOR_BRIDGE_ROLE_ID
425
426    @property
427    def role_family(self) -> str:
428        """Return role family name."""
429        return "color"
430
431    def has_connection(self) -> bool:
432        """Return True to indicate bridge is "connected"."""
433        return True
434
435    def send_message(self, message: ServerMessage) -> None:
436        """Intercept outbound server/state color updates and forward to the bridge."""
437        if isinstance(message, ServerStateMessage) and self._on_color_cb:
438            self._on_color_cb(message.payload)
439
440    def on_connect(self) -> None:
441        """Subscribe to ColorGroupRole on attach."""
442        self._subscribe_to_group_role()
443
444    def on_disconnect(self) -> None:
445        """Unsubscribe from ColorGroupRole on detach."""
446        self._unsubscribe_from_group_role()
447
448
449register_role(BRIDGE_ROLE_ID, lambda client: BridgePlayerRole(client=client))
450register_role(VISUALIZER_BRIDGE_ROLE_ID, lambda client: BridgeVisualizerRole(client=client))
451register_role(COLOR_BRIDGE_ROLE_ID, lambda client: BridgeColorRole(client=client))
452