music-assistant-server

1"""
2MA-facing provider logic for the Spotify Connect plugin.
3
4The provider owns everything Music Assistant sees: the AudioSource, stream
5details, target-player selection, playback claims and volume policy. It is
6backend-agnostic: all Spotify specifics live behind the
7``SpotifyConnectBackend`` contract and reach the provider as normalized
8``BackendEvent``s.
9"""
10
11from __future__ import annotations
12
13import asyncio
14import time
15from typing import TYPE_CHECKING, Final, cast
16
17from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
18from music_assistant_models.enums import (
19    ConfigEntryType,
20    MediaType,
21    PlaybackState,
22    ProviderFeature,
23    SourceControl,
24)
25from music_assistant_models.errors import AudioError, LoginFailed, MediaNotFoundError
26from music_assistant_models.media_items import AudioSource, ProviderMapping
27from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
28
29from music_assistant.constants import CONF_CROSSFADE_DURATION, CONF_ENTRY_WARN_PREVIEW
30from music_assistant.models.plugin import PluginProvider
31
32from .go_librespot import GoLibrespotBackend
33from .models import BackendEventType
34from .soloist import VOLUME_MODE_PLAYER_ONLY, VOLUME_MODE_SYNC_SPOTIFY, SoloistBackend
35
36if TYPE_CHECKING:
37    from collections.abc import AsyncGenerator
38
39    from music_assistant_models.config_entries import ProviderConfig
40    from music_assistant_models.provider import ProviderManifest
41
42    from music_assistant.mass import MusicAssistant
43
44    from .base import SpotifyConnectBackend
45    from .models import BackendEvent, BackendTrackMetadata
46
47CONF_MASS_PLAYER_ID = "mass_player_id"
48CONF_PUBLISH_NAME = "publish_name"
49DEFAULT_PUBLISH_NAME = "Music Assistant"
50
51# Backend selection, collected by the setup flow (stored in setup_data).
52CONF_BACKEND = "backend"
53BACKEND_GO_LIBRESPOT = "go_librespot"
54BACKEND_SOLOIST = "soloist"
55
56# Soloist-specific values collected by the setup flow (see CONF_BACKEND).
57CONF_API_KEY = "soloist_api_key"
58CONF_SOLOIST_CONSENT = "soloist_download_consent"
59CONF_VOLUME_MODE = "volume_mode"
60
61# Playback behavior applied by the Spotify engine itself (both backends).
62CONF_LOUDNESS_NORMALIZATION = "loudness_normalization"
63MAX_CROSSFADE_DURATION = 12  # seconds, matching the Spotify apps' slider
64
65# The selectable volume modes (labels resolve from strings.json), shared
66# between the runtime option and the setup flow.
67VOLUME_MODE_OPTIONS: Final = [
68    ConfigValueOption(VOLUME_MODE_PLAYER_ONLY),
69    ConfigValueOption(VOLUME_MODE_SYNC_SPOTIFY),
70]
71
72# Special value for auto player selection
73PLAYER_ID_AUTO = "__auto__"
74
75SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
76
77# stable id for the single AudioSource this provider exposes;
78# combined with the provider instance_id this forms the persistent uri
79AUDIO_SOURCE_ID = "main"
80
81# When playback is paused the backend stops writing PCM. If no PCM arrives for
82# this long while we're not in a 'playing' state, end the stream (clean EOF) so
83# the player leaves the playing state; the next 'playing' event re-streams.
84PAUSE_EOF_TIMEOUT_S = 0.5
85
86# Seconds to wait for the backend to report 'playing' after a resume request.
87PLAYBACK_START_TIMEOUT_S = 3.0
88
89# Debounce before acting on an externally-triggered 'playing' event (see
90# _deferred_play_media_fire for why).
91PLAY_MEDIA_DEBOUNCE_S = 0.5
92
93# Ignore Spotify volume events for this long after a session becomes active, so
94# the player's own volume wins over the backend's initial value on (re)connect.
95INITIAL_VOLUME_GRACE_S = 3.0
96
97# User-facing message for the "not the active Spotify device" failure.
98# {0} is the Spotify Connect device's published name (see _not_active_error).
99NOT_ACTIVE_DEVICE_MESSAGE = (
100    "'{0}' is not the active Spotify playback device. "
101    "Open the Spotify app, select it as the playback device, and try again."
102)
103
104
105class SpotifyConnectProvider(PluginProvider):
106    """Implementation of a Spotify Connect Plugin (backed by a SpotifyConnectBackend)."""
107
108    reload_on_streams_network_change = True
109
110    def __init__(
111        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
112    ) -> None:
113        """Initialize MusicProvider."""
114        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
115        # Configured default player (PLAYER_ID_AUTO or a specific player id)
116        self._default_player_id: str = (
117            cast("str", self.get_setup_value(CONF_MASS_PLAYER_ID)) or PLAYER_ID_AUTO
118        )
119        self._publish_name = (
120            cast("str", self.get_setup_value(CONF_PUBLISH_NAME)) or DEFAULT_PUBLISH_NAME
121        )
122        # Currently active player (the one currently playing or selected)
123        self._active_player_id: str | None = None
124        self._backend: SpotifyConnectBackend = self._create_backend()
125        self.logger.debug(
126            "Init plugin with name '%s' for player '%s' with instance id '%s'",
127            self.name,
128            self._default_player_id,
129            self.instance_id,
130        )
131        self._stream_metadata = StreamMetadata(title=f"Spotify Connect | {self._publish_name}")
132        self._audio_source = self._build_audio_source()
133        # _in_use_by_queue is the queue currently streaming us. Claimed in
134        # on_source_selected (NOT in get_stream_details — that path also runs
135        # from queue preload, where claiming would block a later cross-queue
136        # handoff). Released in on_source_unselected when the session id
137        # matches, or in _clear_active_player on the backend's 'inactive' event.
138        self._in_use_by_queue: str | None = None
139        # _active_session_id is the controller-provided token for the current
140        # stream request — used to reject stale on_source_unselected callbacks
141        # after a same-queue reconnect supersedes the previous request.
142        self._active_session_id: str | None = None
143        # tracks the backend's play/pause state from its 'playing' / 'paused' /
144        # 'inactive' events; gates the resume kick in on_source_selected (skip if
145        # already playing) and the play_media trigger in the event handler.
146        self._playing: bool = False
147        # True while MA is the active Spotify Connect device (set on 'active',
148        # cleared on 'inactive'); gates get_stream_details and transport commands.
149        self._spotify_session_active: bool = False
150        # holds the single in-flight deferred play_media task scheduled from a
151        # 'playing' event; cancelled when a 'paused' / 'stopped' / 'active' event
152        # arrives during the debounce so we don't act on stale state from a dying
153        # session.
154        self._pending_play_media_task: asyncio.Task[None] | None = None
155        # holds the in-flight stop of a paused player (pipe-fed backends
156        # only); the stop dispatches right away, but a 'playing' event cancels
157        # it while it is still in flight (a slow player can hold it for up to
158        # 10s), so a resume is never killed by a stop landing late.
159        self._pending_pause_stop_task: asyncio.Task[None] | None = None
160        self._last_session_active_time: float = 0
161        self._last_volume_sent: int | None = None
162        # Last context/track URIs seen on the event stream. Used to take playback
163        # back (make ourselves the active Spotify device) when the user switched
164        # the active device away in the Spotify app and then presses play in MA.
165        self._last_context_uri: str | None = None
166        self._last_track_uri: str | None = None
167
168    @property
169    def instance_name_postfix(self) -> str | None:
170        """Return the advertised device name as the multi-instance postfix."""
171        return self._publish_name if self._publish_name != DEFAULT_PUBLISH_NAME else None
172
173    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
174        """Return runtime options for this provider."""
175        # The backend selection and the soloist secrets are managed by the setup
176        # flow (stored in setup_data) and stay hidden; the volume mode is a
177        # visible runtime option for soloist configs.
178        is_soloist = self.get_setup_value(CONF_BACKEND) == BACKEND_SOLOIST
179        return (
180            CONF_ENTRY_WARN_PREVIEW,
181            ConfigEntry(
182                key=CONF_BACKEND,
183                type=ConfigEntryType.STRING,
184                default_value=BACKEND_GO_LIBRESPOT,
185                required=False,
186                hidden=True,
187            ),
188            ConfigEntry(
189                key=CONF_API_KEY,
190                type=ConfigEntryType.SECURE_STRING,
191                required=False,
192                hidden=True,
193            ),
194            ConfigEntry(
195                key=CONF_SOLOIST_CONSENT,
196                type=ConfigEntryType.BOOLEAN,
197                default_value=False,
198                required=False,
199                hidden=True,
200            ),
201            ConfigEntry(
202                key=CONF_VOLUME_MODE,
203                type=ConfigEntryType.STRING,
204                default_value=VOLUME_MODE_PLAYER_ONLY,
205                required=False,
206                options=VOLUME_MODE_OPTIONS,
207                hidden=not is_soloist,
208            ),
209            ConfigEntry(
210                key=CONF_CROSSFADE_DURATION,
211                type=ConfigEntryType.INTEGER,
212                range=(0, MAX_CROSSFADE_DURATION),
213                default_value=0,
214                required=False,
215            ),
216            ConfigEntry(
217                key=CONF_LOUDNESS_NORMALIZATION,
218                type=ConfigEntryType.BOOLEAN,
219                default_value=True,
220                required=False,
221            ),
222        )
223
224    async def handle_async_init(self) -> None:
225        """Handle async initialization of the provider."""
226        await self._backend.start()
227
228    async def unload(self, is_removed: bool = False) -> None:
229        """Handle close/cleanup of the provider."""
230        self._cancel_pending_play_media()
231        self._cancel_pending_pause_stop()
232        await self._backend.stop()
233
234    @property
235    def active_player_id(self) -> str | None:
236        """Return the currently active player ID for this plugin."""
237        return self._active_player_id
238
239    async def get_audio_sources(self) -> list[AudioSource]:
240        """Return the AudioSources this plugin currently exposes."""
241        return [self._audio_source]
242
243    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
244        """
245        Return StreamDetails for streaming the Spotify Connect audio.
246
247        Side-effect-free: ownership is claimed in on_source_selected (which the
248        streams controller fires before this method on the actual stream
249        request). Keeping this idempotent means preload paths can fetch
250        streamdetails without claiming the source and blocking a cross-queue
251        handoff.
252
253        Raises AudioError when MA is not the active Spotify Connect device and
254        no previous playback context is known to resume — the user then has to
255        start playback from the Spotify app once.
256        """
257        if item_id != AUDIO_SOURCE_ID:
258            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
259        # Only refuse when we can neither resume nor take playback back. If a last
260        # context is known we let the stream proceed; on_source_selected then takes
261        # playback back (makes us the active device) before audio is pulled.
262        if not self._playing and not self._spotify_session_active and not self._last_context_uri:
263            raise self._not_active_error()
264        # The backend describes how its audio is consumed: CUSTOM (the core pulls
265        # PCM from get_audio_stream) or a named pipe read directly by ffmpeg.
266        # decoded_audio_format tells the core the PCM format while audio_format
267        # keeps the source codec for display; MA resamples to each player's
268        # format as needed.
269        # expiration=0: never reuse a cached streamdetails so the active-device
270        # check above re-runs on every play attempt.
271        stream_source = await self._backend.get_stream_source()
272        return StreamDetails(
273            provider=self.instance_id,
274            item_id=item_id,
275            audio_format=self._backend.audio_format,
276            decoded_audio_format=self._backend.decoded_audio_format,
277            media_type=MediaType.AUDIO_SOURCE,
278            stream_type=stream_source.stream_type,
279            path=stream_source.path,
280            stream_metadata=self._stream_metadata,
281            extra_input_args=stream_source.extra_input_args,
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 backend's audio pipe for the live AudioSource.
292
293        Only used for backends with a CUSTOM stream source (NAMED_PIPE backends
294        are read directly by the streams controller). When playback pauses the
295        backend stops writing PCM; we then end the stream (clean EOF) so the
296        consuming player leaves the playing state. The next ``playing`` event
297        re-triggers playback.
298
299        :param streamdetails: The StreamDetails of the AudioSource being streamed.
300        :param seek_position: Ignored — seeking is handled upstream by Spotify,
301            not by replaying the bytestream.
302        """
303        if streamdetails.item_id != AUDIO_SOURCE_ID:
304            raise MediaNotFoundError(f"Unknown AudioSource: {streamdetails.item_id}")
305        read_chunk = self._backend.get_audio_reader()
306        if read_chunk is None:
307            raise AudioError("Spotify Connect daemon is not running")
308        # No pacing here: the streams controller's realtime pacer (ffmpeg readrate
309        # with a small initial burst) is the single pacing authority for live
310        # sources. Backpressure through the audio pipe bounds how far the backend
311        # (whose pipe backend is not realtime-paced) runs ahead, while the burst
312        # headroom absorbs scheduling jitter that would otherwise underrun the
313        # player. Pacing a second time here would pin the feed to exactly realtime
314        # and starve that headroom.
315        while True:
316            try:
317                chunk = await asyncio.wait_for(read_chunk(), timeout=PAUSE_EOF_TIMEOUT_S)
318            except TimeoutError:
319                # No PCM for a while. If playback is no longer active (paused /
320                # stopped / session gone) end the stream so the player goes idle;
321                # a brief buffering gap while still playing just keeps waiting.
322                if not self._playing:
323                    return
324                continue
325            if not chunk:
326                return  # audio pipe closed (backend exited / restarting)
327            yield chunk
328
329    async def on_source_selected(
330        self,
331        source_id: str,
332        player_id: str,
333        queue_id: str,
334        stream_session_id: str,
335    ) -> None:
336        """Handle callback when this AudioSource has been selected/started on a player."""
337        if source_id != AUDIO_SOURCE_ID or not player_id:
338            return
339
340        # Cache the queue_id (== user-facing MA player) rather than the
341        # protocol-level player_id. Some protocol players are ephemeral bridges
342        # whose ID is invalid for play_media / queue lookups once torn down.
343        active_player_id = queue_id
344
345        # If there's already a different active player, kick it out. The claim
346        # below replaces the previous queue's claim; the prior stream's
347        # on_source_unselected may fire later, but its session-id guard keeps it
348        # from clobbering the new claim.
349        if self._active_player_id and self._active_player_id != active_player_id:
350            prev_player_id = self._active_player_id
351            self.logger.info(
352                "Source selected on player %s, stopping playback on %s",
353                active_player_id,
354                prev_player_id,
355            )
356            try:
357                await self.mass.players.cmd_stop(prev_player_id)
358            except Exception as err:
359                self.logger.debug("Failed to stop previous player %s: %s", prev_player_id, err)
360
361        # Claim ownership for this queue.
362        self._in_use_by_queue = queue_id
363        self._active_session_id = stream_session_id
364        self._active_player_id = active_player_id
365        self.logger.debug("Active player set to: %s", active_player_id)
366
367        # Only persist the selected player as the new default if not in auto mode
368        if self._default_player_id != PLAYER_ID_AUTO:
369            self._save_last_player_id(active_player_id)
370
371        # Externally triggered: the backend is already playing → nothing to do.
372        # Otherwise acquire playback, then confirm it actually started.
373        if not self._playing:
374            try:
375                if self._spotify_session_active:
376                    # Still the active Spotify device (just paused) → resume.
377                    await self._backend.resume()
378                elif self._last_context_uri:
379                    # The user moved the active device away in the Spotify app.
380                    # Take playback back by (re)starting the last context on us,
381                    # which makes this device the active one again. The track
382                    # restarts from its beginning (there is no resume-at-position
383                    # play call).
384                    self.logger.info("Taking Spotify playback back to Music Assistant")
385                    await self._backend.play(
386                        self._last_context_uri, skip_to_uri=self._last_track_uri
387                    )
388                else:
389                    raise self._not_active_error()
390            except AudioError:
391                raise
392            except Exception as err:
393                raise AudioError(f"Failed to acquire Spotify Connect: {err}") from err
394            if not await self._wait_for_playing():
395                raise self._not_active_error()
396
397        # The backend reports 100% volume until told otherwise; push the player's
398        # volume so the Spotify app's absolute volume commands start from the
399        # real level.
400        await self._sync_player_volume_to_spotify(active_player_id)
401
402    async def on_source_unselected(
403        self, source_id: str, queue_id: str, stream_session_id: str
404    ) -> None:
405        """Release the queue-scoped exclusive claim when MA tears down the stream."""
406        if source_id != AUDIO_SOURCE_ID:
407            return
408        # Reject stale callbacks: only release if this is still the active
409        # session. A queue_id check alone is not sufficient — same-queue
410        # reconnects would otherwise let an old request's late callback clear
411        # the live claim of the new stream.
412        if self._active_session_id != stream_session_id:
413            return
414        self._active_session_id = None
415        if self._in_use_by_queue == queue_id:
416            self._in_use_by_queue = None
417        if self._playing:
418            # MA-side stop/queue-clear: release the Spotify session so the app
419            # drops the device as its playback target — the daemon would
420            # otherwise keep playing into a pipe nobody consumes and the app
421            # would stay tethered to the device. (Teardowns caused by a
422            # Spotify-side pause, deselect or a player handoff never reach
423            # here: those cleared _playing or replaced the session id first.)
424            try:
425                await self._backend.deactivate()
426            except Exception as err:
427                self.logger.debug("Failed to release Spotify session on stream teardown: %s", err)
428
429    async def on_source_removed(self, source_id: str, queue_id: str) -> None:
430        """Release the Spotify session when the queue drops this source."""
431        if source_id != AUDIO_SOURCE_ID or self._active_player_id != queue_id:
432            return
433        if not self._spotify_session_active:
434            return
435        # Released whether or not a stream is still winding down: a paused source
436        # already ended its stream, so its teardown released nothing and the
437        # Spotify app would stay tethered to a device that has no queue left.
438        #
439        # Let the player go first. The backend answers a deactivate with the same
440        # 'inactive' event a deselect in the Spotify app produces, and that stops
441        # the player we were on - which by now is playing whatever took our place.
442        self._active_player_id = None
443        try:
444            await self._backend.deactivate()
445        except Exception as err:
446            self.logger.debug("Failed to release Spotify session on queue clear: %s", err)
447
448    async def on_source_transferred(
449        self, source_id: str, from_queue_id: str, to_queue_id: str
450    ) -> None:
451        """Follow this AudioSource to the queue it was handed over to."""
452        if source_id != AUDIO_SOURCE_ID or self._active_player_id != from_queue_id:
453            return
454        # A transfer while playing re-selects the source on the target and re-claims it there,
455        # but a paused one moves without a stream request: without this the plugin would stay
456        # pointed at the queue it left behind.
457        self._active_player_id = to_queue_id
458
459    async def on_source_control(
460        self,
461        source_id: str,
462        action: SourceControl,
463        value: int | None = None,
464    ) -> None:
465        """Proxy playback control commands to the backend."""
466        if source_id != AUDIO_SOURCE_ID:
467            return
468        if not self._playing and not self._spotify_session_active:
469            raise self._not_active_error()
470        try:
471            if action == SourceControl.PLAY:
472                await self._backend.resume()
473            elif action == SourceControl.PAUSE:
474                await self._backend.pause()
475            elif action == SourceControl.NEXT:
476                await self._backend.next()
477            elif action == SourceControl.PREVIOUS:
478                await self._backend.previous()
479            elif action == SourceControl.SEEK and value is not None:
480                await self._backend.seek(value * 1000)
481        except Exception as err:
482            self.logger.warning("Failed to send %s command to backend: %s", action, err)
483            raise
484
485    async def on_volume_change(self, source_id: str, volume: int) -> None:
486        """Sync the Spotify app's volume slider with the player's new volume."""
487        if source_id != AUDIO_SOURCE_ID:
488            return
489        if not self._playing and not self._spotify_session_active:
490            raise self._not_active_error()
491        # Prevent ping-pong: only push if the value actually changed from what we
492        # last sent to / received from the backend.
493        if self._last_volume_sent == volume:
494            return
495        try:
496            await self._push_volume_to_backend(volume)
497        except Exception as err:
498            self.logger.warning("Failed to send volume command to backend: %s", err)
499            raise
500
501    def _create_backend(self) -> SpotifyConnectBackend:
502        """Construct the configured Spotify Connect backend implementation."""
503        # The backend choice and soloist secrets are collected by the setup flow
504        # into setup_data; a config migrated from before the backend choice
505        # existed yields None here, which intentionally selects go-librespot
506        # (the equality check must keep treating None as the default).
507        if self.get_setup_value(CONF_BACKEND) == BACKEND_SOLOIST:
508            return SoloistBackend(
509                self.mass,
510                instance_id=self.instance_id,
511                publish_name=self._publish_name,
512                name=self.name,
513                logger=self.logger,
514                event_callback=self._handle_backend_event,
515                api_key=cast("str", self.get_setup_value(CONF_API_KEY) or ""),
516                consent=bool(self.get_setup_value(CONF_SOLOIST_CONSENT)),
517                volume_mode=self._resolve_volume_mode(),
518                crossfade_ms=self._resolve_crossfade_ms(),
519                loudness_normalization=self._resolve_loudness_normalization(),
520            )
521        return GoLibrespotBackend(
522            self.mass,
523            instance_id=self.instance_id,
524            publish_name=self._publish_name,
525            name=self.name,
526            logger=self.logger,
527            event_callback=self._handle_backend_event,
528            crossfade_ms=self._resolve_crossfade_ms(),
529            loudness_normalization=self._resolve_loudness_normalization(),
530        )
531
532    def _resolve_volume_mode(self) -> str:
533        """Return the configured volume mode (the provider options page is the only source)."""
534        return cast(
535            "str",
536            self.config.get_value(CONF_VOLUME_MODE) or VOLUME_MODE_PLAYER_ONLY,
537        )
538
539    def _resolve_crossfade_ms(self) -> int:
540        """Return the configured crossfade duration in milliseconds (0 = disabled)."""
541        value = cast("int | None", self.config.get_value(CONF_CROSSFADE_DURATION))
542        return max(0, min(int(value or 0), MAX_CROSSFADE_DURATION)) * 1000
543
544    def _resolve_loudness_normalization(self) -> bool:
545        """Return whether Spotify's loudness normalization should be enabled."""
546        value = self.config.get_value(CONF_LOUDNESS_NORMALIZATION)
547        return True if value is None else bool(value)
548
549    def _not_active_error(self) -> AudioError:
550        """Build the localized 'not the active Spotify device' error, naming this device."""
551        return AudioError(
552            NOT_ACTIVE_DEVICE_MESSAGE.format(self._publish_name),
553            translation_key="not_active_device",
554            translation_args=[self._publish_name],
555            translation_owner=self.translation_owner,
556        )
557
558    def _build_audio_source(self) -> AudioSource:
559        """
560        Construct the AudioSource MediaItem.
561
562        Backends provide a full control surface, so play / pause / seek /
563        next / previous are always available while a session is active — the
564        capability flags are static (no dependency on the Spotify Web API).
565        """
566        return AudioSource(
567            item_id=AUDIO_SOURCE_ID,
568            provider=self.instance_id,
569            name=self.name,
570            provider_mappings={
571                ProviderMapping(
572                    item_id=AUDIO_SOURCE_ID,
573                    provider_domain=self.domain,
574                    provider_instance=self.instance_id,
575                    audio_format=self._backend.audio_format,
576                )
577            },
578            can_play_pause=True,
579            can_seek=True,
580            can_next_previous=True,
581            exclusive=True,
582            allow_external_trigger=True,
583            # Browsable/startable from MA: playback resumes the last known
584            # Spotify context (claiming active device status). Without any
585            # prior context a localized error points the user to the app.
586            can_initiate=True,
587        )
588
589    def _get_target_player_id(self) -> str | None:
590        """
591        Determine the target player ID for playback.
592
593        Priority: an explicitly selected player; else (auto) a currently playing
594        player then the first available; else the configured default player.
595
596        :return: The player ID to use for playback, or None if none available.
597        """
598        if self._active_player_id:
599            if self.mass.players.get_player(self._active_player_id):
600                return self._active_player_id
601            self._active_player_id = None
602
603        if self._default_player_id == PLAYER_ID_AUTO:
604            all_players = list(self.mass.players.all_players(False, False))
605            for player in all_players:
606                if player.state.playback_state == PlaybackState.PLAYING:
607                    self.logger.debug("Auto-selecting playing player: %s", player.display_name)
608                    return player.player_id
609            if all_players:
610                first_player = all_players[0]
611                self.logger.debug(
612                    "Auto-selecting first available player: %s", first_player.display_name
613                )
614                return first_player.player_id
615            return None
616
617        if self.mass.players.get_player(self._default_player_id):
618            return self._default_player_id
619        self.logger.warning(
620            "Configured default player '%s' no longer exists", self._default_player_id
621        )
622        return None
623
624    async def _wait_for_playing(self, timeout: float = PLAYBACK_START_TIMEOUT_S) -> bool:
625        """
626        Wait up to ``timeout`` seconds for the backend to report it is playing.
627
628        :param timeout: Maximum seconds to wait.
629        :return: True once playback is confirmed, False if the timeout elapses.
630        """
631        deadline = self.mass.loop.time() + timeout
632        while True:
633            if self._playing:
634                return True
635            if self.mass.loop.time() >= deadline:
636                return False
637            await asyncio.sleep(0.1)
638
639    async def _stop_paused_player(self, player_id: str) -> None:
640        """
641        Stop the active player after a pause on a backend without stream EOF.
642
643        :param player_id: The player currently consuming the live source.
644        """
645        self.logger.debug("Stopping player %s after pause", player_id)
646        try:
647            # bounded: an unresponsive player (e.g. a throttled web client) must
648            # not hold this task - and the player's playback lock - indefinitely
649            async with asyncio.timeout(10):
650                await self.mass.players.cmd_stop(player_id)
651            self.logger.debug("Player %s stopped after pause", player_id)
652        except TimeoutError:
653            self.logger.warning("Player %s did not stop within 10s after pause", player_id)
654        except Exception as err:
655            self.logger.debug("Failed to stop player %s on pause: %s", player_id, err)
656
657    def _cancel_pending_play_media(self) -> None:
658        """Cancel any pending deferred play_media trigger."""
659        task = self._pending_play_media_task
660        if task is not None and not task.done():
661            task.cancel()
662        self._pending_play_media_task = None
663
664    def _schedule_pause_stop(self, player_id: str) -> None:
665        """
666        Dispatch the stop of the paused player, replacing a still-pending one.
667
668        :param player_id: The player currently consuming the live source.
669        """
670        self._cancel_pending_pause_stop()
671        task = self.mass.create_task(self._stop_paused_player(player_id))
672        self._pending_pause_stop_task = task
673        task.add_done_callback(self._on_pause_stop_done)
674
675    def _cancel_pending_pause_stop(self) -> None:
676        """Cancel any pending deferred stop of a paused player."""
677        task = self._pending_pause_stop_task
678        if task is not None and not task.done():
679            task.cancel()
680        self._pending_pause_stop_task = None
681
682    def _on_pause_stop_done(self, task: asyncio.Task[None]) -> None:
683        """Drop the pause-stop handle once its task finished (unless already replaced)."""
684        if self._pending_pause_stop_task is task:
685            self._pending_pause_stop_task = None
686
687    async def _deferred_play_media_fire(self) -> None:
688        """
689        Trigger play_media after a short debounce.
690
691        The backend can emit a stale 'playing' from a dying session just before it
692        reconnects; acting on it immediately would start a stream for a session
693        that is about to be replaced. Debouncing — and cancelling the task on a
694        later 'paused' / 'stopped' / 'active' event — avoids a play→stop→replay loop.
695        """
696        try:
697            await asyncio.sleep(PLAY_MEDIA_DEBOUNCE_S)
698        except asyncio.CancelledError:
699            return
700        if not self._playing or self._in_use_by_queue:
701            return
702        target_player_id = self._get_target_player_id()
703        if not target_player_id:
704            self.logger.warning(
705                "Spotify Connect playback started but no player available. "
706                "Select this source on a player to start playback."
707            )
708            return
709        self.logger.info(
710            "Starting Spotify Connect playback [%s] on player %s",
711            self.instance_id,
712            target_player_id,
713        )
714        self._active_player_id = target_player_id
715        self.mass.create_task(
716            self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
717        )
718
719    def _clear_active_player(self) -> None:
720        """Clear the active player and reset playback state when a session ends."""
721        prev_player_id = self._active_player_id
722        self._active_player_id = None
723        self._in_use_by_queue = None
724        self._active_session_id = None
725        self._playing = False
726        if prev_player_id:
727            self.logger.debug("Playback ended on player %s, clearing active player", prev_player_id)
728            self.mass.players.trigger_player_update(prev_player_id)
729
730    def _save_last_player_id(self, player_id: str) -> None:
731        """Persist the selected player ID as the new default."""
732        if self._default_player_id == player_id:
733            return
734        try:
735            self._update_setup_data(CONF_MASS_PLAYER_ID, player_id)
736            self._default_player_id = player_id
737        except Exception as err:
738            self.logger.debug("Failed to persist player ID: %s", err)
739
740    async def _handle_backend_event(self, event: BackendEvent) -> None:
741        """Dispatch a single normalized event received from the backend."""
742        if event.type is BackendEventType.CONNECTION_LOST:
743            # The backend's Spotify session is gone (e.g. daemon exit). Reset
744            # session state so a dead/restarting backend isn't treated as active
745            # and controllable; a fresh 'active' event re-establishes it.
746            self._playing = False
747            self._spotify_session_active = False
748            return
749        if event.type is BackendEventType.FATAL_ERROR:
750            self.unload_with_error(event.error or "Spotify Connect backend failed")
751            return
752        if event.type is BackendEventType.ERROR:
753            # non-fatal backend error: surface it in the log only
754            self.logger.warning("Spotify Connect backend error: %s", event.error)
755            return
756        if event.type is BackendEventType.AUTH_REQUIRED:
757            self._handle_auth_required()
758            return
759
760        # Remember the latest context/track so we can take playback back if the
761        # user moves the active device away in the Spotify app (see on_source_selected).
762        if event.context_uri:
763            self._last_context_uri = event.context_uri
764        if event.track_uri:
765            self._last_track_uri = event.track_uri
766
767        if event.type is BackendEventType.SESSION_ACTIVE:
768            self._spotify_session_active = True
769            self._last_session_active_time = time.time()
770            # A (re)activation supersedes any deferred play_media scheduled from a
771            # previous session's stale 'playing'; the fresh 'playing' that follows
772            # schedules a new one.
773            self._cancel_pending_play_media()
774            self.logger.info("Spotify Connect session active for %s", self.name)
775            # A new session starts at the backend's 100% volume default; push the
776            # target player's volume so the Spotify app's slider is correct from
777            # device selection, before any playback starts. (In the soloist
778            # player_only mode the backend pins 100% and ignores the pushed
779            # value — the app slider staying at 100 there is by design.)
780            if player_id := self._get_target_player_id():
781                await self._sync_player_volume_to_spotify(player_id)
782        elif event.type is BackendEventType.SESSION_INACTIVE:
783            self.logger.info("Spotify Connect session inactive for %s", self.name)
784            self._spotify_session_active = False
785            prev_player_id = self._active_player_id
786            self._clear_active_player()
787            if prev_player_id:
788                # bounded like the pause path: a slow player must not hold the
789                # stop (and its playback lock) indefinitely
790                self._schedule_pause_stop(prev_player_id)
791            return
792        elif event.type is BackendEventType.PLAYING:
793            self._playing = True
794            # A resume can arrive while the pause-stop is still in flight on a
795            # slow player; cancel it so it doesn't kill the restarted stream.
796            # (a stop that already completed is fine: play_media below restarts)
797            self._cancel_pending_pause_stop()
798            # Externally triggered playback: kick a play_media on the target MA
799            # player so the audio reaches a speaker. Deferred so a rapid
800            # playing/active burst from a reconnecting session can cancel it.
801            # Only while the session is active: a daemon playing without being
802            # the active Connect device (e.g. right after a deactivate) must
803            # not grab MA players in a loop.
804            if (
805                not self._in_use_by_queue
806                and self._spotify_session_active
807                and (self._pending_play_media_task is None or self._pending_play_media_task.done())
808            ):
809                self._pending_play_media_task = self.mass.create_task(
810                    self._deferred_play_media_fire()
811                )
812        elif event.type in (BackendEventType.PAUSED, BackendEventType.STOPPED):
813            was_playing = self._playing
814            self._playing = False
815            # A pause/stop is the definitive "don't start": cancel a deferred fire
816            # from a now-stale 'playing'. The active get_audio_stream sees the PCM
817            # stop and ends the stream (clean EOF), so the player leaves the playing
818            # state; the next 'playing' event re-fires play_media to resume.
819            self._cancel_pending_play_media()
820            # A pipe-fed backend keeps delivering silence on pause (no EOF), so
821            # the player must be stopped actively; the claim stays so the next
822            # 'playing' event resumes playback like the EOF path does. Only the
823            # playing→paused transition fires it: the backend reports a pause
824            # through multiple events (state delta + snapshot).
825            if (
826                was_playing
827                and not self._backend.stream_ends_on_pause
828                and (player_id := self._active_player_id)
829            ):
830                self._schedule_pause_stop(player_id)
831
832        if event.type is BackendEventType.METADATA and event.metadata is not None:
833            self._apply_metadata(event.metadata)
834        elif event.type is BackendEventType.POSITION and event.position is not None:
835            self._stream_metadata.elapsed_time = event.position
836            self._stream_metadata.elapsed_time_last_updated = int(time.time())
837
838        if event.type is BackendEventType.VOLUME and event.volume is not None:
839            await self._handle_volume_event(event.volume)
840
841        # push metadata update to the active queue item's streamdetails
842        if self._in_use_by_queue:
843            self.mass.streams.update_stream_metadata(
844                self._in_use_by_queue,
845                AUDIO_SOURCE_ID,
846                self.instance_id,
847                self._stream_metadata,
848            )
849
850    def _handle_auth_required(self) -> None:
851        """Handle a lost Spotify login: reset session state and unload with an auth error."""
852        # the backend lost its Spotify login mid-session: stop treating the
853        # device as active and unload with an auth error so the UI flags
854        # the provider and routes the user through the setup flow
855        self._playing = False
856        self._spotify_session_active = False
857        self.logger.warning("Spotify Connect backend for %s requires (re)authentication", self.name)
858        self.unload_with_error(
859            LoginFailed(
860                "Spotify authentication required",
861                translation_key="soloist_auth_required",
862                translation_owner=self.translation_owner,
863            )
864        )
865
866    def _apply_metadata(self, metadata: BackendTrackMetadata) -> None:
867        """Update the live StreamMetadata from a normalized metadata event."""
868        self._stream_metadata.uri = metadata.track_uri
869        if metadata.title:
870            self._stream_metadata.title = metadata.title
871        self._stream_metadata.artist = metadata.artist
872        self._stream_metadata.album = metadata.album
873        self._stream_metadata.image_url = metadata.image_url
874        self._stream_metadata.description = None
875        self._stream_metadata.duration = metadata.duration
876        self._stream_metadata.elapsed_time = metadata.position
877        self._stream_metadata.elapsed_time_last_updated = int(time.time())
878
879    async def _handle_volume_event(self, volume: int) -> None:
880        """
881        Apply a Spotify-side volume change to the linked MA player.
882
883        :param volume: The reported volume as a 0-100 percentage.
884        """
885        # Ignore our own echo: the backend emits a 'volume' event for the value we
886        # just pushed in on_volume_change; re-applying it would ping-pong.
887        if volume == self._last_volume_sent:
888            return
889        # Ignore the volume the backend reports right after a session becomes
890        # active — the player's own volume should win in that window.
891        if time.time() - self._last_session_active_time < INITIAL_VOLUME_GRACE_S:
892            self.logger.debug("Ignoring initial volume_changed event after session active")
893            return
894        if not self._in_use_by_queue:
895            return
896        previous_volume = self._last_volume_sent
897        self._last_volume_sent = volume
898        try:
899            await self.mass.players.cmd_volume_set(self._in_use_by_queue, volume)
900        except Exception as err:
901            # Volume sync is best-effort: the player may not support volume, or the
902            # command may fail. Restore the cached value so a retry isn't wrongly
903            # deduped, and never let it bubble up and drop the events loop.
904            self._last_volume_sent = previous_volume
905            self.logger.debug("Could not set volume on %s: %s", self._in_use_by_queue, err)
906
907    async def _sync_player_volume_to_spotify(self, player_id: str) -> None:
908        """
909        Push a player's current volume to the backend (best-effort).
910
911        :param player_id: The MA player whose volume to push.
912        """
913        player = self.mass.players.get_player(player_id)
914        if player is None or player.state.volume_level is None:
915            return
916        # clamp: the logical volume can be out of range until volume limit
917        # enforcement runs
918        volume = max(0, min(100, player.state.volume_level))
919        # No dedupe against _last_volume_sent here: it holds the last value
920        # exchanged with the backend, not the backend's current volume, which
921        # resets to its 100% default on a new session or backend restart.
922        try:
923            await self._push_volume_to_backend(volume)
924        except Exception as err:
925            self.logger.debug("Failed to sync player volume to Spotify: %s", err)
926
927    async def _push_volume_to_backend(self, volume: int) -> None:
928        """
929        Send an absolute 0-100 volume to the backend.
930
931        :param volume: Volume percentage to send.
932        :raises Exception: If the request to the backend fails.
933        """
934        previous_volume = self._last_volume_sent
935        # Record BEFORE the call: the backend echoes a 'volume' event back, and
936        # that echo can arrive over the event stream while we're still awaiting
937        # set_volume. Recording up front lets _handle_volume_event dedupe it
938        # instead of bouncing it back as a player volume change.
939        self._last_volume_sent = volume
940        try:
941            await self._backend.set_volume(volume)
942        except Exception:
943            # restore on failure so a retry of this value isn't wrongly deduped
944            self._last_volume_sent = previous_volume
945            raise
946