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        """
179        Set volume and notify bridge.
180
181        The level is on the scale the role reports, and is stored, announced to
182        the client and handed to the bridge unchanged.
183        """
184        self._volume = volume
185        self._emit_volume_changed()
186        if self._on_volume_change_cb:
187            self._on_volume_change_cb(volume)
188
189    def set_player_mute(self, muted: bool) -> None:
190        """Set mute state and notify bridge."""
191        self._muted = muted
192        self._emit_volume_changed()
193        if self._on_mute_change_cb:
194            self._on_mute_change_cb(muted)
195
196    def update_player_state(self, *, volume: int | None = None, muted: bool | None = None) -> None:
197        """
198        Adopt volume/mute state the bridged player is already in.
199
200        Use this instead of set_player_volume/set_player_mute for state that came
201        from the player's own side (device feedback, a mute applied elsewhere): the
202        bridge callbacks are not invoked, so the value is not pushed back at the
203        player it was just read from.
204
205        :param volume: Volume level (0-100) the player is at, or None to leave it.
206        :param muted: Mute state the player is in, or None to leave it.
207        """
208        changed = False
209        if volume is not None and volume != self._volume:
210            self._volume = volume
211            changed = True
212        if muted is not None and muted != self._muted:
213            self._muted = muted
214            changed = True
215        if changed:
216            self._emit_volume_changed()
217
218    def on_audio_chunk(self, chunk: AudioChunk) -> None:
219        """Receive audio chunk from PushStream and forward to callback."""
220        if self._on_audio_chunk_cb:
221            self._on_audio_chunk_cb(chunk)
222
223    def on_connect(self) -> None:
224        """Subscribe to PlayerGroupRole on attach."""
225        self._subscribe_to_group_role()
226
227    def on_disconnect(self) -> None:
228        """Unsubscribe from PlayerGroupRole on detach."""
229        self._unsubscribe_from_group_role()
230
231    def has_connection(self) -> bool:
232        """Return True to indicate bridge is "connected" for audio purposes."""
233        return True
234
235    def supports_preconnect_audio(self) -> bool:
236        """Return True -- bridge can receive audio before the stream starts."""
237        return True
238
239    def on_stream_start(self) -> None:
240        """Log stream start and invoke callback."""
241        LOGGER.debug("BridgePlayerRole stream started for client %s", self._client.client_id)
242        if self._on_stream_start_cb:
243            self._on_stream_start_cb()
244
245    def on_stream_end(self) -> None:
246        """Log stream end and invoke the stream-end callback."""
247        LOGGER.debug("BridgePlayerRole stream ended for client %s", self._client.client_id)
248        if self._on_stream_end_cb:
249            self._on_stream_end_cb()
250
251    def _emit_volume_changed(self) -> None:
252        """Emit VolumeChangedEvent so the SendspinPlayer stays in sync."""
253        self.emit_client_event(VolumeChangedEvent(volume=self._volume, muted=self._muted))
254
255
256class BridgeVisualizerRole(Role):
257    """
258    Custom Sendspin visualizer role for in-process bridges.
259
260    Runs the server's visualizer feature extraction on the group's audio and
261    forwards extracted frames, beat schedules, and stream lifecycle events to
262    callbacks instead of sending binary frames over a WebSocket.
263
264    Created by the role factory registry. After creation, the bridge must call
265    set_callbacks() and setup_visualizer(), then attach the roles via the
266    client's attach_preinitialized_roles(): external clients never get a
267    transport attach, so the group-role subscription (which beat schedules
268    and lifecycle broadcasts flow through) has to be made explicitly.
269    """
270
271    def __init__(self, client: SendspinClient) -> None:
272        """
273        Initialize the bridge visualizer role.
274
275        :param client: The Sendspin client this role belongs to.
276        """
277        self._client = client
278        self._support: ClientHelloVisualizerSupport | None = None
279        self._extractor: VisualizerFeatureExtractor | None = None
280        self._beat_availability = BeatAvailability.PENDING
281        self._on_frame_cb: Callable[[ExtractedFrame], None] | None = None
282        self._on_beats_cb: Callable[[list[BeatTiming]], None] | None = None
283        self._on_beats_clear_cb: Callable[[], None] | None = None
284        self._on_stream_start_cb: Callable[[], None] | None = None
285        self._on_stream_clear_cb: Callable[[], None] | None = None
286        self._on_stream_end_cb: Callable[[], None] | None = None
287
288    def set_callbacks(
289        self,
290        *,
291        on_frame: Callable[[ExtractedFrame], None],
292        on_beats: Callable[[list[BeatTiming]], None],
293        on_beats_clear: Callable[[], None],
294        on_stream_start: Callable[[], None],
295        on_stream_clear: Callable[[], None],
296        on_stream_end: Callable[[], None],
297    ) -> None:
298        """
299        Wire up bridge callbacks after role creation.
300
301        :param on_frame: Callback receiving extracted visualizer frames.
302        :param on_beats: Callback receiving beat schedule segments.
303        :param on_beats_clear: Callback when the beat schedule is dropped.
304        :param on_stream_start: Callback when the stream starts.
305        :param on_stream_clear: Callback when buffered stream state is cleared (seek).
306        :param on_stream_end: Callback when the stream ends.
307        """
308        self._on_frame_cb = on_frame
309        self._on_beats_cb = on_beats
310        self._on_beats_clear_cb = on_beats_clear
311        self._on_stream_start_cb = on_stream_start
312        self._on_stream_clear_cb = on_stream_clear
313        self._on_stream_end_cb = on_stream_end
314
315    def setup_visualizer(self, support: ClientHelloVisualizerSupport) -> None:
316        """
317        Configure feature extraction from the registered hello's support object.
318
319        :param support: The visualizer support object from the client hello.
320        """
321        self._support = support
322
323    @property
324    def role_id(self) -> str:
325        """Return role identifier."""
326        return VISUALIZER_BRIDGE_ROLE_ID
327
328    @property
329    def role_family(self) -> str:
330        """Return role family name."""
331        return "visualizer"
332
333    def get_audio_requirements(self) -> AudioRequirements:
334        """Return audio requirements for visualizer analysis."""
335        return AudioRequirements(
336            sample_rate=48_000,
337            bit_depth=16,
338            channels=2,
339            frame_duration_us=25_000,
340        )
341
342    def has_connection(self) -> bool:
343        """Return True to indicate bridge is "connected" for audio purposes."""
344        return True
345
346    def supports_preconnect_audio(self) -> bool:
347        """Return True -- bridge receives audio without a transport attach."""
348        return True
349
350    def replay_from_pcm_cache(self) -> bool:
351        """Replay buffered PCM on late join (visualizer is analysis-only)."""
352        return True
353
354    @property
355    def wants_beats(self) -> bool:
356        """True if the bridge requested beats and beats are not unavailable."""
357        return (
358            self._support is not None
359            and "beat" in self._support.types
360            and self._beat_availability is not BeatAvailability.UNAVAILABLE
361        )
362
363    def append_beats(self, beats: list[BeatTiming]) -> None:
364        """Forward a beat schedule segment to the bridge."""
365        if self._on_beats_cb and beats:
366            self._on_beats_cb(list(beats))
367
368    def clear_beats(self) -> None:
369        """Forward a beat-schedule clear to the bridge."""
370        if self._on_beats_clear_cb:
371            self._on_beats_clear_cb()
372
373    def set_beat_availability(self, availability: BeatAvailability) -> None:
374        """Track beat availability; UNAVAILABLE drops the delivered schedule."""
375        self._beat_availability = availability
376        if availability is BeatAvailability.UNAVAILABLE:
377            self.clear_beats()
378
379    def on_connect(self) -> None:
380        """Subscribe to VisualizerGroupRole on attach."""
381        self._subscribe_to_group_role()
382
383    def on_disconnect(self) -> None:
384        """Unsubscribe from VisualizerGroupRole on detach."""
385        self._unsubscribe_from_group_role()
386
387    def on_stream_start(self) -> None:
388        """Build a fresh extractor for the new stream and notify the bridge."""
389        if self._support is not None:
390            req = self.get_audio_requirements()
391            self._extractor = VisualizerFeatureExtractor(
392                sample_rate=req.sample_rate,
393                channels=req.channels,
394                config=StreamStartVisualizer.from_support(self._support),
395            )
396        LOGGER.debug("BridgeVisualizerRole stream started for client %s", self._client.client_id)
397        if self._on_stream_start_cb:
398            self._on_stream_start_cb()
399
400    def on_audio_chunk(self, chunk: AudioChunk) -> None:
401        """Extract features from group audio and forward each frame."""
402        if self._extractor is None or self._on_frame_cb is None:
403            return
404        for frame in self._extractor.process_chunk(chunk.data, chunk.timestamp_us):
405            self._on_frame_cb(frame)
406
407    def on_stream_clear(self) -> None:
408        """Reset extractor state on seek/clear and notify the bridge."""
409        if self._extractor is not None:
410            self._extractor.reset()
411        if self._on_stream_clear_cb:
412            self._on_stream_clear_cb()
413
414    def on_stream_end(self) -> None:
415        """Tear down the extractor and notify the bridge."""
416        self._extractor = None
417        LOGGER.debug("BridgeVisualizerRole stream ended for client %s", self._client.client_id)
418        if self._on_stream_end_cb:
419            self._on_stream_end_cb()
420
421
422class BridgeColorRole(Role):
423    """
424    Custom Sendspin color role for in-process bridges.
425
426    Receives color palette state updates from ColorGroupRole and forwards
427    them to a callback instead of sending server/state over a WebSocket.
428    """
429
430    def __init__(self, client: SendspinClient) -> None:
431        """
432        Initialize the bridge color role.
433
434        :param client: The Sendspin client this role belongs to.
435        """
436        self._client = client
437        self._on_color_cb: Callable[[ServerStatePayload], None] | None = None
438
439    def set_callbacks(self, *, on_color: Callable[[ServerStatePayload], None]) -> None:
440        """
441        Wire up the color update callback after role creation.
442
443        :param on_color: Callback receiving server/state payloads with color updates.
444        """
445        self._on_color_cb = on_color
446
447    @property
448    def role_id(self) -> str:
449        """Return role identifier."""
450        return COLOR_BRIDGE_ROLE_ID
451
452    @property
453    def role_family(self) -> str:
454        """Return role family name."""
455        return "color"
456
457    def has_connection(self) -> bool:
458        """Return True to indicate bridge is "connected"."""
459        return True
460
461    def send_message(self, message: ServerMessage) -> None:
462        """Intercept outbound server/state color updates and forward to the bridge."""
463        if isinstance(message, ServerStateMessage) and self._on_color_cb:
464            self._on_color_cb(message.payload)
465
466    def on_connect(self) -> None:
467        """Subscribe to ColorGroupRole on attach."""
468        self._subscribe_to_group_role()
469
470    def on_disconnect(self) -> None:
471        """Unsubscribe from ColorGroupRole on detach."""
472        self._unsubscribe_from_group_role()
473
474
475register_role(BRIDGE_ROLE_ID, lambda client: BridgePlayerRole(client=client))
476register_role(VISUALIZER_BRIDGE_ROLE_ID, lambda client: BridgeVisualizerRole(client=client))
477register_role(COLOR_BRIDGE_ROLE_ID, lambda client: BridgeColorRole(client=client))
478