/
/
/
1"""
2Spotify Soloist backend for the Spotify Connect provider.
3
4Soloist is Spotify's official headless Connect client for Linux. The daemon is
5installed/updated through ``SoloistBinaryManager``, plays its audio into a
6private PulseAudio capture sink (read back by MA as a named pipe) and is driven
7over its local WebSocket API (``SoloistClient``).
8
9SECURITY NOTE: the daemon takes the user's personal API key on its command
10line; nothing in this module may ever log the process argv.
11"""
12
13from __future__ import annotations
14
15import asyncio
16import re
17from asyncio import FIRST_COMPLETED
18from contextlib import suppress
19from dataclasses import replace
20from pathlib import Path
21from typing import TYPE_CHECKING, Any, Final
22
23from aiohttp import ClientError
24from music_assistant_models.enums import ContentType, RepeatMode, StreamType
25from music_assistant_models.errors import AudioError
26from music_assistant_models.media_items import AudioFormat
27
28from music_assistant.helpers.process import AsyncProcess
29from music_assistant.helpers.pulse_capture import (
30 CAPTURE_CHANNELS,
31 CAPTURE_SAMPLE_RATE,
32 PipeSink,
33 get_pulse_capture_server,
34)
35from music_assistant.providers.spotify_connect.base import (
36 AUDIO_QUALITY_LOSSLESS,
37 SpotifyConnectBackend,
38 spotify_source_audio_format,
39)
40from music_assistant.providers.spotify_connect.models import (
41 BackendEvent,
42 BackendEventType,
43 BackendPlaybackOptions,
44 BackendQueueEntry,
45 BackendQueueState,
46 BackendStreamSource,
47 BackendTrackMetadata,
48 QueueEntrySource,
49)
50
51from .prefs import write_audio_prefs
52from .runtime import (
53 EXIT_CODE_BUILD_EXPIRED,
54 BuildExpiredError,
55 SoloistAuthState,
56 SoloistBinaryManager,
57 SoloistClient,
58 SoloistContextChanged,
59 SoloistDeviceChanged,
60 SoloistError,
61 SoloistErrorMessage,
62 SoloistOptionsChanged,
63 SoloistPlaybackState,
64 SoloistPositionSync,
65 SoloistQueueChanged,
66 SoloistTrackChanged,
67 SoloistVolumeChanged,
68)
69
70if TYPE_CHECKING:
71 import logging
72
73 from music_assistant.helpers.pulse_capture import PulseCaptureServer
74 from music_assistant.mass import MusicAssistant
75 from music_assistant.providers.spotify_connect.models import (
76 AudioChunkReader,
77 BackendEventCallback,
78 )
79
80 from .runtime import SoloistEntity, SoloistEvent, SoloistPlaybackOptions, SoloistQueueEntry
81
82# How Spotify-side volume changes are handled (see set_volume).
83VOLUME_MODE_PLAYER_ONLY: Final = "player_only"
84VOLUME_MODE_SYNC_SPOTIFY: Final = "sync_spotify"
85
86# Bounded soloist playback cache (docs: 0 = unlimited, otherwise at least 100 MB).
87CACHE_SIZE_MB: Final = 512
88
89# Supervised-restart policy: give up after this many consecutive daemon
90# failures; the counter resets once the events websocket delivers again.
91MAX_RESTART_ATTEMPTS: Final = 5
92RESTART_DELAY_S: Final = 2
93
94# How long the daemon's log reader may keep draining buffered output after the
95# process itself exited.
96DAEMON_LOG_DRAIN_TIMEOUT_S: Final = 5
97
98# Proactive binary refresh interval: soloist builds expire 90 days after their
99# build date, so a daily check swaps in a fresh build long before a long-lived
100# instance would hit the expiry.
101BINARY_REFRESH_INTERVAL_S: Final = 24 * 3600
102
103# How often the supervisor checks whether the pulse daemon restarted (which
104# invalidates the capture sink), so the sink is replaced proactively instead
105# of on the (side-effect-free) stream request.
106GENERATION_WATCH_INTERVAL_S: Final = 5
107
108# item uri prefixes Spotify never serves losslessly, whatever the tier is set to.
109_SPOKEN_URI_PREFIXES: Final = ("spotify:episode:", "spotify:chapter:")
110
111# playback_state/playback_changed status values mapped to normalized events;
112# undocumented values degrade to OTHER.
113_STATUS_EVENTS: Final[dict[str, BackendEventType]] = {
114 "playing": BackendEventType.PLAYING,
115 "paused": BackendEventType.PAUSED,
116 "buffering": BackendEventType.BUFFERING,
117 "stopped": BackendEventType.STOPPED,
118 "idle": BackendEventType.STOPPED,
119}
120
121
122class SoloistBackend(SpotifyConnectBackend):
123 """
124 Spotify Connect backend wrapping a supervised Spotify Soloist daemon.
125
126 The daemon plays into a private PulseAudio capture sink whose FIFO is
127 consumed by the streams controller as a named pipe; control and state flow
128 over the daemon's local WebSocket API.
129 """
130
131 def __init__( # noqa: PLR0913
132 self,
133 mass: MusicAssistant,
134 *,
135 identity_key: str,
136 publish_name: str,
137 name: str,
138 logger: logging.Logger,
139 event_callback: BackendEventCallback,
140 api_key: str,
141 consent: bool,
142 volume_mode: str = VOLUME_MODE_PLAYER_ONLY,
143 crossfade_ms: int = 0,
144 loudness_normalization: bool = True,
145 audio_quality: str = AUDIO_QUALITY_LOSSLESS,
146 ) -> None:
147 """
148 Initialize the backend (cheap; the daemon is launched in ``start``).
149
150 :param mass: The MusicAssistant instance.
151 :param identity_key: Unique identity of this daemon (one per connected
152 player); keys the data/cache dirs and the capture sink name.
153 :param publish_name: Device name advertised to the Spotify app.
154 :param name: Display name of the owning provider instance (log messages).
155 :param logger: Logger to use for diagnostics.
156 :param event_callback: Awaited with a normalized BackendEvent for every
157 state change the daemon reports.
158 :param api_key: The user's personal Spotify Soloist API key (secret,
159 kept out of all logs).
160 :param consent: Whether the user consented to downloading the soloist
161 binary from Spotify's CDN.
162 :param volume_mode: VOLUME_MODE_PLAYER_ONLY to let MA/the player own the
163 volume exclusively, VOLUME_MODE_SYNC_SPOTIFY to mirror the Spotify
164 app's volume onto the MA player.
165 :param crossfade_ms: Crossfade duration between tracks in milliseconds
166 (0 disables crossfade).
167 :param loudness_normalization: Whether Spotify's loudness normalization
168 should be applied to the audio.
169 :param audio_quality: Ceiling for the streaming quality Spotify is asked
170 to deliver (one of the AUDIO_QUALITY_* tiers).
171 """
172 self.mass = mass
173 self.logger = logger
174 self.name = name
175 self._publish_name = publish_name
176 self._event_callback = event_callback
177 self._api_key = api_key
178 self._consent = consent
179 self._volume_mode = volume_mode
180 self._crossfade_ms = crossfade_ms
181 self._loudness_normalization = loudness_normalization
182 self._audio_quality = audio_quality
183 self._data_dir = Path(mass.storage_path) / "spotify_connect" / identity_key / "soloist-data"
184 self._cache_dir = Path(mass.cache_path) / identity_key / "soloist-cache"
185 # PA sink names end up in space-delimited module arguments and env vars
186 self._sink_prefix = re.sub(r"[^A-Za-z0-9_.-]", "_", identity_key)
187 self._binary: Path | None = None
188 # digest of the build the running daemon was spawned from; the shared
189 # install can move ahead of it when a sibling instance updates first
190 self._build_sha: str | None = None
191 self._server: PulseCaptureServer | None = None
192 self._sink: PipeSink | None = None
193 self._sink_generation: int = -1
194 # serializes sink (re)creation against teardown in stop()
195 self._sink_lock = asyncio.Lock()
196 self._client: SoloistClient | None = None
197 self._stop_called: bool = False
198 self._daemon_task: asyncio.Task[None] | None = None
199 self._events_task: asyncio.Task[None] | None = None
200 self._refresh_task: asyncio.Task[None] | None = None
201 self._watcher_task: asyncio.Task[None] | None = None
202 self._proc: AsyncProcess | None = None
203 self._restart_error_count = 0
204 # set when a daemon close is intentional (sink replaced), so the
205 # supervisor respawns immediately instead of counting a failure
206 self._respawn_requested: bool = False
207 # serializes all volume handling (sink compensation + event forwarding)
208 self._volume_lock = asyncio.Lock()
209 # serializes the two-command repeat sequences of concurrent set_repeat calls
210 self._repeat_lock = asyncio.Lock()
211 # last volume reported by the daemon (None until the first event)
212 self._spotify_volume: int | None = None
213 # guards the player_only 100%-pin so overlapping resets are not issued
214 self._pin_in_flight: bool = False
215 self._last_context_uri: str | None = None
216 self._last_track_uri: str | None = None
217 # the capture sink delivers fixed s32le/44.1kHz/2ch PCM â that is what
218 # actually arrives on the named pipe, and every decision about the bytes
219 # follows it because it is reported as the decoded format
220 self._capture_format = AudioFormat(
221 content_type=ContentType.PCM_S32LE,
222 codec_type=ContentType.PCM_S32LE,
223 sample_rate=CAPTURE_SAMPLE_RATE,
224 bit_depth=32,
225 channels=CAPTURE_CHANNELS,
226 )
227 self._tier_format = spotify_source_audio_format(
228 audio_quality, lossless=audio_quality == AUDIO_QUALITY_LOSSLESS
229 )
230 # spoken content is Ogg Vorbis whatever the tier says; on the lossy tiers
231 # that is the tier format itself
232 self._spoken_format = (
233 spotify_source_audio_format(audio_quality, lossless=False)
234 if audio_quality == AUDIO_QUALITY_LOSSLESS
235 else self._tier_format
236 )
237
238 @property
239 def audio_format(self) -> AudioFormat:
240 """Return the source audio format (advertised to clients for display)."""
241 # a Connect session plays whatever the Spotify app picked, so the uri of the
242 # item playing when the stream starts is the only media-type signal there is;
243 # an unknown item is treated as music, the dominant case for a ceiling claim
244 if self._last_track_uri and self._last_track_uri.startswith(_SPOKEN_URI_PREFIXES):
245 return self._spoken_format
246 return self._tier_format
247
248 @property
249 def decoded_audio_format(self) -> AudioFormat:
250 """Return the PCM format the capture sink's named pipe delivers."""
251 # a copy per stream: the core mirrors what ffmpeg probes onto this object,
252 # which must not land on the one format every stream shares
253 return replace(self._capture_format)
254
255 @property
256 def stream_ends_on_pause(self) -> bool:
257 """The pipe sink never signals end of stream; the provider stops the player."""
258 # the sink renders only while a client is connected: silence while the daemon
259 # holds its stream open, nothing at all once the daemon drops it. A reader
260 # therefore sees neither audio nor EOF, so MA has to end the stream itself.
261 return False
262
263 @property
264 def supports_queue_control(self) -> bool:
265 """The soloist session implements the queue verbs and queue/options events."""
266 return True
267
268 async def start(self) -> None:
269 """Start the backend and its supervised soloist daemon."""
270 # setup errors (unsupported platform, missing consent, download failure,
271 # expired build) propagate so the provider load fails with a clear error
272 manager = SoloistBinaryManager(self.mass)
273 self._binary = await manager.ensure_fresh(self._consent)
274 self._build_sha = manager.diagnostics().get("sha256")
275 self._server = await get_pulse_capture_server(self.mass).acquire()
276 try:
277 await self._ensure_fresh_sink()
278
279 def _prepare_dirs() -> None:
280 self._data_dir.mkdir(parents=True, exist_ok=True)
281 # the data dir persists the Spotify device identity and login
282 # session; keep it readable by the MA user only
283 self._data_dir.chmod(0o700)
284 self._cache_dir.mkdir(parents=True, exist_ok=True)
285
286 await asyncio.to_thread(_prepare_dirs)
287 self._client = SoloistClient(self.mass, self._data_dir, self.logger)
288 # Two self-healing supervisors: one keeps the daemon process alive,
289 # the other keeps the events websocket connected (reconnecting
290 # across daemon restarts).
291 self._daemon_task = self.mass.create_task(self._daemon_runner())
292 self._events_task = self.mass.create_task(self._events_runner())
293 # Two housekeeping loops: a daily binary refresh (builds expire 90
294 # days after their build date) and a watcher replacing the capture
295 # sink after a pulse daemon restart.
296 self._refresh_task = self.mass.create_task(self._binary_refresh_loop())
297 self._watcher_task = self.mass.create_task(self._generation_watcher())
298 except BaseException:
299 # a failed startup aborts the provider load before unload() would
300 # ever run â release everything acquired so far ourselves
301 with suppress(Exception):
302 await self.stop()
303 raise
304
305 async def stop(self) -> None:
306 """Stop the daemon, its supervisors and the capture resources (idempotent)."""
307 self._stop_called = True
308 for task in (self._events_task, self._daemon_task, self._refresh_task, self._watcher_task):
309 if task and not task.done():
310 task.cancel()
311 with suppress(asyncio.CancelledError):
312 await task
313 self._events_task = None
314 self._daemon_task = None
315 self._refresh_task = None
316 self._watcher_task = None
317 if (proc := self._proc) is not None:
318 self._proc = None
319 await proc.close()
320 # the lock lets an in-flight _ensure_fresh_sink finish before the
321 # capture resources go away (later callers fail on the stop flag)
322 async with self._sink_lock:
323 if (sink := self._sink) is not None:
324 self._sink = None
325 await sink.unload()
326 if (server := self._server) is not None:
327 self._server = None
328 await server.release()
329
330 async def get_stream_source(self) -> BackendStreamSource:
331 """
332 Return the NAMED_PIPE stream source delivering the capture sink's PCM.
333
334 Side-effect-free (it also runs from queue preload): the sink is only
335 read here â (re)creation is owned by the daemon supervisor and the
336 generation watcher.
337
338 ffmpeg reads the FIFO directly and is the single pacing owner:
339 ``-readrate 1`` paces the read to realtime with a small initial burst
340 as jitter headroom â nothing else may sleep-pace this audio path.
341
342 :raises AudioError: No usable capture sink is currently available.
343 """
344 sink = self._sink
345 server = self._server
346 if sink is None or server is None or self._sink_generation != server.generation:
347 raise AudioError("Spotify Connect capture sink is not available")
348 return BackendStreamSource(
349 stream_type=StreamType.NAMED_PIPE,
350 path=str(sink.fifo_path),
351 extra_input_args=["-readrate", "1", "-readrate_initial_burst", "0.5"],
352 )
353
354 def get_audio_reader(self) -> AudioChunkReader | None:
355 """Return None: the audio is delivered through the NAMED_PIPE stream source."""
356 return None
357
358 async def play(self, uri: str, *, skip_to_uri: str | None = None) -> None:
359 """
360 Start playing a Spotify URI/context, making this device the active one.
361
362 :param uri: Spotify URI (track, album, playlist, ...) â typically a context.
363 :param skip_to_uri: Ignored â soloist's play command has no skip-to-track
364 option, so playback starts at the beginning of the context.
365 """
366 assert self._client is not None
367 # a bare play starts local playback without a Connect transfer, leaving
368 # the Spotify apps unaware; claim active device status first
369 await self._client.activate(await_result=True)
370 await self._client.play(uri)
371 if skip_to_uri:
372 self.logger.debug(
373 "skip_to_uri is not supported by soloist; starting %s from the beginning", uri
374 )
375
376 async def resume(self) -> None:
377 """Resume playback on the active session."""
378 assert self._client is not None
379 # re-claim active device status first (idempotent when already active):
380 # a resume after a deactivate would otherwise start local playback
381 # without a Connect transfer, leaving the Spotify apps unaware
382 await self._client.activate(await_result=True)
383 await self._client.resume()
384
385 async def pause(self) -> None:
386 """Pause playback on the active session."""
387 assert self._client is not None
388 await self._client.pause()
389
390 async def deactivate(self) -> None:
391 """Release this device as the active Spotify Connect device."""
392 assert self._client is not None
393 # pause first so the session's resume position is preserved; tolerate
394 # a rejected or unacknowledged pause and still give up the device
395 with suppress(SoloistError, TimeoutError):
396 await self._client.pause(await_result=True)
397 await self._client.deactivate()
398
399 async def next(self) -> None:
400 """Skip to the next track."""
401 assert self._client is not None
402 await self._client.skip_next()
403
404 async def previous(self) -> None:
405 """Skip to the previous track (or rewind the current one)."""
406 assert self._client is not None
407 await self._client.skip_prev()
408
409 async def seek(self, position_ms: int) -> None:
410 """
411 Seek to an absolute position in the current track.
412
413 :param position_ms: Target position in milliseconds.
414 """
415 assert self._client is not None
416 await self._client.seek(position_ms)
417
418 async def set_volume(self, volume: int) -> None:
419 """
420 Set the Spotify-side playback volume.
421
422 :param volume: Absolute volume as a 0-100 percentage. In player-only
423 mode the value is ignored and the daemon stays pinned at 100%.
424 """
425 assert self._client is not None
426 if self._volume_mode == VOLUME_MODE_SYNC_SPOTIFY:
427 # the daemon echoes the change back as a volume_changed event; the
428 # provider's dedupe keeps that echo from bouncing back to the player
429 async with self._volume_lock:
430 await self._client.set_volume(volume)
431 return
432 # player_only: the MA player owns the volume, so the daemon is pinned at
433 # 100% (unity PCM into the sink); only (re)send the pin when it is known
434 # (or possibly) off
435 async with self._volume_lock:
436 if self._spotify_volume == 100:
437 return
438 await self._client.set_volume(100)
439
440 async def add_to_queue(self, uri: str) -> None:
441 """
442 Add a track to the session's play queue.
443
444 :param uri: Spotify track URI to queue.
445 """
446 assert self._client is not None
447 await self._client.add_to_queue(uri)
448
449 async def set_shuffle(self, enabled: bool) -> None:
450 """
451 Enable or disable shuffle on the active session.
452
453 :param enabled: True to enable shuffle, False to disable it.
454 """
455 assert self._client is not None
456 await self._client.set_shuffle(enabled)
457
458 async def set_repeat(self, repeat: RepeatMode) -> None:
459 """
460 Set the repeat mode on the active session.
461
462 Awaits the engine's acknowledgement of both underlying commands, so
463 this call can block and raise; never call it from the backend event
464 callback (the acknowledgements arrive on the same loop and the wait
465 could only time out).
466
467 :param repeat: OFF for no repeat, ONE for the current track, ALL for
468 the playing context.
469 """
470 assert self._client is not None
471 if repeat == RepeatMode.UNKNOWN:
472 raise ValueError("cannot apply an unknown repeat mode")
473 # soloist models repeat as two independent booleans (track/context) with
474 # both-true undefined: the lock serializes concurrent callers and the
475 # awaited acks order the pair, so one flag is always cleared before the
476 # other is raised
477 async with self._repeat_lock:
478 if repeat == RepeatMode.ONE:
479 await self._client.set_repeat_context(False, await_result=True)
480 await self._client.set_repeat_track(True, await_result=True)
481 return
482 await self._client.set_repeat_track(False, await_result=True)
483 await self._client.set_repeat_context(repeat == RepeatMode.ALL, await_result=True)
484
485 async def request_queue(self, limit: int = 10) -> None:
486 """
487 Ask the session to (re)emit its queue view (arrives as a QUEUE_CHANGED event).
488
489 :param limit: Maximum number of upcoming entries the snapshot should
490 include.
491 """
492 assert self._client is not None
493 await self._client.get_queue(limit)
494
495 async def _ensure_fresh_sink(self) -> PipeSink:
496 """
497 Return the capture sink, replacing it when the pulse daemon restarted.
498
499 Only called from the supervisor paths (daemon spawn), never from the
500 side-effect-free stream request.
501
502 :raises AudioError: The backend is (being) stopped.
503 """
504 async with self._sink_lock:
505 # checked under the lock: a concurrent stop() sets the flag before
506 # it waits for the lock to tear down the capture resources
507 if self._stop_called or self._server is None:
508 raise AudioError("Spotify Connect backend is stopping")
509 if self._sink is not None and self._sink_generation == self._server.generation:
510 return self._sink
511 if self._sink is not None:
512 # sinks do not survive a daemon restart; this only cleans up the FIFO
513 await self._sink.unload()
514 self._sink = await PipeSink.create(self._server, self._sink_prefix)
515 self._sink_generation = self._server.generation
516 # the daemon plays into the sink named in its spawn env (PULSE_SINK),
517 # so a running process must respawn against the recreated sink
518 await self._close_daemon_for_respawn()
519 return self._sink
520
521 async def _recover_sink(self) -> None:
522 """
523 Fail-closed recovery: drop the sink and daemon so the supervisor rebuilds both.
524
525 Used when the current sink can no longer be trusted (its compensation
526 state is unknown after a failed volume call, or the pulse daemon
527 restarted underneath it): audio through such a sink risks a stale
528 reciprocal gain of up to 100x, so both the sink and the daemon are
529 torn down and recreated by the daemon supervisor.
530 """
531 async with self._sink_lock:
532 if self._stop_called:
533 return
534 if (sink := self._sink) is not None:
535 self._sink = None
536 with suppress(Exception):
537 await sink.unload()
538 await self._close_daemon_for_respawn()
539
540 async def _close_daemon_for_respawn(self) -> None:
541 """Close a running daemon so its supervisor respawns it right away (not a failure)."""
542 if (proc := self._proc) is not None:
543 self._respawn_requested = True
544 await proc.close()
545
546 async def _generation_watcher(self) -> None:
547 """Proactively replace the capture sink when the pulse daemon restarted."""
548 while True:
549 await asyncio.sleep(GENERATION_WATCH_INTERVAL_S)
550 server = self._server
551 if server is None or self._sink is None:
552 continue
553 if self._sink_generation != server.generation:
554 self.logger.info("Pulse daemon restart detected; recreating the capture sink")
555 await self._recover_sink()
556
557 def _daemon_args(self) -> list[str]:
558 """
559 Build the soloist daemon's argv.
560
561 SECURITY: the argv carries the user's API key â it must never be logged
562 or end up in any error message.
563 """
564 assert self._binary is not None
565 return [
566 str(self._binary),
567 "--device-name",
568 self._publish_name,
569 "--api-key",
570 self._api_key,
571 "--data-dir",
572 str(self._data_dir),
573 "--cache-dir",
574 str(self._cache_dir),
575 # bounded playback cache (0 would be unlimited)
576 "--cache-size",
577 str(CACHE_SIZE_MB),
578 # start at 100%: MA (or the sink compensation) owns the real volume
579 "--initial-volume",
580 "100",
581 # local WebSocket API on a free loopback port; the daemon publishes
582 # the actual endpoint in its data dir where SoloistClient finds it
583 "--ws",
584 "127.0.0.1:0",
585 ]
586
587 def _write_audio_prefs(self) -> None:
588 """Write the configured audio behavior into the engine's prefs stores (blocking)."""
589 # best-effort here: nothing outside this backend depends on the result
590 write_audio_prefs(
591 self._data_dir,
592 self.logger,
593 crossfade_ms=self._crossfade_ms,
594 loudness_normalization=self._loudness_normalization,
595 audio_quality=self._audio_quality,
596 )
597
598 async def _daemon_runner(self) -> None:
599 """Run and supervise the soloist daemon, restarting (and refreshing) as needed."""
600 # Loop forever; stop() cancels this task and the explicit stop-check below
601 # handles a graceful exit without a restart.
602 while True:
603 proc: AsyncProcess | None = None
604 returncode: int | None = None
605 try:
606 # the sink may have been replaced (pulse restart) while the
607 # daemon was down; never spawn against a stale sink
608 sink = await self._ensure_fresh_sink()
609 server = self._server
610 assert server is not None # guaranteed by _ensure_fresh_sink
611 # the engine only reads its prefs at startup (and scrubs foreign
612 # keys from the global store when it rewrites it), so refresh the
613 # audio settings on every spawn, while the daemon is down
614 await asyncio.to_thread(self._write_audio_prefs)
615 # the explicit process name keeps AsyncProcess logging free of
616 # the argv (which carries the API key)
617 self._proc = proc = AsyncProcess(
618 self._daemon_args(),
619 # the daemon writes all of its logging to stdout and only
620 # ever puts argument-parsing complaints on stderr, so the
621 # two are merged into one captured stream. Capturing is
622 # what makes the redaction below reachable at all: an
623 # unset stdout is inherited, which would leak the daemon's
624 # output straight to the server console instead.
625 stdout=True,
626 stderr=asyncio.subprocess.STDOUT,
627 name=f"soloist[{self.name}]",
628 env=server.child_env(sink.sink_name),
629 )
630 await proc.start()
631 self.logger.info("Started Spotify Connect background daemon [%s]", self.name)
632 await self._reset_volume_state(sink)
633 await self._await_daemon_exit(proc)
634 except asyncio.CancelledError:
635 raise
636 except Exception as err:
637 self.logger.warning("soloist daemon error [%s]: %s", self.name, err)
638 finally:
639 if proc:
640 await proc.close()
641 returncode = proc.returncode
642 # The daemon â and thus the Spotify session â is gone. Tell the
643 # provider so a dead/restarting daemon isn't treated as active and
644 # controllable; a fresh 'active' event re-establishes it on reconnect.
645 self._proc = None
646 try:
647 await self._event_callback(BackendEvent(BackendEventType.CONNECTION_LOST))
648 except Exception:
649 # never let a callback error replace a propagating
650 # cancellation or kill the daemon supervisor
651 self.logger.exception("Error while handling daemon exit")
652 if self._stop_called:
653 break
654 if self._respawn_requested:
655 # intentional close (the sink was replaced): respawn right away,
656 # this is not a daemon failure
657 self._respawn_requested = False
658 continue
659 self.logger.info("Spotify Connect background daemon stopped for %s", self.name)
660 if returncode == EXIT_CODE_BUILD_EXPIRED and not await self._refresh_expired_binary():
661 return
662 self._restart_error_count += 1
663 if self._restart_error_count >= MAX_RESTART_ATTEMPTS:
664 await self._event_callback(
665 BackendEvent(
666 BackendEventType.FATAL_ERROR,
667 # fatal errors are plain (non-localized) strings for now,
668 # matching the go-librespot backend
669 error="soloist daemon failed to start multiple times.",
670 # repeated soloist exits are dominated by engine-level
671 # problems (e.g. a bad or revoked API key) that hit every
672 # daemon alike
673 provider_wide=True,
674 )
675 )
676 return
677 await asyncio.sleep(RESTART_DELAY_S)
678
679 async def _await_daemon_exit(self, proc: AsyncProcess) -> None:
680 """
681 Wait for the daemon to exit, forwarding its log for as long as it runs.
682
683 :param proc: The running daemon process.
684 """
685 # The log is drained by a side task rather than inline: a close() from
686 # one of the other supervisors (sink replacement, binary refresh) locks
687 # readers out of the process streams for good, so waiting on the reader
688 # would hang here and the daemon would never be respawned.
689 log_task = asyncio.create_task(self._log_daemon_output(proc))
690 wait_task = asyncio.create_task(proc.wait())
691 try:
692 # Watch both: nothing else drains the daemon's stdout, so a reader
693 # that died would leave the daemon blocked on a full pipe and this
694 # wait would never return.
695 await asyncio.wait((wait_task, log_task), return_when=FIRST_COMPLETED)
696 reader_error = (
697 log_task.exception() if log_task.done() and not log_task.cancelled() else None
698 )
699 if reader_error is not None:
700 self.logger.error(
701 "soloist log reader failed [%s]: %s; restarting the daemon",
702 self.name,
703 reader_error,
704 )
705 await proc.close()
706 await wait_task
707 # an exited daemon still has its last (often most telling) lines in
708 # the stream buffer; the shield keeps the reader alive across the
709 # timeout so it can drain them
710 with suppress(TimeoutError):
711 await asyncio.wait_for(asyncio.shield(log_task), DAEMON_LOG_DRAIN_TIMEOUT_S)
712 finally:
713 # a reader locked out by a close() from another supervisor never
714 # ends on its own; joining it consumes its outcome the way the
715 # other process readers in the codebase do
716 for task in (wait_task, log_task):
717 task.cancel()
718 for task in (wait_task, log_task):
719 with suppress(asyncio.CancelledError, Exception):
720 await task
721
722 async def _log_daemon_output(self, proc: AsyncProcess) -> None:
723 """
724 Forward the daemon's log lines to our logger until its output ends.
725
726 :param proc: The running daemon process.
727 """
728 async for line in proc.iter_stdout():
729 # the third-party binary's own output may echo argv (which carries
730 # the api key), so redact it before logging
731 text = line.replace(self._api_key, "<redacted>") if self._api_key else line
732 self.logger.debug("[%s] %s", self.name, text)
733
734 async def _reset_volume_state(self, sink: PipeSink) -> None:
735 """
736 Realign the volume bookkeeping with a freshly spawned daemon.
737
738 :param sink: The capture sink the daemon plays into.
739 """
740 # the daemon always starts at --initial-volume 100; without this reset a
741 # stale reciprocal sink compensation from before a crash would amplify
742 # and clip the captured audio (sync_spotify mode)
743 async with self._volume_lock:
744 self._spotify_volume = 100
745 try:
746 await sink.set_volume(100)
747 except Exception as err:
748 # fail closed: a stale reciprocal gain may still be active on
749 # the sink â recreate sink and daemon rather than clipping
750 self.logger.warning(
751 "Failed to reset capture sink volume (%s); recreating capture sink", err
752 )
753 await self._recover_sink()
754
755 async def _refresh_expired_binary(self) -> bool:
756 """
757 Replace the expired soloist build before the next daemon restart.
758
759 :return: True when the supervisor may restart the daemon, False when a
760 fatal error was reported and the supervisor must stop.
761 """
762 self.logger.warning("soloist build expired; looking for a replacement build")
763 try:
764 # force: the daemon itself reported expiry, so the recently-verified
765 # fast path must not hand back the same binary
766 manager = SoloistBinaryManager(self.mass)
767 self._binary = await manager.ensure_fresh(self._consent, force=True)
768 self._build_sha = manager.diagnostics().get("sha256")
769 except BuildExpiredError:
770 await self._event_callback(
771 BackendEvent(
772 BackendEventType.FATAL_ERROR,
773 error=(
774 "The Spotify Soloist build expired and no replacement could be "
775 "installed. Check the server's internet connection and reload "
776 "this provider."
777 ),
778 # the binary (and its expiry) is shared by every daemon
779 provider_wide=True,
780 )
781 )
782 return False
783 except SoloistError as err:
784 # transient refresh problem: keep restarting, bounded by the failure cap
785 self.logger.warning("Unable to refresh the expired soloist build: %s", err)
786 return True
787
788 async def _binary_refresh_loop(self) -> None:
789 """Periodically refresh the soloist binary ahead of its 90-day build expiry."""
790 while True:
791 await asyncio.sleep(BINARY_REFRESH_INTERVAL_S)
792 try:
793 manager = SoloistBinaryManager(self.mass)
794 # a replaced build keeps the same install path, so compare the
795 # install metadata's digest against the build this daemon runs
796 # (a sibling instance may have updated the shared install)
797 binary = await manager.ensure_fresh(self._consent)
798 new_sha = manager.diagnostics().get("sha256")
799 if binary == self._binary and new_sha == self._build_sha:
800 continue
801 self.logger.info("A fresh soloist build was installed; restarting the daemon")
802 self._binary = binary
803 self._build_sha = new_sha
804 await self._close_daemon_for_respawn()
805 except asyncio.CancelledError:
806 raise
807 except Exception as err:
808 self.logger.warning("Periodic soloist binary refresh failed: %s", err)
809
810 async def _events_runner(self) -> None:
811 """Keep the soloist events websocket connected, reconnecting as needed."""
812 assert self._client is not None
813 while not self._stop_called:
814 try:
815 if not await self._client.wait_until_ready():
816 await asyncio.sleep(RESTART_DELAY_S)
817 continue
818 await self._client.listen_events(self._handle_event)
819 except asyncio.CancelledError:
820 raise
821 except (TimeoutError, OSError, ClientError, SoloistError) as err:
822 # ordinary connection drop; the loop reconnects quietly
823 self.logger.debug("soloist events websocket dropped: %s", err)
824 except Exception:
825 # a defect in event handling must not kill the control plane,
826 # but unlike a connection drop it has to surface loudly
827 self.logger.exception("Unexpected error while handling soloist events")
828 if not self._stop_called:
829 await asyncio.sleep(RESTART_DELAY_S)
830
831 async def _handle_event(self, event: SoloistEvent) -> None:
832 """Adapt a raw soloist event and emit its normalized counterpart."""
833 self.logger.debug("Received %s event [%s]", event.type, self.name)
834 # A delivered event means the websocket â and thus the daemon â is
835 # healthy: reset the restart backoff counter the daemon supervisor uses.
836 # (The endpoint files the events runner polls can be stale leftovers
837 # from a previous run, so the reset cannot happen on wait_until_ready.)
838 self._restart_error_count = 0
839 if isinstance(event.data, SoloistVolumeChanged):
840 await self._handle_volume_changed(event.data.volume)
841 return
842 if (
843 isinstance(event.data, SoloistPlaybackState)
844 and event.data.volume is not None
845 and event.data.volume != self._spotify_volume
846 ):
847 # a playback_state snapshot (e.g. right after a websocket reconnect)
848 # carries the daemon's current volume; resync the pin/compensation
849 # before forwarding the playback event itself
850 await self._handle_volume_changed(event.data.volume)
851 if self._sink is None:
852 # fail-closed recovery tore down the capture path; drop this
853 # stale snapshot â the respawned daemon reports fresh state
854 return
855 if (
856 isinstance(event.data, SoloistPlaybackState)
857 and (item := event.data.item) is not None
858 and item.uri != self._last_track_uri
859 ):
860 # the snapshot describes a track we have no metadata for yet (an
861 # already-playing session at (re)connect); emit its metadata so it
862 # does not stay stale until the next track_changed
863 await self._event_callback(
864 self._make_event(BackendEventType.METADATA, metadata=_entity_metadata(item))
865 )
866 if isinstance(event.data, SoloistPlaybackState) and event.data.options is not None:
867 # a state snapshot/delta carrying the playback options doubles as an
868 # options report; emit a separate OPTIONS_CHANGED so consumers only
869 # ever need to watch one event type for options
870 await self._event_callback(
871 self._make_event(
872 BackendEventType.OPTIONS_CHANGED,
873 options=_backend_options(event.data.options),
874 )
875 )
876 await self._event_callback(self._translate_event(event))
877
878 async def _handle_volume_changed(self, volume: int) -> None:
879 """
880 Apply a Spotify-side volume change according to the configured volume mode.
881
882 :param volume: The reported volume as a 0-100 percentage.
883 """
884 async with self._volume_lock:
885 self._spotify_volume = volume
886 if self._volume_mode != VOLUME_MODE_SYNC_SPOTIFY:
887 # player_only: MA/the player owns the volume. Keep the daemon
888 # pinned at 100% so the captured PCM stays at unity gain, and
889 # never forward VOLUME events (they would fight the MA volume).
890 if volume != 100 and self._client is not None and not self._pin_in_flight:
891 self._pin_in_flight = True
892 try:
893 await self._client.set_volume(100)
894 except Exception as err:
895 self.logger.debug("Failed to reset soloist volume: %s", err)
896 # mark the volume unknown so the next snapshot
897 # reporting the same value still retries the pin
898 self._spotify_volume = None
899 finally:
900 self._pin_in_flight = False
901 return
902 # sync_spotify: the daemon attenuates the PCM it plays into the sink
903 # with Spotify's cubic volume curve; undo that with the reciprocal
904 # raw sink gain (sink_pct = 10000 / spotify_pct) so the FIFO always
905 # carries unity-gain audio, then forward the volume for the MA
906 # player to apply (subject to the provider's dedupe/grace policy).
907 # NOTE: the percentage reciprocal is the exact linear inverse
908 # because pulse's software volume is cubic in the percentage too
909 # (pa_sw_volume_to_linear(p) = (p/100)^3) â validated by capture
910 # measurement, do not "fix" this to an explicit cube.
911 if self._sink is not None:
912 # explicit zero handling: no reciprocal exists, silence the sink
913 sink_pct = 0.0 if volume <= 0 else round(10000 / volume, 2)
914 try:
915 await self._sink.set_volume(sink_pct)
916 except Exception as err:
917 # fail closed: the compensation state is now unknown, so
918 # recreate sink and daemon and do NOT forward the volume â
919 # the player must not adopt a value whose compensation
920 # never applied
921 self.logger.warning(
922 "Failed to set capture sink volume (%s); recreating capture sink", err
923 )
924 await self._recover_sink()
925 return
926 await self._event_callback(
927 self._make_event(BackendEventType.VOLUME, volume=max(0, volume))
928 )
929
930 def _translate_event(self, event: SoloistEvent) -> BackendEvent:
931 """Map a raw soloist event onto the normalized BackendEvent model."""
932 data = event.data
933 if isinstance(data, SoloistAuthState):
934 # a logged-out daemon keeps advertising itself for Connect, so being
935 # signed out is just a session that ended: awaiting a first pairing,
936 # the user signing out, or another account taking the device over
937 return self._make_event(
938 BackendEventType.SESSION_ACTIVE
939 if data.logged_in and data.is_active
940 else BackendEventType.SESSION_INACTIVE
941 )
942 if isinstance(data, SoloistDeviceChanged):
943 return self._make_event(
944 BackendEventType.SESSION_ACTIVE
945 if data.is_active
946 else BackendEventType.SESSION_INACTIVE
947 )
948 if isinstance(data, SoloistPlaybackState):
949 # covers both the playback_state snapshot and the playback_changed delta
950 self._cache_uris(track=data.item, context=data.context)
951 return self._make_event(_STATUS_EVENTS.get(data.status, BackendEventType.OTHER))
952 if isinstance(data, SoloistTrackChanged) and data.item is not None:
953 self._cache_uris(track=data.item)
954 return self._make_event(BackendEventType.METADATA, metadata=_entity_metadata(data.item))
955 if isinstance(data, SoloistPositionSync):
956 return self._make_event(
957 BackendEventType.POSITION, position=data.position.position_ms // 1000
958 )
959 if isinstance(data, SoloistErrorMessage):
960 return self._make_event(BackendEventType.ERROR, error=data.message)
961 if isinstance(data, SoloistQueueChanged):
962 return self._make_event(
963 BackendEventType.QUEUE_CHANGED,
964 queue=BackendQueueState(
965 previous=_queue_entries(data.previous),
966 upcoming=_queue_entries(data.upcoming),
967 ),
968 )
969 if isinstance(data, SoloistOptionsChanged):
970 return self._make_event(
971 BackendEventType.OPTIONS_CHANGED, options=_backend_options(data.options)
972 )
973 if isinstance(data, SoloistContextChanged):
974 self._cache_uris(context=data.context)
975 # context changes, command acks and unknown events
976 return self._make_event(BackendEventType.OTHER)
977
978 def _make_event(self, event_type: BackendEventType, **fields: Any) -> BackendEvent:
979 """Build a BackendEvent carrying the latest known context/track uris."""
980 return BackendEvent(
981 event_type,
982 context_uri=self._last_context_uri,
983 track_uri=self._last_track_uri,
984 **fields,
985 )
986
987 def _cache_uris(
988 self, *, track: SoloistEntity | None = None, context: SoloistEntity | None = None
989 ) -> None:
990 """Remember the latest context/track uris seen on the event stream."""
991 if track is not None and track.uri:
992 self._last_track_uri = track.uri
993 if context is not None and context.uri:
994 self._last_context_uri = context.uri
995
996
997def _entity_metadata(item: SoloistEntity) -> BackendTrackMetadata:
998 """
999 Extract normalized track metadata from an entity's decorations.
1000
1001 Mapped from the observed 1.3.7 wire format: the artist lives under
1002 ``creators[].entity``, the album under ``parent.entity`` and artwork
1003 under ``visual_identity.cover[]`` â all traversed defensively since the
1004 decorations bag is extensible.
1005
1006 :param item: The soloist entity (track) to extract the metadata from.
1007 """
1008 decorations = item.decorations or {}
1009 playback = _as_dict(decorations.get("playback"))
1010 duration_ms = playback.get("duration_ms")
1011 return BackendTrackMetadata(
1012 track_uri=item.uri,
1013 title=_identity_title(decorations),
1014 artist=_creator_name(decorations.get("creators")),
1015 album=_nested_entity_name(decorations.get("parent")),
1016 image_url=_cover_url(decorations),
1017 duration=int(duration_ms) // 1000 if isinstance(duration_ms, int | float) else None,
1018 )
1019
1020
1021def _queue_entries(entries: list[SoloistQueueEntry]) -> list[BackendQueueEntry]:
1022 """
1023 Map soloist queue entries onto normalized queue entries.
1024
1025 Entries without a resolvable uri are skipped.
1026
1027 :param entries: The soloist queue entries (previous or upcoming listing).
1028 """
1029 result: list[BackendQueueEntry] = []
1030 for entry in entries:
1031 if entry.item is None or not entry.item.uri:
1032 continue
1033 result.append(
1034 BackendQueueEntry(
1035 uid=entry.uid,
1036 uri=entry.item.uri,
1037 source=QueueEntrySource(entry.source),
1038 name=_identity_title(entry.item.decorations or {}),
1039 )
1040 )
1041 return result
1042
1043
1044# soloist's repeat vocabulary mapped onto the MA repeat modes
1045_REPEAT_MODES: Final = {
1046 "off": RepeatMode.OFF,
1047 "context": RepeatMode.ALL,
1048 "track": RepeatMode.ONE,
1049}
1050
1051
1052def _backend_options(options: SoloistPlaybackOptions) -> BackendPlaybackOptions:
1053 """Map soloist playback options onto the normalized model."""
1054 return BackendPlaybackOptions(
1055 shuffle=options.shuffle,
1056 repeat=_REPEAT_MODES.get(options.repeat, RepeatMode.UNKNOWN),
1057 )
1058
1059
1060def _as_dict(value: Any) -> dict[str, Any]:
1061 """Return the value if it is a dict, an empty dict otherwise."""
1062 return value if isinstance(value, dict) else {}
1063
1064
1065def _as_str(value: Any) -> str | None:
1066 """Return the value if it is a non-empty string, None otherwise."""
1067 return value if isinstance(value, str) and value else None
1068
1069
1070def _identity_title(decorations: dict[str, Any]) -> str | None:
1071 """Return the display title from a decorations bag's identity."""
1072 identity = _as_dict(decorations.get("identity"))
1073 return _as_str(identity.get("name")) or _as_str(identity.get("title"))
1074
1075
1076def _entity_name(value: Any) -> str | None:
1077 """Return the display name of a nested entity-like decoration value."""
1078 if isinstance(value, str):
1079 return value or None
1080 if isinstance(value, dict):
1081 return _as_str(value.get("name")) or _as_str(_as_dict(value.get("identity")).get("name"))
1082 return None
1083
1084
1085def _nested_entity_name(value: Any) -> str | None:
1086 """Return the identity name of a ``{"entity": {...}}`` decoration (album parent)."""
1087 entity = _as_dict(_as_dict(value).get("entity"))
1088 return _entity_name(_as_dict(entity.get("decorations"))) or _entity_name(entity)
1089
1090
1091def _creator_name(value: Any) -> str | None:
1092 """Return the name of the first creator credit (``creators[].entity``)."""
1093 if isinstance(value, list) and value:
1094 return _nested_entity_name(value[0])
1095 return None
1096
1097
1098def _cover_url(decorations: dict[str, Any]) -> str | None:
1099 """
1100 Return the artwork url from ``visual_identity.cover[]``.
1101
1102 Prefers the large rendition; falls back to the last (largest) entry.
1103 """
1104 covers = _as_dict(decorations.get("visual_identity")).get("cover")
1105 if not isinstance(covers, list) or not covers:
1106 return None
1107 for entry in covers:
1108 if isinstance(entry, dict) and entry.get("size") == "large":
1109 if url := _as_str(entry.get("url")):
1110 return url
1111 for entry in reversed(covers):
1112 if isinstance(entry, dict) and (url := _as_str(entry.get("url"))):
1113 return url
1114 return None
1115