music-assistant-server

1"""
2Spotify Connect plugin for Music Assistant.
3
4We tie a single player to a single Spotify Connect daemon (go-librespot).
5The provider has multi instance support, so multiple players can be linked to
6multiple Spotify Connect daemons.
7
8go-librespot is driven entirely over its local HTTP+WebSocket API: the WebSocket
9``/events`` stream feeds player/session/metadata/volume state into Music Assistant,
10and transport + volume commands are issued via the REST endpoints. Playback control
11works without a configured Spotify *music* provider or the Spotify Web API.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import json
18import os
19import time
20from collections.abc import AsyncGenerator
21from contextlib import suppress
22from pathlib import Path
23from typing import TYPE_CHECKING, Any, cast
24
25from music_assistant_models.enums import (
26    ContentType,
27    MediaType,
28    PlaybackState,
29    ProviderFeature,
30    SourceControl,
31    StreamType,
32)
33from music_assistant_models.errors import AudioError, MediaNotFoundError
34from music_assistant_models.media_items import AudioFormat, AudioSource, ProviderMapping
35from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
36
37from music_assistant.constants import CONF_ENTRY_WARN_PREVIEW
38from music_assistant.helpers.process import AsyncProcess
39from music_assistant.helpers.util import (
40    interface_name_for_ip,
41    is_port_in_use,
42    select_free_port,
43)
44from music_assistant.models.plugin import PluginProvider
45
46from .client import GoLibrespotClient
47from .helpers import generate_device_id, get_go_librespot_binary
48
49if TYPE_CHECKING:
50    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
51    from music_assistant_models.provider import ProviderManifest
52
53    from music_assistant.mass import MusicAssistant
54    from music_assistant.models import ProviderInstanceType
55
56CONF_MASS_PLAYER_ID = "mass_player_id"
57CONF_PUBLISH_NAME = "publish_name"
58DEFAULT_PUBLISH_NAME = "Music Assistant"
59
60# Special value for auto player selection
61PLAYER_ID_AUTO = "__auto__"
62
63SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
64
65# stable id for the single AudioSource this provider exposes;
66# combined with the provider instance_id this forms the persistent uri
67AUDIO_SOURCE_ID = "main"
68
69# go-librespot volume scale; we pin volume_steps to this so the daemon's 0..max
70# volume maps 1:1 to a 0-100 percentage.
71VOLUME_STEPS = 100
72
73# Read size for pulling PCM off the daemon's stdout.
74STREAM_READ_CHUNK = 16384
75
76# When playback is paused the daemon stops writing PCM. If no PCM arrives for
77# this long while we're not in a 'playing' state, end the stream (clean EOF) so
78# the player leaves the playing state; the next 'playing' event re-streams.
79PAUSE_EOF_TIMEOUT_S = 0.5
80
81# Port range the go-librespot API server binds to (loopback only, one per instance).
82API_PORT_RANGE_START = 38800
83API_PORT_RANGE_END = 38900
84
85# Seconds to wait for the daemon to report 'playing' after a resume request.
86PLAYBACK_START_TIMEOUT_S = 3.0
87
88# Debounce before acting on an externally-triggered 'playing' event (see
89# _deferred_play_media_fire for why).
90PLAY_MEDIA_DEBOUNCE_S = 0.5
91
92# Ignore Spotify volume events for this long after a session becomes active, so
93# the player's own volume wins over librespot's initial value on (re)connect.
94INITIAL_VOLUME_GRACE_S = 3.0
95
96# User-facing message for the "not the active Spotify device" failure.
97# {0} is the Spotify Connect device's published name (see _not_active_error).
98NOT_ACTIVE_DEVICE_MESSAGE = (
99    "'{0}' is not the active Spotify playback device. "
100    "Open the Spotify app, select it as the playback device, and try again."
101)
102
103
104async def setup(
105    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
106) -> ProviderInstanceType:
107    """Initialize provider(instance) with given configuration."""
108    return SpotifyConnectProvider(mass, manifest, config)
109
110
111class SpotifyConnectProvider(PluginProvider):
112    """Implementation of a Spotify Connect Plugin (backed by go-librespot)."""
113
114    reload_on_streams_network_change = True
115
116    def __init__(
117        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
118    ) -> None:
119        """Initialize MusicProvider."""
120        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
121        # Configured default player (PLAYER_ID_AUTO or a specific player id)
122        self._default_player_id: str = (
123            cast("str", self.get_setup_value(CONF_MASS_PLAYER_ID)) or PLAYER_ID_AUTO
124        )
125        self._publish_name = (
126            cast("str", self.get_setup_value(CONF_PUBLISH_NAME)) or DEFAULT_PUBLISH_NAME
127        )
128        # Currently active player (the one currently playing or selected)
129        self._active_player_id: str | None = None
130        self.cache_dir = os.path.join(self.mass.cache_path, self.instance_id)
131        self._binary: str | None = None
132        self._api_port: int = 0
133        self._client: GoLibrespotClient | None = None
134        self._stop_called: bool = False
135        self._daemon_task: asyncio.Task[None] | None = None
136        self._events_task: asyncio.Task[None] | None = None
137        self._proc: AsyncProcess | None = None
138        self._restart_error_count = 0
139        self.logger.debug(
140            "Init plugin with name '%s' for player '%s' with instance id '%s'",
141            self.name,
142            self._default_player_id,
143            self.instance_id,
144        )
145        # _audio_format is the original Spotify source codec (Ogg Vorbis 320 kbps),
146        # advertised to clients for display. _decoded_audio_format is the raw PCM
147        # go-librespot actually writes to its stdout after decoding — what
148        # get_audio_stream yields and what the streams controller hands ffmpeg as
149        # the input format. We always emit the source's own format here; MA is
150        # responsible for converting it to whatever each player needs.
151        self._audio_format = AudioFormat(
152            content_type=ContentType.OGG,
153            codec_type=ContentType.VORBIS,
154            sample_rate=44100,
155            bit_depth=16,
156            channels=2,
157            bit_rate=320,
158        )
159        self._decoded_audio_format = AudioFormat(
160            content_type=ContentType.PCM_S16LE,
161            codec_type=ContentType.PCM_S16LE,
162            sample_rate=44100,
163            bit_depth=16,
164            channels=2,
165        )
166        self._stream_metadata = StreamMetadata(title=f"Spotify Connect | {self._publish_name}")
167        self._audio_source = self._build_audio_source()
168        # _in_use_by_queue is the queue currently streaming us. Claimed in
169        # on_source_selected (NOT in get_stream_details — that path also runs
170        # from queue preload, where claiming would block a later cross-queue
171        # handoff). Released in on_source_unselected when the session id
172        # matches, or in _clear_active_player on the daemon's 'inactive' event.
173        self._in_use_by_queue: str | None = None
174        # _active_session_id is the controller-provided token for the current
175        # stream request — used to reject stale on_source_unselected callbacks
176        # after a same-queue reconnect supersedes the previous request.
177        self._active_session_id: str | None = None
178        # tracks the daemon's play/pause state from its 'playing' / 'paused' /
179        # 'inactive' events; gates the resume kick in on_source_selected (skip if
180        # already playing) and the play_media trigger in the event handler.
181        self._playing: bool = False
182        # True while MA is the active Spotify Connect device (set on 'active',
183        # cleared on 'inactive'); gates get_stream_details and transport commands.
184        self._spotify_session_active: bool = False
185        # holds the single in-flight deferred play_media task scheduled from a
186        # 'playing' event; cancelled when a 'paused' / 'stopped' / 'active' event
187        # arrives during the debounce so we don't act on stale state from a dying
188        # session.
189        self._pending_play_media_task: asyncio.Task[None] | None = None
190        self._last_session_active_time: float = 0
191        self._last_volume_sent: int | None = None
192        # Last context/track URIs seen on the event stream. Used to take playback
193        # back (make ourselves the active Spotify device) when the user switched
194        # the active device away in the Spotify app and then presses play in MA.
195        self._last_context_uri: str | None = None
196        self._last_track_uri: str | None = None
197
198    @property
199    def instance_name_postfix(self) -> str | None:
200        """Return the advertised device name as the multi-instance postfix."""
201        return self._publish_name if self._publish_name != DEFAULT_PUBLISH_NAME else None
202
203    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
204        """Return runtime options for this provider."""
205        return (CONF_ENTRY_WARN_PREVIEW,)
206
207    async def handle_async_init(self) -> None:
208        """Handle async initialization of the provider."""
209        self._binary = get_go_librespot_binary()
210        self._api_port = await select_free_port(
211            API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1"
212        )
213        self._client = GoLibrespotClient(
214            self.mass, f"http://127.0.0.1:{self._api_port}", self.logger
215        )
216        # Two self-healing supervisors: one keeps the daemon process alive, the
217        # other keeps the events websocket connected (reconnecting across daemon
218        # restarts). The events runner resets the daemon's restart backoff once
219        # the websocket is healthy again.
220        self._daemon_task = self.mass.create_task(self._daemon_runner())
221        self._events_task = self.mass.create_task(self._events_runner())
222
223    async def unload(self, is_removed: bool = False) -> None:
224        """Handle close/cleanup of the provider."""
225        self._stop_called = True
226        self._cancel_pending_play_media()
227        for task in (self._events_task, self._daemon_task):
228            if task and not task.done():
229                task.cancel()
230                with suppress(asyncio.CancelledError):
231                    await task
232
233    @property
234    def active_player_id(self) -> str | None:
235        """Return the currently active player ID for this plugin."""
236        return self._active_player_id
237
238    async def get_audio_sources(self) -> list[AudioSource]:
239        """Return the AudioSources this plugin currently exposes."""
240        return [self._audio_source]
241
242    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
243        """
244        Return StreamDetails for streaming the Spotify Connect audio.
245
246        Side-effect-free: ownership is claimed in on_source_selected (which the
247        streams controller fires before this method on the actual stream
248        request). Keeping this idempotent means preload paths can fetch
249        streamdetails without claiming the source and blocking a cross-queue
250        handoff.
251
252        Raises AudioError when MA is not the active Spotify Connect device, since
253        playback can only be acquired while a Spotify session is connected to us
254        (entry must come from the Spotify app — see can_initiate below).
255        """
256        if item_id != AUDIO_SOURCE_ID:
257            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
258        # Only refuse when we can neither resume nor take playback back. If a last
259        # context is known we let the stream proceed; on_source_selected then takes
260        # playback back (makes us the active device) before audio is pulled.
261        if not self._playing and not self._spotify_session_active and not self._last_context_uri:
262            raise self._not_active_error()
263        # CUSTOM: the core pulls PCM from get_audio_stream. Reading the daemon's
264        # stdout means a consumer is always attached, so go-librespot's non-blocking
265        # pipe open succeeds, and it lets us end the stream cleanly when playback
266        # pauses so the player leaves the playing state. decoded_audio_format tells
267        # the core the PCM is s16le while audio_format keeps the Ogg/Vorbis source
268        # codec for display; MA resamples to each player's format as needed.
269        # `-fflags nobuffer` keeps ffmpeg's own input buffering low so the
270        # controller's realtime pacer owns the (small, bounded) read-ahead.
271        # expiration=0: never reuse a cached streamdetails so the active-device
272        # check above re-runs on every play attempt.
273        return StreamDetails(
274            provider=self.instance_id,
275            item_id=item_id,
276            audio_format=self._audio_format,
277            decoded_audio_format=self._decoded_audio_format,
278            media_type=MediaType.AUDIO_SOURCE,
279            stream_type=StreamType.CUSTOM,
280            stream_metadata=self._stream_metadata,
281            extra_input_args=["-fflags", "nobuffer"],
282            expiration=0,
283        )
284
285    async def get_audio_stream(
286        self,
287        streamdetails: StreamDetails,
288        seek_position: int = 0,
289    ) -> AsyncGenerator[bytes]:
290        """
291        Yield raw PCM from the go-librespot daemon's stdout for the live AudioSource.
292
293        When playback pauses the daemon stops writing PCM; we then end the stream
294        (clean EOF) so the consuming player leaves the playing state. The next
295        ``playing`` event re-triggers playback. ``seek_position`` is ignored —
296        seeking is handled upstream by Spotify, not by replaying the bytestream.
297        """
298        if streamdetails.item_id != AUDIO_SOURCE_ID:
299            raise MediaNotFoundError(f"Unknown AudioSource: {streamdetails.item_id}")
300        proc = self._proc
301        if proc is None:
302            raise AudioError("Spotify Connect daemon is not running")
303        # No pacing here: the streams controller's realtime pacer (ffmpeg readrate
304        # with a small initial burst) is the single pacing authority for live
305        # sources. Backpressure through the stdout pipe bounds how far the daemon
306        # (whose pipe backend is not realtime-paced) runs ahead, while the burst
307        # headroom absorbs scheduling jitter that would otherwise underrun the
308        # player. Pacing a second time here would pin the feed to exactly realtime
309        # and starve that headroom.
310        while True:
311            try:
312                chunk = await asyncio.wait_for(
313                    proc.read(STREAM_READ_CHUNK), timeout=PAUSE_EOF_TIMEOUT_S
314                )
315            except TimeoutError:
316                # No PCM for a while. If playback is no longer active (paused /
317                # stopped / session gone) end the stream so the player goes idle;
318                # a brief buffering gap while still playing just keeps waiting.
319                if not self._playing:
320                    return
321                continue
322            if not chunk:
323                return  # daemon stdout closed (process exited / restarting)
324            yield chunk
325
326    async def on_source_selected(
327        self,
328        source_id: str,
329        player_id: str,
330        queue_id: str,
331        stream_session_id: str,
332    ) -> None:
333        """Handle callback when this AudioSource has been selected/started on a player."""
334        if source_id != AUDIO_SOURCE_ID or not player_id:
335            return
336
337        # Cache the queue_id (== user-facing MA player) rather than the
338        # protocol-level player_id. Some protocol players are ephemeral bridges
339        # whose ID is invalid for play_media / queue lookups once torn down.
340        active_player_id = queue_id
341
342        # If there's already a different active player, kick it out. The claim
343        # below replaces the previous queue's claim; the prior stream's
344        # on_source_unselected may fire later, but its session-id guard keeps it
345        # from clobbering the new claim.
346        if self._active_player_id and self._active_player_id != active_player_id:
347            prev_player_id = self._active_player_id
348            self.logger.info(
349                "Source selected on player %s, stopping playback on %s",
350                active_player_id,
351                prev_player_id,
352            )
353            try:
354                await self.mass.players.cmd_stop(prev_player_id)
355            except Exception as err:
356                self.logger.debug("Failed to stop previous player %s: %s", prev_player_id, err)
357
358        # Claim ownership for this queue.
359        self._in_use_by_queue = queue_id
360        self._active_session_id = stream_session_id
361        self._active_player_id = active_player_id
362        self.logger.debug("Active player set to: %s", active_player_id)
363
364        # Only persist the selected player as the new default if not in auto mode
365        if self._default_player_id != PLAYER_ID_AUTO:
366            self._save_last_player_id(active_player_id)
367
368        # Externally triggered: the daemon is already playing → nothing to do.
369        # Otherwise acquire playback, then confirm it actually started.
370        if not self._playing:
371            assert self._client is not None
372            try:
373                if self._spotify_session_active:
374                    # Still the active Spotify device (just paused) → resume.
375                    await self._client.resume()
376                elif self._last_context_uri:
377                    # The user moved the active device away in the Spotify app.
378                    # Take playback back by (re)starting the last context on us,
379                    # which makes go-librespot the active device again. The track
380                    # restarts from its beginning (go-librespot has no
381                    # resume-at-position play call).
382                    self.logger.info("Taking Spotify playback back to Music Assistant")
383                    await self._client.play(
384                        self._last_context_uri, skip_to_uri=self._last_track_uri
385                    )
386                else:
387                    raise self._not_active_error()
388            except AudioError:
389                raise
390            except Exception as err:
391                raise AudioError(f"Failed to acquire Spotify Connect: {err}") from err
392            if not await self._wait_for_playing():
393                raise self._not_active_error()
394
395        # go-librespot reports 100% volume until told otherwise (with
396        # external_volume it ignores initial_volume); push the player's volume
397        # so the Spotify app's absolute volume commands start from the real level.
398        await self._sync_player_volume_to_spotify(active_player_id)
399
400    async def on_source_unselected(
401        self, source_id: str, queue_id: str, stream_session_id: str
402    ) -> None:
403        """Release the queue-scoped exclusive claim when MA tears down the stream."""
404        if source_id != AUDIO_SOURCE_ID:
405            return
406        # Reject stale callbacks: only release if this is still the active
407        # session. A queue_id check alone is not sufficient — same-queue
408        # reconnects would otherwise let an old request's late callback clear
409        # the live claim of the new stream.
410        if self._active_session_id != stream_session_id:
411            return
412        self._active_session_id = None
413        if self._in_use_by_queue == queue_id:
414            self._in_use_by_queue = None
415
416    async def on_source_control(
417        self,
418        source_id: str,
419        action: SourceControl,
420        value: int | None = None,
421    ) -> None:
422        """Proxy playback control commands to go-librespot's REST API."""
423        if source_id != AUDIO_SOURCE_ID:
424            return
425        if not self._playing and not self._spotify_session_active:
426            raise self._not_active_error()
427        assert self._client is not None
428        try:
429            if action == SourceControl.PLAY:
430                await self._client.resume()
431            elif action == SourceControl.PAUSE:
432                await self._client.pause()
433            elif action == SourceControl.NEXT:
434                await self._client.next()
435            elif action == SourceControl.PREVIOUS:
436                await self._client.prev()
437            elif action == SourceControl.SEEK and value is not None:
438                await self._client.seek(value * 1000)
439        except Exception as err:
440            self.logger.warning("Failed to send %s command to go-librespot: %s", action, err)
441            raise
442
443    async def on_volume_change(self, source_id: str, volume: int) -> None:
444        """Sync the Spotify app's volume slider with the player's new volume."""
445        if source_id != AUDIO_SOURCE_ID:
446            return
447        if not self._playing and not self._spotify_session_active:
448            raise self._not_active_error()
449        # Prevent ping-pong: only push if the value actually changed from what we
450        # last sent to / received from the daemon.
451        if self._last_volume_sent == volume:
452            return
453        try:
454            await self._push_volume_to_daemon(volume)
455        except Exception as err:
456            self.logger.warning("Failed to send volume command to go-librespot: %s", err)
457            raise
458
459    def _not_active_error(self) -> AudioError:
460        """Build the localized 'not the active Spotify device' error, naming this device."""
461        return AudioError(
462            NOT_ACTIVE_DEVICE_MESSAGE.format(self._publish_name),
463            translation_key="not_active_device",
464            translation_args=[self._publish_name],
465            translation_owner=self.translation_owner,
466        )
467
468    def _build_audio_source(self) -> AudioSource:
469        """
470        Construct the AudioSource MediaItem.
471
472        go-librespot exposes a full REST control API, so play / pause / seek /
473        next / previous are always available while a session is active — the
474        capability flags are static (no dependency on the Spotify Web API).
475        """
476        return AudioSource(
477            item_id=AUDIO_SOURCE_ID,
478            provider=self.instance_id,
479            name=self.name,
480            provider_mappings={
481                ProviderMapping(
482                    item_id=AUDIO_SOURCE_ID,
483                    provider_domain=self.domain,
484                    provider_instance=self.instance_id,
485                    audio_format=self._audio_format,
486                )
487            },
488            can_play_pause=True,
489            can_seek=True,
490            can_next_previous=True,
491            exclusive=True,
492            allow_external_trigger=True,
493            # Cold-start from MA is unreliable (Spotify needs an existing
494            # playback context), so only allow external entry via the Spotify app.
495            can_initiate=False,
496        )
497
498    def _get_target_player_id(self) -> str | None:
499        """
500        Determine the target player ID for playback.
501
502        Priority: an explicitly selected player; else (auto) a currently playing
503        player then the first available; else the configured default player.
504
505        :return: The player ID to use for playback, or None if none available.
506        """
507        if self._active_player_id:
508            if self.mass.players.get_player(self._active_player_id):
509                return self._active_player_id
510            self._active_player_id = None
511
512        if self._default_player_id == PLAYER_ID_AUTO:
513            all_players = list(self.mass.players.all_players(False, False))
514            for player in all_players:
515                if player.state.playback_state == PlaybackState.PLAYING:
516                    self.logger.debug("Auto-selecting playing player: %s", player.display_name)
517                    return player.player_id
518            if all_players:
519                first_player = all_players[0]
520                self.logger.debug(
521                    "Auto-selecting first available player: %s", first_player.display_name
522                )
523                return first_player.player_id
524            return None
525
526        if self.mass.players.get_player(self._default_player_id):
527            return self._default_player_id
528        self.logger.warning(
529            "Configured default player '%s' no longer exists", self._default_player_id
530        )
531        return None
532
533    async def _wait_for_playing(self, timeout: float = PLAYBACK_START_TIMEOUT_S) -> bool:
534        """
535        Wait up to ``timeout`` seconds for the daemon to report it is playing.
536
537        :param timeout: Maximum seconds to wait.
538        :return: True once playback is confirmed, False if the timeout elapses.
539        """
540        deadline = self.mass.loop.time() + timeout
541        while True:
542            if self._playing:
543                return True
544            if self.mass.loop.time() >= deadline:
545                return False
546            await asyncio.sleep(0.1)
547
548    def _cancel_pending_play_media(self) -> None:
549        """Cancel any pending deferred play_media trigger."""
550        task = self._pending_play_media_task
551        if task is not None and not task.done():
552            task.cancel()
553        self._pending_play_media_task = None
554
555    async def _deferred_play_media_fire(self) -> None:
556        """
557        Trigger play_media after a short debounce.
558
559        The daemon can emit a stale 'playing' from a dying session just before it
560        reconnects; acting on it immediately would start a stream for a session
561        that is about to be replaced. Debouncing — and cancelling the task on a
562        later 'paused' / 'stopped' / 'active' event — avoids a play→stop→replay loop.
563        """
564        try:
565            await asyncio.sleep(PLAY_MEDIA_DEBOUNCE_S)
566        except asyncio.CancelledError:
567            return
568        if not self._playing or self._in_use_by_queue:
569            return
570        target_player_id = self._get_target_player_id()
571        if not target_player_id:
572            self.logger.warning(
573                "Spotify Connect playback started but no player available. "
574                "Select this source on a player to start playback."
575            )
576            return
577        self.logger.info(
578            "Starting Spotify Connect playback [%s] on player %s",
579            self.instance_id,
580            target_player_id,
581        )
582        self._active_player_id = target_player_id
583        self.mass.create_task(
584            self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
585        )
586
587    def _clear_active_player(self) -> None:
588        """Clear the active player and reset playback state when a session ends."""
589        prev_player_id = self._active_player_id
590        self._active_player_id = None
591        self._in_use_by_queue = None
592        self._active_session_id = None
593        self._playing = False
594        if prev_player_id:
595            self.logger.debug("Playback ended on player %s, clearing active player", prev_player_id)
596            self.mass.players.trigger_player_update(prev_player_id)
597
598    def _save_last_player_id(self, player_id: str) -> None:
599        """Persist the selected player ID as the new default."""
600        if self._default_player_id == player_id:
601            return
602        try:
603            self._update_setup_data(CONF_MASS_PLAYER_ID, player_id)
604            self._default_player_id = player_id
605        except Exception as err:
606            self.logger.debug("Failed to persist player ID: %s", err)
607
608    def _write_config(self, source_ip: str | None) -> None:
609        """
610        Write the go-librespot ``config.yml`` for this instance.
611
612        go-librespot reads a YAML config; JSON is valid YAML, so we emit JSON to
613        sidestep an extra dependency and any string-quoting pitfalls (the device
614        name is user-provided). The config dir doubles as the credential/device
615        cache so the Spotify Connect device stays paired across restarts.
616
617        :param source_ip: Local address of the player-facing interface, or None to
618            advertise the Spotify Connect device on all interfaces.
619        """
620        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
621        config: dict[str, Any] = {
622            "device_name": self._publish_name,
623            "device_type": "speaker",
624            "device_id": generate_device_id(self.instance_id),
625            "bitrate": 320,
626            "audio_backend": "pipe",
627            # write decoded PCM to the daemon's stdout, which we capture and
628            # forward (the process pipe is always attached, so the daemon's
629            # non-blocking pipe open never fails for lack of a reader). s16le is
630            # the Spotify source representation; MA converts it per player.
631            "audio_output_pipe": "/dev/stdout",
632            "audio_output_pipe_format": "s16le",
633            # external_volume: don't let go-librespot attenuate the PCM — MA / the
634            # target player owns the actual volume. We still receive 'volume'
635            # events and push volume back so the Spotify app slider stays in sync.
636            # No initial_volume: go-librespot ignores it with external_volume set;
637            # _sync_player_volume_to_spotify pushes the player's volume instead.
638            "external_volume": True,
639            "volume_steps": VOLUME_STEPS,
640            "zeroconf_enabled": True,
641            "credentials": {"type": "zeroconf", "zeroconf": {"persist_credentials": True}},
642            "server": {"enabled": True, "address": "127.0.0.1", "port": self._api_port},
643        }
644        # Advertise the Spotify Connect device only on the interface the streams
645        # server binds to, so it lands on the right network on multi-homed hosts.
646        # go-librespot selects advertise interfaces by name, so map the IP to one.
647        if source_ip:
648            if iface_name := interface_name_for_ip(source_ip):
649                config["zeroconf_interfaces_to_advertise"] = [iface_name]
650            else:
651                self.logger.debug(
652                    "No interface found for stream bind IP %s; advertising on all interfaces",
653                    source_ip,
654                )
655        config_file = os.path.join(self.cache_dir, "config.yml")
656        with open(config_file, "w", encoding="utf-8") as fileobj:
657            json.dump(config, fileobj, indent=2)
658
659    async def _daemon_runner(self) -> None:
660        """Run and supervise the go-librespot daemon, restarting it if it exits."""
661        assert self._binary
662        assert self._client
663        # Loop forever; unload() cancels this task and the explicit stop-check below
664        # handles a graceful exit without a restart.
665        while True:
666            # If the API port was taken while the daemon was down, move to a
667            # fresh port instead of crash-looping on a bind error.
668            if await is_port_in_use(self._api_port, host="127.0.0.1"):
669                self._api_port = await select_free_port(
670                    API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1"
671                )
672                self._client.base_url = f"http://127.0.0.1:{self._api_port}"
673                self.logger.warning(
674                    "API port in use by another process; switching to port %s", self._api_port
675                )
676            self._write_config(await self.mass.streams.get_source_ip())
677            proc: AsyncProcess | None = None
678            try:
679                # stdout carries the decoded PCM (audio_output_pipe=/dev/stdout) and
680                # is consumed by get_audio_stream; stderr carries the daemon's logs,
681                # read here. Because the process pipe owns stdout from spawn there is
682                # always a reader, so go-librespot's non-blocking pipe open never
683                # fails for lack of a consumer.
684                self._proc = proc = AsyncProcess(
685                    [self._binary, "--config_dir", self.cache_dir],
686                    stdout=True,
687                    stderr=True,
688                    name=f"go-librespot[{self.name}]",
689                )
690                await proc.start()
691                self.logger.info("Started Spotify Connect background daemon [%s]", self.name)
692                async for line in proc.iter_stderr():
693                    self.logger.debug("[%s] %s", self.name, line)
694            except asyncio.CancelledError:
695                raise
696            except Exception as err:
697                self.logger.warning("go-librespot daemon error [%s]: %s", self.name, err)
698            finally:
699                if proc:
700                    await proc.close()
701                # The daemon — and thus the Spotify session — is gone. Reset session
702                # state so a dead/restarting daemon isn't treated as active and
703                # controllable; a fresh 'active' event re-establishes it on reconnect.
704                self._proc = None
705                self._playing = False
706                self._spotify_session_active = False
707            if self._stop_called:
708                break
709            self.logger.info("Spotify Connect background daemon stopped for %s", self.name)
710            self._restart_error_count += 1
711            if self._restart_error_count >= 5:
712                self.unload_with_error("go-librespot daemon failed to start multiple times.")
713                return
714            await asyncio.sleep(2)
715
716    async def _events_runner(self) -> None:
717        """Keep the go-librespot events websocket connected, reconnecting as needed."""
718        assert self._client is not None
719        while not self._stop_called:
720            try:
721                if not await self._client.wait_until_ready():
722                    await asyncio.sleep(2)
723                    continue
724                # A live websocket means the daemon is healthy: reset the restart
725                # backoff counter the daemon supervisor uses.
726                self._restart_error_count = 0
727                await self._client.listen_events(self._handle_event)
728            except asyncio.CancelledError:
729                raise
730            except Exception as err:
731                self.logger.debug("go-librespot events websocket dropped: %s", err)
732            if not self._stop_called:
733                await asyncio.sleep(2)
734
735    async def _handle_event(self, event_type: str, data: dict[str, Any]) -> None:
736        """Dispatch a single go-librespot websocket event."""
737        self.logger.debug("Received event [%s]: %s %s", self.name, event_type, data)
738        # Remember the latest context/track so we can take playback back if the
739        # user moves the active device away in the Spotify app (see on_source_selected).
740        if context_uri := data.get("context_uri"):
741            self._last_context_uri = context_uri
742        if track_uri := data.get("uri"):
743            self._last_track_uri = track_uri
744
745        if event_type == "active":
746            self._spotify_session_active = True
747            self._last_session_active_time = time.time()
748            # A (re)activation supersedes any deferred play_media scheduled from a
749            # previous session's stale 'playing'; the fresh 'playing' that follows
750            # schedules a new one.
751            self._cancel_pending_play_media()
752            self.logger.info("Spotify Connect session active for %s", self.name)
753            # A new session starts at the daemon's 100% volume default; push the
754            # target player's volume so the Spotify app's slider is correct from
755            # device selection, before any playback starts.
756            if player_id := self._get_target_player_id():
757                await self._sync_player_volume_to_spotify(player_id)
758        elif event_type == "inactive":
759            self.logger.info("Spotify Connect session inactive for %s", self.name)
760            self._spotify_session_active = False
761            prev_player_id = self._active_player_id
762            self._clear_active_player()
763            if prev_player_id:
764                self.mass.create_task(self.mass.players.cmd_stop(prev_player_id))
765            return
766        elif event_type == "playing":
767            self._playing = True
768            # Externally triggered playback: kick a play_media on the target MA
769            # player so the audio reaches a speaker. Deferred so a rapid
770            # playing/active burst from a reconnecting session can cancel it.
771            if not self._in_use_by_queue and (
772                self._pending_play_media_task is None or self._pending_play_media_task.done()
773            ):
774                self._pending_play_media_task = self.mass.create_task(
775                    self._deferred_play_media_fire()
776                )
777        elif event_type in ("paused", "stopped"):
778            self._playing = False
779            # A pause/stop is the definitive "don't start": cancel a deferred fire
780            # from a now-stale 'playing'. The active get_audio_stream sees the PCM
781            # stop and ends the stream (clean EOF), so the player leaves the playing
782            # state; the next 'playing' event re-fires play_media to resume.
783            self._cancel_pending_play_media()
784
785        self._apply_metadata(event_type, data)
786
787        if event_type == "volume":
788            await self._handle_volume_event(data)
789
790        # push metadata update to the active queue item's streamdetails
791        if self._in_use_by_queue:
792            self.mass.streams.update_stream_metadata(
793                self._in_use_by_queue,
794                AUDIO_SOURCE_ID,
795                self.instance_id,
796                self._stream_metadata,
797            )
798
799    def _apply_metadata(self, event_type: str, data: dict[str, Any]) -> None:
800        """Update the live StreamMetadata from a 'metadata' or 'seek' event."""
801        if event_type == "metadata":
802            self._stream_metadata.uri = data.get("uri")
803            if name := data.get("name"):
804                self._stream_metadata.title = name
805            artists = data.get("artist_names") or []
806            self._stream_metadata.artist = artists[0] if artists else None
807            self._stream_metadata.album = data.get("album_name")
808            self._stream_metadata.image_url = data.get("album_cover_url")
809            self._stream_metadata.description = None
810            duration_ms = data.get("duration")
811            self._stream_metadata.duration = duration_ms // 1000 if duration_ms else None
812            self._stream_metadata.elapsed_time = int(data.get("position", 0)) // 1000
813            self._stream_metadata.elapsed_time_last_updated = int(time.time())
814        elif event_type == "seek":
815            self._stream_metadata.elapsed_time = int(data.get("position", 0)) // 1000
816            self._stream_metadata.elapsed_time_last_updated = int(time.time())
817
818    async def _handle_volume_event(self, data: dict[str, Any]) -> None:
819        """Apply a Spotify-side volume change to the linked MA player."""
820        value = data.get("value")
821        max_value = data.get("max") or VOLUME_STEPS
822        if value is None:
823            return
824        volume = int(int(value) / int(max_value) * 100)
825        # Ignore our own echo: go-librespot emits a 'volume' event for the value we
826        # just pushed in on_volume_change; re-applying it would ping-pong.
827        if volume == self._last_volume_sent:
828            return
829        # Ignore the volume that librespot reports right after a session becomes
830        # active — the player's own volume should win in that window.
831        if time.time() - self._last_session_active_time < INITIAL_VOLUME_GRACE_S:
832            self.logger.debug("Ignoring initial volume_changed event after session active")
833            return
834        if not self._in_use_by_queue:
835            return
836        previous_volume = self._last_volume_sent
837        self._last_volume_sent = volume
838        try:
839            await self.mass.players.cmd_volume_set(self._in_use_by_queue, volume)
840        except Exception as err:
841            # Volume sync is best-effort: the player may not support volume, or the
842            # command may fail. Restore the cached value so a retry isn't wrongly
843            # deduped, and never let it bubble up and drop the events loop.
844            self._last_volume_sent = previous_volume
845            self.logger.debug("Could not set volume on %s: %s", self._in_use_by_queue, err)
846
847    async def _sync_player_volume_to_spotify(self, player_id: str) -> None:
848        """
849        Push a player's current volume to go-librespot (best-effort).
850
851        :param player_id: The MA player whose volume to push.
852        """
853        player = self.mass.players.get_player(player_id)
854        if player is None or player.state.volume_level is None:
855            return
856        # clamp: the logical volume can be out of range until volume limit
857        # enforcement runs
858        volume = max(0, min(100, player.state.volume_level))
859        # No dedupe against _last_volume_sent here: it holds the last value
860        # exchanged with the daemon, not the daemon's current volume, which
861        # resets to its 100% default on a new session or daemon restart.
862        try:
863            await self._push_volume_to_daemon(volume)
864        except Exception as err:
865            self.logger.debug("Failed to sync player volume to Spotify: %s", err)
866
867    async def _push_volume_to_daemon(self, volume: int) -> None:
868        """
869        Send an absolute 0-100 volume to go-librespot.
870
871        :param volume: Volume percentage to send.
872        :raises Exception: If the request to the daemon fails.
873        """
874        assert self._client is not None
875        previous_volume = self._last_volume_sent
876        # Record BEFORE the call: go-librespot echoes a 'volume' event back, and
877        # that echo can arrive over the WS while we're still awaiting set_volume.
878        # Recording up front lets _handle_volume_event dedupe it instead of
879        # bouncing it back as a player volume change.
880        self._last_volume_sent = volume
881        try:
882            await self._client.set_volume(round(volume / 100 * VOLUME_STEPS))
883        except Exception:
884            # restore on failure so a retry of this value isn't wrongly deduped
885            self._last_volume_sent = previous_volume
886            raise
887