/
/
/
1"""
2Spotify Soloist playback backend for the Spotify music provider.
3
4Plays each item with its own single-track run of ``soloist``, Spotify's
5official headless client. A run renders into a private PulseAudio capture sink
6whose FIFO is read back slightly above realtime pace and handed to Music
7Assistant as that item's stream.
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
12Shared infrastructure (binary manager, WebSocket client, audio prefs, pulse
13capture) is owned by the Spotify Connect provider / core helpers and reused here.
14"""
15
16from __future__ import annotations
17
18import asyncio
19import os
20import shutil
21from contextlib import suppress
22from functools import partial
23from pathlib import Path
24from typing import TYPE_CHECKING, Final, NoReturn
25
26from music_assistant_models.enums import ContentType, MediaType
27from music_assistant_models.errors import AudioError, LoginFailed, MusicAssistantError
28from music_assistant_models.media_items import AudioFormat
29
30from music_assistant.constants import (
31 CONF_VALUE_DISABLED,
32 CONF_VALUE_ENABLED,
33 CONF_VOLUME_NORMALIZATION,
34)
35from music_assistant.helpers.process import AsyncProcess
36from music_assistant.helpers.pulse_capture import (
37 CAPTURE_CHANNELS,
38 CAPTURE_SAMPLE_RATE,
39 PipeSink,
40 get_pulse_capture_server,
41)
42from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
43from music_assistant.providers.spotify.constants import (
44 CONF_AUDIO_QUALITY,
45 CONF_SOLOIST_API_KEY,
46 CONF_SOLOIST_CONSENT,
47 CONF_SOLOIST_SESSION_DIR,
48 SOLOIST_DATA_DIR_NAME,
49 SOLOIST_DEVICE_NAME,
50)
51from music_assistant.providers.spotify.helpers import soloist_session_present
52from music_assistant.providers.spotify_connect.base import (
53 AUDIO_QUALITY_LOSSLESS,
54 spotify_source_audio_format,
55)
56from music_assistant.providers.spotify_connect.soloist import (
57 SoloistBinaryManager,
58 SoloistClient,
59 SoloistError,
60 write_audio_prefs,
61)
62from music_assistant.providers.spotify_connect.soloist.runtime import (
63 EXIT_CODE_BUILD_EXPIRED,
64 WS_ADDR_FILE,
65 WS_PORT_FILE,
66 SoloistAuthState,
67 SoloistDeviceChanged,
68 SoloistPlaybackState,
69 SoloistPositionSync,
70 SoloistTrackChanged,
71)
72
73from .base import SpotifyPlaybackBackend, StreamSupersededError
74
75if TYPE_CHECKING:
76 import logging
77 from collections.abc import AsyncGenerator
78
79 from music_assistant_models.streamdetails import StreamDetails
80
81 from music_assistant.helpers.json import SerializableType
82 from music_assistant.helpers.pulse_capture import PulseCaptureServer
83 from music_assistant.providers.spotify.provider import SpotifyProvider
84 from music_assistant.providers.spotify_connect.soloist.runtime import SoloistEvent
85
86# The capture sink delivers fixed s32le/44.1kHz/2ch PCM. Soloist decodes
87# internally and never exposes the source codec or bit depth (lossless up to
88# 24-bit fits the 32-bit container losslessly), so this is what is handed over
89# whatever the tier; what the user is shown comes from source_audio_format.
90_FRAME_BYTES: Final[int] = 4 * CAPTURE_CHANNELS
91_BYTES_PER_SECOND: Final[int] = CAPTURE_SAMPLE_RATE * _FRAME_BYTES
92
93# Bounded soloist playback cache (docs: 0 = unlimited, otherwise at least 100 MB).
94_CACHE_SIZE_MB: Final[int] = 512
95
96# The reader clocks the pipe sink, and thus how fast Spotify must deliver.
97# Both values are the result of live listening tests against the measured
98# delivery envelope (1.1x sustained is clean even on a cold cache, 1.2x+
99# starves mid-track):
100# - the sustained 1.1x banks a downstream cushion (~6s per minute) that
101# absorbs the source-less gap at the next track boundary; at exactly 1.0x
102# every boundary gap reaches the player, where timeline-synced outputs
103# drop the late audio instead of delaying it.
104# - the burst must stay SMALL: during the burst window the read is unpaced,
105# and demanding audio faster than the warming fetch pipeline can deliver
106# destabilizes it â large bursts audibly worsened track starts.
107_PACE_RATE: Final[float] = 1.1
108_PACE_BURST_S: Final[float] = 1.0
109
110_READ_CHUNK_SIZE: Final[int] = 32768
111# one wait slice on the FIFO read; state (process exit, errors) is checked between slices
112_READ_SLICE_S: Final[float] = 1.0
113# a suspended sink delivers nothing while soloist rebuffers; give recovery some room
114_STALL_TIMEOUT_S: Final[float] = 30.0
115# waiting for the daemon's WS endpoint, events and the requested item to appear
116_STARTUP_TIMEOUT_S: Final[float] = 30.0
117# How long a jump on a live session is given to land. Generous against the
118# milliseconds the engine actually takes, but short enough that a jump which
119# will not land still leaves the queue patience for the fresh session that
120# serves the item instead.
121_JUMP_TIMEOUT_S: Final[float] = 5.0
122# how often to check whether the events task has the WebSocket up yet
123_CONNECT_POLL_S: Final[float] = 0.05
124_SEEK_CONFIRM_TIMEOUT_S: Final[float] = 15.0
125# a seek is re-sent at this interval until a position anchor confirms it
126_SEEK_RETRY_INTERVAL_S: Final[float] = 0.5
127# position reported by a seek anchor may fall slightly before the requested target
128_SEEK_TOLERANCE_MS: Final[int] = 2000
129# Infrastructure silence precedes the session's first decoded sample; trim at
130# most this much, once per session. Kept small on purpose: the trim cannot tell
131# capture pre-roll from a genuinely digitally-silent intro, so the budget bounds
132# what an intro can lose while still covering the measured pre-roll (~140 ms).
133_MAX_LEAD_TRIM_S: Final[float] = 0.5
134# below this much audio an item rendered nothing: the engine either refused it
135# (clean exit) or never got going, and anything above it played and was cut short
136_REFUSED_DELIVERY_MS: Final[int] = 1000
137# what the lead trim can take off the start of an otherwise complete delivery
138_MAX_LEAD_TRIM_MS: Final[int] = int(_MAX_LEAD_TRIM_S * 1000)
139# The elastic cushion between the paced FIFO reader and the consumer. A full
140# cushion suspends the capture sink, which pauses the engine: the FIFO itself
141# holds well under a second, so the reader must keep draining it whenever the
142# consumer (the item's buffer, at its memory-tiered capacity) stops taking audio.
143_RUN_CUSHION_S: Final[float] = 10.0
144# how close to its target a position report has to come to confirm a seek
145_SEEK_CONFIRM_GRACE_MS: Final[int] = 3000
146
147# The sink renders exact-zero padding while the engine idles between items, and the
148# item-change event arrives after those frames landed in the outgoing item's channel.
149# Zero frames past a short grace, inside the item's own tail zone, are that padding.
150_TAIL_PAD_ZONE_S: Final[float] = 10.0
151_TAIL_PAD_GRACE_S: Final[float] = 1.0
152# The engine allows one daemon per data directory and refuses to start otherwise,
153# exiting with a plain code 1 - its message is the only way to tell that case
154# apart from any other startup failure.
155_DATA_DIR_BUSY_MARKER: Final[str] = "another session is running"
156# A daemon that cannot log in advertises itself for pairing instead of failing,
157# and then sits there until the startup budget runs out. The engine reports no
158# other way that a stored session is gone.
159_UNPAIRED_MARKER: Final[str] = "waiting for login"
160# how long the log reader is given to catch up on a daemon's parting words
161_LOG_DRAIN_TIMEOUT_S: Final[float] = 2.0
162
163
164class SoloistSessionBusyError(ProviderStreamLimitError):
165 """
166 Raised when the one Soloist run is delivering a different item.
167
168 A ProviderStreamLimitError so a speculative prepare gives up softly and the
169 item is not marked unplayable, but with a message of its own: "already
170 playing something else" says more than the slot-budget phrasing.
171 """
172
173 def __init__(self, provider: MusicProvider) -> None:
174 """
175 Initialize the error.
176
177 :param provider: The provider whose session is busy.
178 """
179 # deliberately skips ProviderStreamLimitError.__init__, whose whole job is
180 # to phrase the message in terms of that source-stream budget
181 MusicAssistantError.__init__(
182 self,
183 f"{provider.name} is already playing something else",
184 translation_key="soloist_session_busy",
185 translation_owner="provider.spotify",
186 translation_args=[provider.name],
187 )
188 self.provider_instance = provider.instance_id
189 self.limit = 1
190
191
192class SoloistBackend(SpotifyPlaybackBackend):
193 """
194 Fetches Spotify audio with one ``soloist --single-track`` engine run per item.
195
196 Requires a stored paired session in the per-instance data directory
197 (provisioned by the setup flow via ``soloist --pair``).
198 """
199
200 _server: PulseCaptureServer | None = None
201 _binary: Path | None = None
202 _run: _SingleTrackRun | None = None
203
204 def __init__(self, provider: SpotifyProvider) -> None:
205 """
206 Initialize the backend.
207
208 :param provider: The owning Spotify provider instance.
209 """
210 super().__init__(provider)
211 # Guards every write of _run AND every run teardown.
212 # The engine allows one daemon per data directory, so a replacement can
213 # only be spawned once the previous one is gone â holding this across
214 # the teardown is what sequences that.
215 self._run_lock = asyncio.Lock()
216
217 def source_audio_format(self, media_type: MediaType) -> AudioFormat:
218 """
219 Return the format Spotify is asked to stream for this item.
220
221 The engine decodes internally and never reports what it fetched, so this
222 is the configured ceiling rather than a measurement â the same thing the
223 Spotify apps show. Only music is served losslessly; spoken content is
224 Ogg Vorbis whatever the setting says.
225
226 :param media_type: What is being streamed.
227 """
228 quality = self._audio_quality
229 return spotify_source_audio_format(
230 quality,
231 lossless=media_type == MediaType.TRACK and quality == AUDIO_QUALITY_LOSSLESS,
232 )
233
234 @property
235 def handoff_audio_format(self) -> AudioFormat:
236 """Return the PCM the capture sink actually delivers."""
237 return AudioFormat(
238 content_type=ContentType.PCM_S32LE,
239 codec_type=ContentType.PCM_S32LE,
240 sample_rate=CAPTURE_SAMPLE_RATE,
241 bit_depth=32,
242 channels=CAPTURE_CHANNELS,
243 )
244
245 @property
246 def is_realtime(self) -> bool:
247 """Soloist delivers at playback pace (~1.1x ceiling): no read-ahead."""
248 return True
249
250 def session_normalizes(self, streamdetails: StreamDetails) -> bool | None:
251 """
252 Return whether the session serving this item's queue is normalizing.
253
254 None when no session serves that queue, in which case the configuration is
255 the only thing to go on.
256
257 :param streamdetails: Stream details of the item being asked about.
258 """
259 run = self._run_for(streamdetails)
260 return run.engine_normalizes if run is not None else None
261
262 async def setup(self) -> None:
263 """
264 Validate the binary and paired session, and start the capture server.
265
266 :raises LoginFailed: When the API key or paired session is missing, which
267 requires the user to re-run the setup flow.
268 """
269 if not self._api_key:
270 raise LoginFailed(
271 "Spotify Soloist API key missing",
272 translation_key="soloist_pairing_required",
273 translation_owner="provider.spotify",
274 )
275 # setup errors (unsupported platform, missing consent, download failure,
276 # expired build) propagate so the provider load fails with a clear error
277 manager = SoloistBinaryManager(self.mass)
278 self._binary = await manager.ensure_fresh(self._consent)
279 await self._adopt_paired_session()
280 if not await asyncio.to_thread(self._has_stored_session):
281 raise LoginFailed(
282 "Spotify Soloist is not paired with a Spotify account",
283 translation_key="soloist_pairing_required",
284 translation_owner="provider.spotify",
285 )
286 await asyncio.to_thread(self._cache_dir.mkdir, parents=True, exist_ok=True)
287 self._server = await get_pulse_capture_server(self.mass).acquire()
288
289 async def unload(self) -> None:
290 """Stop the run and release the capture server."""
291 async with self._run_lock:
292 if (run := self._run) is not None:
293 # dropped only once the teardown finished, so a cancellation
294 # part-way leaves a later stop() something to clean up
295 await run.stop()
296 self._run = None
297 if (server := self._server) is not None:
298 self._server = None
299 await server.release()
300
301 async def stream_spotify_uri(
302 self,
303 spotify_uri: str,
304 seek_position: int = 0,
305 *,
306 streamdetails: StreamDetails | None = None,
307 continuation: bool = False,
308 ) -> AsyncGenerator[bytes]:
309 """
310 Yield the PCM audio for one Spotify URI as its own single-track run.
311
312 :param spotify_uri: Canonical Spotify URI (``spotify:track:<id>`` or
313 ``spotify:episode:<id>``).
314 :param seek_position: Position in seconds to start from.
315 :param streamdetails: The StreamDetails this audio is requested for.
316 :param continuation: Ignored: every URI is its own engine run, so a
317 later chapter needs no special handling.
318 """
319 if self._server is None or self._binary is None:
320 raise AudioError("Spotify Soloist backend is not started")
321 run = await self._acquire_run(
322 spotify_uri, seek_position, streamdetails, continuation=continuation
323 )
324 try:
325 async for chunk in run.stream():
326 yield chunk
327 finally:
328 await run.stop()
329 async with self._run_lock:
330 if self._run is run:
331 self._run = None
332
333 async def discard_run(self, run: _SingleTrackRun) -> None:
334 """
335 Stop a run for good, dropping it if it is still the current one.
336
337 The teardown happens under the run lock, not after it: the engine
338 refuses to start while another daemon still holds its data directory, so
339 a replacement must not be spawned until this one is gone.
340
341 :param run: The run to tear down.
342 """
343 async with self._run_lock:
344 await run.stop()
345 if self._run is run:
346 self._run = None
347
348 async def get_diagnostics(self) -> dict[str, SerializableType]:
349 """Return diagnostic details about the backend (never any secret)."""
350 run = self._run
351 return {
352 "soloist": SoloistBinaryManager(self.mass).diagnostics(),
353 "paired": await asyncio.to_thread(self._has_stored_session),
354 "session_active": run is not None,
355 }
356
357 async def _acquire_run(
358 self,
359 spotify_uri: str,
360 seek_position: int,
361 streamdetails: StreamDetails | None,
362 *,
363 continuation: bool,
364 ) -> _SingleTrackRun:
365 """
366 Start the engine run for this item, replacing one this request supersedes.
367
368 The engine allows one daemon per data directory (and Spotify one stream
369 per account), so a run still serving another stream is reported as
370 capacity: a speculative prepare gives up softly and the real request,
371 made once that stream has been released, gets the slot.
372
373 :raises StreamSupersededError: When a continuation finds a run for the
374 same item already started by the stream that replaced it.
375 """
376 media_key = streamdetails.uri if streamdetails is not None else None
377 waited = False
378 while True:
379 async with self._run_lock:
380 if (run := self._run) is None:
381 # cheap thanks to the shared verify cache; swaps in a fresh build
382 # when the installed one is nearing its 90-day expiry
383 try:
384 self._binary = await SoloistBinaryManager(self.mass).ensure_fresh(
385 self._consent
386 )
387 except SoloistError as err:
388 raise AudioError(f"Spotify Soloist binary unavailable: {err}") from err
389 fresh = _SingleTrackRun(self, spotify_uri, seek_position * 1000, streamdetails)
390 try:
391 await fresh.start()
392 except BaseException:
393 await fresh.stop()
394 raise
395 self._run = fresh
396 return fresh
397 if run.media_key != media_key or media_key is None:
398 raise SoloistSessionBusyError(self.provider)
399 if continuation and run.spotify_uri != spotify_uri:
400 # this stream was replaced (a seek of the same item started a
401 # fresh run) before it could continue into its next chapter -
402 # a run it must not take back
403 raise StreamSupersededError(f"The stream of {spotify_uri} was replaced")
404 if seek_position:
405 # a positive seek can only target the item being delivered (a
406 # prefetch never seeks): the run restarts at the target
407 await run.stop()
408 self._run = None
409 continue
410 if waited:
411 raise SoloistSessionBusyError(self.provider)
412 # Same item, no seek, run still held. A restart-from-zero races the
413 # release of the stream it replaces, so give that release a moment -
414 # but never steal a held run: a second queue occurrence of the same
415 # track asks with these same details, and stopping its playing twin
416 # would cut it mid-track.
417 waited = True
418 await self._wait_run_released(run)
419
420 async def _wait_run_released(self, run: _SingleTrackRun, timeout: float = 2.0) -> None:
421 """Wait briefly for the given run to be released by the stream holding it."""
422 deadline = asyncio.get_running_loop().time() + timeout
423 while self._run is run and asyncio.get_running_loop().time() < deadline:
424 await asyncio.sleep(0.05)
425
426 @property
427 def _api_key(self) -> str:
428 """Return the stored Soloist API key."""
429 return str(self.provider.get_setup_value(CONF_SOLOIST_API_KEY) or "")
430
431 @property
432 def _audio_quality(self) -> str:
433 """
434 Return the configured streaming quality ceiling.
435
436 Stated rather than left to the engine's own default, which would
437 otherwise decide it silently. Spotify serves the best the account is
438 entitled to below the ceiling.
439 """
440 return str(self.provider.config.get_value(CONF_AUDIO_QUALITY) or AUDIO_QUALITY_LOSSLESS)
441
442 @property
443 def _consent(self) -> bool:
444 """Return whether the user consented to downloading the binary."""
445 return bool(self.provider.get_setup_value(CONF_SOLOIST_CONSENT))
446
447 @property
448 def _data_dir(self) -> Path:
449 """Return the per-instance soloist data directory (paired session)."""
450 return (
451 Path(self.mass.storage_path)
452 / "spotify"
453 / self.provider.instance_id
454 / SOLOIST_DATA_DIR_NAME
455 )
456
457 @property
458 def _cache_dir(self) -> Path:
459 """Return the per-instance soloist playback cache directory."""
460 return Path(self.mass.cache_path) / self.provider.instance_id / "soloist-cache"
461
462 @property
463 def _sink_prefix(self) -> str:
464 """Return the capture sink name prefix (PA-safe form of the instance id)."""
465 return "".join(
466 ch if ch.isalnum() or ch in "_.-" else "_" for ch in self.provider.instance_id
467 )
468
469 def _run_for(self, streamdetails: StreamDetails) -> _SingleTrackRun | None:
470 """
471 Return the run serving this very item, if one is.
472
473 :param streamdetails: Stream details of the item being asked about.
474 """
475 run = self._run
476 if run is None or run.media_key != streamdetails.uri:
477 return None
478 return run
479
480 async def _adopt_paired_session(self) -> None:
481 """Adopt a session paired by the setup flow into the per-instance data dir."""
482 pending = str(self.provider.get_setup_value(CONF_SOLOIST_SESSION_DIR) or "")
483 if not pending:
484 return
485 await asyncio.to_thread(self._copy_paired_session, pending)
486 # config writes schedule tasks and must therefore run on the event loop
487 self.provider._update_setup_data(CONF_SOLOIST_SESSION_DIR, None)
488
489 def _copy_paired_session(self, pending: str) -> None:
490 """
491 Copy the paired session files into the canonical data dir (blocking).
492
493 A copy (not a move) so the flow-private source survives a failed
494 provider load: the setup flow can then retry its finish step and adopt
495 the same pairing again. The source is removed when the flow ends.
496 """
497 source = Path(self.mass.storage_path) / pending
498 canonical = self._data_dir
499 if source.is_dir() and source != canonical:
500 # stage first: a failed copy must never leave the canonical dir
501 # half-written or destroy the currently working session
502 staging = canonical.with_name(canonical.name + ".new")
503 shutil.rmtree(staging, ignore_errors=True)
504 canonical.parent.mkdir(parents=True, exist_ok=True)
505 shutil.copytree(source, staging)
506 staging.chmod(0o700)
507 if canonical.exists():
508 shutil.rmtree(canonical)
509 staging.replace(canonical)
510
511 def _has_stored_session(self) -> bool:
512 """Return whether the data dir holds a stored (paired) session (blocking)."""
513 return soloist_session_present(self._data_dir)
514
515 def _prepare_data_dir(self, *, normalize: bool) -> None:
516 """
517 Prepare the data dir for a fresh session spawn (blocking).
518
519 :param normalize: Whether the engine should normalize loudness itself.
520 """
521 self._data_dir.mkdir(parents=True, exist_ok=True)
522 self._data_dir.chmod(0o700)
523 # endpoint files from a previous run would point the client at a dead port
524 for endpoint_file in (WS_ADDR_FILE, WS_PORT_FILE):
525 (self._data_dir / endpoint_file).unlink(missing_ok=True)
526 # The engine reads its prefs at startup only, so they are refreshed on
527 # every spawn. Whoever normalizes, only one of us may: with the engine
528 # doing it the provider declares the audio pre-normalized, which takes
529 # MA's own normalization out of the path (see
530 # SpotifyProvider.delivers_normalized_audio).
531 # crossfade 0: Music Assistant mixes the queue's crossfade itself, so the
532 # engine plays every track clean from its first sample and an item's
533 # delivered audio lines up with its analysis (waveform, beat grid, light sync)
534 if not write_audio_prefs(
535 self._data_dir,
536 self.logger,
537 crossfade_ms=0,
538 loudness_normalization=normalize,
539 audio_quality=self._audio_quality,
540 ):
541 # the provider has told the rest of the server who normalizes this
542 # audio; running the engine on settings that may say otherwise would
543 # mean normalizing twice, or not at all
544 raise AudioError("Spotify Soloist audio settings could not be applied")
545
546 def _session_args(self, spotify_uri: str) -> list[str]:
547 """
548 Build the argv for one single-track engine run.
549
550 SECURITY: the argv carries the user's API key â it must never be logged
551 or end up in any error message.
552 """
553 assert self._binary is not None
554 return [
555 str(self._binary),
556 # play exactly this item and exit when it finishes: the stored
557 # session is restored without advertising a Spotify Connect device,
558 # and shuffle/repeat start disabled
559 "--single-track",
560 spotify_uri,
561 # required by the binary even though single-track mode never
562 # advertises it anywhere
563 "--device-name",
564 SOLOIST_DEVICE_NAME,
565 "--api-key",
566 self._api_key,
567 "--data-dir",
568 str(self._data_dir),
569 "--cache-dir",
570 str(self._cache_dir),
571 # bounded playback cache (0 would be unlimited)
572 "--cache-size",
573 str(_CACHE_SIZE_MB),
574 # unity volume so unaltered PCM reaches the capture sink
575 "--initial-volume",
576 "100",
577 # local WebSocket API on a free loopback port; the daemon publishes
578 # the actual endpoint in its data dir where SoloistClient finds it
579 "--ws",
580 "127.0.0.1:0",
581 ]
582
583
584class _SingleTrackRun:
585 """
586 One engine run playing exactly one Spotify URI, streamed as it renders.
587
588 Single-track mode starts the stored session without advertising a Spotify
589 Connect device, plays the one URI with shuffle and repeat off, and exits
590 when the item finishes. The capture FIFO is reader-clocked: it is read at
591 ``_PACE_RATE`` into a small bounded cushion whose backpressure suspends the
592 capture sink, which is what pauses the engine when the consumer stops
593 taking audio.
594 """
595
596 def __init__(
597 self,
598 backend: SoloistBackend,
599 spotify_uri: str,
600 seek_position_ms: int,
601 streamdetails: StreamDetails | None,
602 ) -> None:
603 """
604 Initialize a run (nothing is spawned yet).
605
606 :param backend: The owning backend.
607 :param spotify_uri: Canonical Spotify URI the engine is to play.
608 :param seek_position_ms: Position to start from, in milliseconds.
609 :param streamdetails: The StreamDetails the audio is requested for.
610 """
611 self.backend = backend
612 self.mass = backend.mass
613 self.logger: logging.Logger = backend.logger
614 self.spotify_uri = spotify_uri
615 self.media_key = streamdetails.uri if streamdetails is not None else None
616 self.queue_id = streamdetails.queue_id if streamdetails is not None else None
617 # what the engine was actually told at spawn, which is what the streams
618 # core has to agree with - the setting may be toggled while this plays
619 self.engine_normalizes = False
620 self._seek_target_ms = seek_position_ms
621 self._duration_ms: int | None = (
622 streamdetails.duration * 1000
623 if streamdetails is not None and streamdetails.duration
624 else None
625 )
626 self._client: SoloistClient | None = None
627 self._sink: PipeSink | None = None
628 self._proc: AsyncProcess | None = None
629 self._tasks: list[asyncio.Task[None]] = []
630 self._log_task: asyncio.Task[None] | None = None
631 self._transport: asyncio.ReadTransport | None = None
632 self._reader: asyncio.StreamReader | None = None
633 self._error: str | None = None
634 self._logged_in: bool | None = None
635 self._data_dir_busy = False
636 self._unpaired = False
637 self._stopped = False
638 self._item_over = False
639 self._engine_exited = False
640 self._teardown_done = False
641 self._engine_playing = False
642 self._sink_running = False
643 self._sink_lock = asyncio.Lock()
644 # the engine reported it is playing this run's uri
645 self._started = asyncio.Event()
646 self._seek_confirmed = asyncio.Event()
647 self._position_ms: int | None = None
648 # the elastic cushion between the paced reader and the consumer; a full
649 # cushion suspends the sink, which pauses the engine
650 cushion_chunks = max(2, int(_RUN_CUSHION_S * _BYTES_PER_SECOND / _READ_CHUNK_SIZE))
651 self._chunks: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=cushion_chunks)
652 self._delivery_done = False
653 self._delivered = 0
654 self._read_bytes = 0
655 self._tail_zeros = 0
656
657 async def start(self) -> None:
658 """Spawn the engine on this run's URI and get its audio flowing."""
659 backend = self.backend
660 server = backend._server
661 assert server is not None
662 assert backend._binary is not None
663 self.engine_normalizes = self._engine_normalization_enabled()
664 await asyncio.to_thread(
665 partial(backend._prepare_data_dir, normalize=self.engine_normalizes)
666 )
667 self._sink = sink = await PipeSink.create(server, backend._sink_prefix)
668 # unity gain so the FIFO carries the engine's PCM unaltered; the sink
669 # stays suspended until the (seeked) item is ready so no infrastructure
670 # silence accumulates
671 await sink.set_volume(100)
672 await sink.suspend()
673 self._proc = proc = AsyncProcess(
674 backend._session_args(self.spotify_uri),
675 # the daemon writes all of its logging to stdout and only ever puts
676 # argument-parsing complaints on stderr, so the two are merged into
677 # one captured stream. Capturing is what makes the redaction below
678 # reachable at all: an unset stdout is inherited, which would leak
679 # the daemon's output straight to the server console instead.
680 stdout=True,
681 stderr=asyncio.subprocess.STDOUT,
682 # the explicit process name keeps AsyncProcess logging free of the
683 # argv (which carries the API key)
684 name=f"soloist[{backend.provider.name}]",
685 env=server.child_env(sink.sink_name),
686 )
687 await proc.start()
688 # kept out of _tasks so stop() does not cancel it before the daemon has
689 # exited, but a reader that dies still has to fail the run: nothing
690 # else drains stdout and the daemon would block on a full pipe
691 self._log_task = asyncio.create_task(self._log_output(proc))
692 self._log_task.add_done_callback(self._task_done)
693 self._client = client = SoloistClient(self.mass, backend._data_dir, self.logger)
694 client_ready = asyncio.Event()
695 self._spawn_task(self._run_events(client, client_ready))
696 # Commands travel over the events connection, and the engine takes them
697 # in three stages: it publishes its endpoint, then accepts a connection,
698 # then restores its session and logs in. A failure reported while the
699 # endpoint is still awaited - the engine having no session to log in
700 # with - is watched for throughout.
701 try:
702 async with asyncio.timeout(_STARTUP_TIMEOUT_S):
703 while not self._error and not (
704 client_ready.is_set() and client.connected and self._logged_in
705 ):
706 await asyncio.sleep(_CONNECT_POLL_S)
707 except TimeoutError:
708 self._raise_startup_error("could not connect and sign in")
709 if self._error or not client.connected:
710 self._raise_startup_error("did not start up correctly")
711 await self._await_playback_started(proc)
712 if self._seek_target_ms:
713 await self._cold_seek(client, self._seek_target_ms)
714 # the reader must be attached before the sink starts producing, or the
715 # sink's first writes go to a reader-less FIFO and are dropped
716 self._reader, self._transport = await _open_fifo_reader(sink.fifo_path)
717 self._spawn_task(self._watch_exit(proc))
718 self._spawn_task(self._read_capture())
719 await self._set_sink(running=True)
720
721 async def stream(self) -> AsyncGenerator[bytes]:
722 """Yield the run's PCM audio, ending where the engine ended the item."""
723 while True:
724 # the flag ends a delivery whose sentinel found the cushion full:
725 # only an empty cushion can leave the consumer blocked below, and
726 # an empty cushion always has room for the sentinel
727 if self._delivery_done and self._chunks.empty():
728 break
729 if (chunk := await self._chunks.get()) is None:
730 break
731 self._delivered += len(chunk)
732 yield chunk
733 if self._error is not None:
734 raise AudioError(self._error)
735 self._validate_delivery()
736
737 async def stop(self) -> None:
738 """
739 Tear the run down: stop the daemon, the reader and the capture sink.
740
741 Safe to call again after a cancelled teardown: every step is idempotent
742 and the run is only marked torn down once they have all run, so a
743 cancellation part-way cannot leave the daemon or the sink behind.
744 """
745 if self._teardown_done:
746 return
747 self._stopped = True
748 self._release_waiters()
749 if (self._engine_exited or self._item_over) and (proc := self._proc) is not None:
750 # The engine exits on its own at its item's end (a wander event can
751 # precede the exit by a moment): reap it before anything else, so
752 # close() below finds a returncode and returns right away. Closing
753 # an unreaped process instead silently waits out its stream-lock
754 # and flush budgets - ten seconds on every natural track end.
755 with suppress(Exception):
756 await asyncio.wait_for(proc.wait(), 5)
757 await _cancel_and_join(self._tasks)
758 self._tasks.clear()
759 if self._transport is not None:
760 self._transport.close()
761 self._transport = None
762 if (proc := self._proc) is not None:
763 # Closed straight away, with no grace period for a natural exit: on
764 # an aborted stream the engine is mid-item and never quits on its
765 # own; after a natural end it has already exited and close() only
766 # reaps it. The log reader stays alive across the close: nothing
767 # else drains the daemon's stdout, and a full pipe would keep it
768 # from exiting. A forced close must never be judged by its exit code.
769 with suppress(Exception):
770 await proc.close()
771 if proc.returncode is None:
772 # close() has exhausted its kill attempts, so nothing here can do
773 # better; the daemon keeps the data directory and the next spawn
774 # reports it as busy
775 self.logger.warning("The Spotify Soloist daemon could not be stopped")
776 # dropped only now: a cancellation during the awaits above must leave
777 # the retry something to close, or the daemon keeps the data
778 # directory and every later run is refused
779 self._proc = None
780 if self._log_task is not None:
781 await _cancel_and_join([self._log_task])
782 self._log_task = None
783 if (sink := self._sink) is not None:
784 with suppress(Exception):
785 await sink.unload()
786 self._sink = None
787 self._teardown_done = True
788
789 # ---- internals ----
790
791 def _spawn_task(self, coro: object) -> None:
792 """Track a run-scoped task so stop() can cancel and join it."""
793 task: asyncio.Task[None] = asyncio.create_task(coro) # type: ignore[arg-type]
794 task.add_done_callback(self._task_done)
795 self._tasks.append(task)
796
797 def _task_done(self, task: asyncio.Task[None]) -> None:
798 """Fail the run when one of its tasks died of an unexpected error."""
799 if task.cancelled() or (err := task.exception()) is None:
800 return
801 self.logger.error("Spotify Soloist task failed: %s", err, exc_info=err)
802 self._fail(f"Spotify playback stopped unexpectedly: {err}")
803
804 def _fail(self, message: str) -> None:
805 """Record a fatal error, unblock the consumer and tear the run down."""
806 if self._error is not None or self._stopped:
807 return
808 self._error = message
809 self._release_waiters()
810 # the teardown runs as its own task: it cancels the very tasks this is
811 # called from, and the daemon has to go either way
812 self.mass.create_task(self.backend.discard_run, self)
813
814 def _release_waiters(self) -> None:
815 """Unblock everything waiting on this run: startup, seek and the consumer."""
816 self._started.set()
817 self._seek_confirmed.set()
818 if self._item_over and self._error is None:
819 # a cleanly ended run has marked its delivery done; the consumer is
820 # still entitled to the cushioned tail and ends once it drains
821 return
822 # a failed or aborted run's cushion holds audio nobody will take anymore;
823 # the sentinel has to reach the consumer either way
824 while True:
825 try:
826 self._chunks.get_nowait()
827 except asyncio.QueueEmpty:
828 break
829 with suppress(asyncio.QueueFull):
830 self._chunks.put_nowait(None)
831
832 def _validate_delivery(self) -> None:
833 """
834 Raise when the engine died on this item, or played none of it.
835
836 Audio that arrives and then stops belongs to something ending the item
837 early - a skip, a seek or the account playing elsewhere - so a partial
838 delivery is only a failure when the engine itself did not end cleanly.
839 """
840 if self._stopped or self._duration_ms is None:
841 return
842 proc = self._proc
843 exit_code = proc.returncode if proc is not None else None
844 if self._engine_exited and exit_code not in (None, 0):
845 # audio may well have arrived first, but the engine did not end the
846 # item: it died on it, whatever the queue got to play
847 raise AudioError(
848 f"Spotify playback stopped unexpectedly (engine exit code {exit_code})"
849 )
850 played_ms = self._delivered / _BYTES_PER_SECOND * 1000
851 expected_ms = self._duration_ms - self._seek_target_ms
852 if (
853 played_ms > _REFUSED_DELIVERY_MS
854 # a seek can leave an item with almost nothing left to play
855 or expected_ms <= _REFUSED_DELIVERY_MS
856 # a complete delivery arrives short by what the lead trim took off it
857 or played_ms + _MAX_LEAD_TRIM_MS >= expected_ms
858 ):
859 return
860 if self._engine_exited:
861 raise AudioError(
862 "Spotify would not play this track: it is unavailable for this account "
863 "or region, or Spotify refused it for now"
864 )
865 raise AudioError("Spotify stopped before it played any of this track")
866
867 def _engine_normalization_enabled(self) -> bool:
868 """
869 Return whether the engine should normalize the loudness it delivers.
870
871 The player's own volume normalization switch decides first: turning it
872 off means nobody normalizes, not that the job passes to Spotify.
873 """
874 if not self.backend.provider.spotify_normalization_configured:
875 return False
876 if self.queue_id is None:
877 # nothing to read the switch from, so the provider option stands
878 return True
879 return (
880 self.mass.config.get_effective_player_queue_config_value(
881 self.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
882 )
883 != CONF_VALUE_DISABLED
884 )
885
886 async def _await_playback_started(self, proc: AsyncProcess) -> None:
887 """Wait until the engine reports it is playing this run's item."""
888 try:
889 async with asyncio.timeout(_STARTUP_TIMEOUT_S):
890 exit_task = asyncio.ensure_future(proc.wait())
891 started_task = asyncio.ensure_future(self._started.wait())
892 try:
893 await asyncio.wait(
894 {exit_task, started_task}, return_when=asyncio.FIRST_COMPLETED
895 )
896 finally:
897 exit_task.cancel()
898 started_task.cancel()
899 except TimeoutError:
900 self._raise_startup_error("did not start playing in time")
901 if self._error or not self._started.is_set():
902 if proc.returncode is not None:
903 # let the log reader catch up, so the daemon's own complaint can
904 # be reported instead of a generic startup failure
905 await self._drain_log()
906 if proc.returncode == EXIT_CODE_BUILD_EXPIRED:
907 # an expired build exits with code 10 right at spawn
908 await self._handle_expired_build()
909 self._raise_startup_error("stopped before it started playing")
910
911 async def _drain_log(self) -> None:
912 """Give the log reader a moment to deliver the daemon's parting words."""
913 if self._log_task is not None:
914 with suppress(TimeoutError):
915 async with asyncio.timeout(_LOG_DRAIN_TIMEOUT_S):
916 await asyncio.shield(self._log_task)
917
918 async def _handle_expired_build(self) -> NoReturn:
919 """Replace the expired soloist build and fail the item with an accurate message."""
920 try:
921 # bypass the verify cache â it would hand back the same expired binary
922 self.backend._binary = await SoloistBinaryManager(self.mass).ensure_fresh(
923 self.backend._consent, force=True
924 )
925 except SoloistError as err:
926 raise AudioError(
927 "Spotify Soloist build expired and no replacement could be installed"
928 ) from err
929 raise AudioError("Spotify Soloist build expired; a replacement was installed, retry")
930
931 def _raise_startup_error(self, detail: str) -> NoReturn:
932 """Raise the most specific startup failure for the requested item."""
933 # a pairing that never logged in is checked first: it also fails the
934 # run, and its recovery (back through the setup flow) beats failing
935 # every track with a generic error
936 if self._unpaired or self._logged_in is False:
937 # the stored session no longer logs in: route the user through the
938 # setup flow instead of failing every item (mirrors librespot's
939 # INVALID_CREDENTIALS handling)
940 error = LoginFailed(
941 "Spotify Soloist pairing lost",
942 translation_key="soloist_pairing_required",
943 translation_owner="provider.spotify",
944 )
945 provider = self.backend.provider
946 if provider.available:
947 provider.unload_with_error(error)
948 raise error
949 if self._data_dir_busy:
950 # a daemon from an earlier Music Assistant process is still holding
951 # this provider's data directory; nothing here can reach it
952 raise AudioError(
953 "Another Spotify Soloist session is still running for this provider "
954 "and has to be stopped first (restarting Music Assistant clears it)"
955 )
956 if self._error:
957 raise AudioError(self._error)
958 raise AudioError(f"Spotify {detail}")
959
960 async def _cold_seek(self, client: SoloistClient, target_ms: int) -> None:
961 """
962 Seek the engine to the target position before any PCM is released.
963
964 The sink is still suspended, so no pre-seek audio enters the FIFO; PCM
965 demand only starts once a position report confirms the seek landed.
966 """
967 # the engine silently drops a seek that arrives while the track is still
968 # loading (verified via event trace), so re-send it until a position
969 # report confirms it landed
970 deadline = asyncio.get_running_loop().time() + _SEEK_CONFIRM_TIMEOUT_S
971 while True:
972 await client.seek(target_ms)
973 with suppress(TimeoutError):
974 async with asyncio.timeout(_SEEK_RETRY_INTERVAL_S):
975 await self._seek_confirmed.wait()
976 if self._seek_confirmed.is_set():
977 if self._error:
978 raise AudioError(f"Spotify Soloist failed: {self._error}")
979 return
980 if asyncio.get_running_loop().time() >= deadline:
981 raise AudioError(f"Spotify Soloist did not confirm seeking to {target_ms}ms")
982
983 async def _log_output(self, proc: AsyncProcess) -> None:
984 """Log the daemon's output with the API key redacted."""
985 api_key = self.backend._api_key
986 async for line in proc.iter_stdout():
987 # the third-party binary's own output may echo argv (which carries
988 # the api key), so redact it before logging
989 text = line.replace(api_key, "<redacted>") if api_key else line
990 if _DATA_DIR_BUSY_MARKER in text:
991 self._data_dir_busy = True
992 if _UNPAIRED_MARKER in text and not self._unpaired:
993 await self._check_pairing_lost()
994 self.logger.debug("[soloist] %s", text)
995
996 async def _check_pairing_lost(self) -> None:
997 """
998 Fail the run when the engine has no stored session left to log in with.
999
1000 The engine reports being unpaired while it is still restoring a session
1001 too, so its report is confirmed against the stored session: acting on
1002 it alone would fail every playback on a perfectly good pairing.
1003 """
1004 if await asyncio.to_thread(self.backend._has_stored_session):
1005 return
1006 self._unpaired = True
1007 self._fail("Spotify is no longer signed in and has to be paired again")
1008
1009 async def _watch_exit(self, proc: AsyncProcess) -> None:
1010 """
1011 Record the daemon's exit the moment it happens.
1012
1013 The engine exits when its item finishes, and the sink renders silence from
1014 then on: the reader needs a signal that does not depend on the process
1015 being reaped, or the item's end is only found by wading through that
1016 silence - the tail-zone budget late, every track.
1017 """
1018 await proc.wait()
1019 self._engine_exited = True
1020
1021 async def _read_capture(self) -> None:
1022 """
1023 Read the capture FIFO for the run's whole life and cushion it for the consumer.
1024
1025 The pace is the run's clock: the pipe sink applies no rate limit of its
1026 own, so how fast this reads is how fast the engine plays. Reading
1027 slightly above realtime is what banks the lead a boundary's crossfade
1028 uses; reading unpaced makes PulseAudio render silence instead of
1029 applying backpressure, and the engine runs off the end of its content.
1030 """
1031 reader = self._reader
1032 proc = self._proc
1033 assert reader is not None
1034 assert proc is not None
1035 loop = asyncio.get_running_loop()
1036 shaper = _CaptureShaper()
1037 pace_anchor: float | None = None
1038 paced_bytes = 0
1039 stalled_for = 0.0
1040 while not self._stopped:
1041 if self._item_over:
1042 # the engine moved on: everything after this point is the next
1043 # (autoplayed) track's audio, not this item's
1044 self._finish_delivery()
1045 return
1046 try:
1047 chunk = await asyncio.wait_for(reader.read(_READ_CHUNK_SIZE), _READ_SLICE_S)
1048 except TimeoutError:
1049 if self._engine_exited:
1050 # the engine exited and its tail has drained: the item is over
1051 self._finish_delivery()
1052 return
1053 # No data. Either the sink is suspended on purpose (the engine
1054 # is paused or the cushion is at its cap) or the run has died;
1055 # only the former is fine.
1056 if self._sink_running:
1057 stalled_for += _READ_SLICE_S
1058 if stalled_for >= _STALL_TIMEOUT_S:
1059 self._fail("Spotify stopped sending audio")
1060 return
1061 # Restart the pacing clock rather than carry the gap: making up
1062 # lost time would mean an unpaced burst, which over-demands the
1063 # engine's fetch pipeline exactly when it is recovering.
1064 pace_anchor = None
1065 continue
1066 stalled_for = 0.0
1067 if not chunk:
1068 # writer end closed: the capture sink is gone (pulse restart)
1069 self._fail("Spotify playback was interrupted")
1070 return
1071 if not (chunk := shaper.shape(chunk)):
1072 continue
1073 if pace_anchor is None:
1074 pace_anchor = loop.time()
1075 paced_bytes = 0
1076 paced_bytes += len(chunk)
1077 chunk = self._scrub(chunk)
1078 if chunk and self._engine_exited and chunk.count(0) == len(chunk):
1079 # the engine is gone: what the sink renders from here on is only
1080 # padding, so the item's audio has fully arrived
1081 chunk = b""
1082 if chunk:
1083 self._read_bytes += len(chunk)
1084 if not await self._hand_over(chunk):
1085 return
1086 elif self._engine_exited:
1087 self._finish_delivery()
1088 return
1089 resume_at = pace_anchor + paced_bytes / (_BYTES_PER_SECOND * _PACE_RATE) - _PACE_BURST_S
1090 if (delay := resume_at - loop.time()) > 0:
1091 await asyncio.sleep(delay)
1092
1093 def _scrub(self, chunk: bytes) -> bytes:
1094 """Drop the sink's tail padding (the capture shaper already trims the lead)."""
1095 if self._duration_ms is not None and chunk.count(0) == len(chunk):
1096 # zeros inside the item's own tail zone are the sink idling while the
1097 # engine winds down, not content; an item no longer than the zone has
1098 # no distinguishable tail and its silence is left alone
1099 target = int((self._duration_ms - self._seek_target_ms) / 1000 * _BYTES_PER_SECOND)
1100 zone = int(_TAIL_PAD_ZONE_S * _BYTES_PER_SECOND)
1101 if target > zone and self._read_bytes >= target - zone:
1102 self._tail_zeros += len(chunk)
1103 if self._tail_zeros > int(_TAIL_PAD_GRACE_S * _BYTES_PER_SECOND):
1104 return b""
1105 else:
1106 self._tail_zeros = 0
1107 return chunk
1108
1109 async def _hand_over(self, chunk: bytes) -> bool:
1110 """
1111 Cushion one chunk for the consumer, pausing the engine when it is full.
1112
1113 :return: False when the run ended while waiting for cushion space.
1114 """
1115 if self._chunks.full():
1116 # the consumer is not taking audio (its buffer is at capacity):
1117 # suspend the sink so the engine pauses instead of overflowing the
1118 # FIFO, and resume once there is room again
1119 await self._set_sink(running=False)
1120 while not self._stopped and self._error is None:
1121 with suppress(TimeoutError):
1122 async with asyncio.timeout(_READ_SLICE_S):
1123 await self._chunks.put(chunk)
1124 break
1125 else:
1126 return False
1127 await self._set_sink(running=self._engine_playing)
1128 return True
1129 self._chunks.put_nowait(chunk)
1130 return True
1131
1132 def _finish_delivery(self) -> None:
1133 """Mark the item's audio as fully handed over."""
1134 self._delivery_done = True
1135 # the sentinel wakes a consumer already blocked on an empty cushion;
1136 # when the cushion is full, the flag alone ends the stream once the
1137 # consumer has drained it
1138 with suppress(asyncio.QueueFull):
1139 self._chunks.put_nowait(None)
1140
1141 async def _set_sink(self, *, running: bool) -> None:
1142 """Run the capture sink only while its audio has somewhere to go."""
1143 async with self._sink_lock:
1144 if (sink := self._sink) is None or running == self._sink_running:
1145 return
1146 try:
1147 if running:
1148 await sink.resume()
1149 else:
1150 await sink.suspend()
1151 except Exception as err:
1152 # fail closed: a sink with unknown suspend state would leak stall
1153 # silence into (or withhold audio from) the delivered PCM
1154 self._fail(f"Spotify playback could not be controlled: {err}")
1155 return
1156 self._sink_running = running
1157
1158 async def _run_events(self, client: SoloistClient, client_ready: asyncio.Event) -> None:
1159 """Keep the WebSocket client connected and feed its events into the run state."""
1160 proc = self._proc
1161 assert proc is not None
1162 if not await client.wait_until_ready(_STARTUP_TIMEOUT_S):
1163 # a natural exit right at startup still has to release the waiters
1164 if not self._engine_exited and proc.returncode is None:
1165 self._fail("Spotify playback did not start up correctly")
1166 client_ready.set()
1167 return
1168 client_ready.set()
1169 while not self._stopped:
1170 try:
1171 await client.listen_events(self._handle_event)
1172 except asyncio.CancelledError:
1173 raise
1174 except Exception as err:
1175 if self._engine_exited or proc.returncode is not None:
1176 # the daemon exited (the item finished); the socket dying with
1177 # it is not an error
1178 return
1179 self.logger.debug("Soloist event connection lost, reconnecting: %s", err)
1180 await asyncio.sleep(_CONNECT_POLL_S)
1181
1182 async def _handle_event(self, event: SoloistEvent) -> None:
1183 """Track what the engine is doing with this run's one item."""
1184 data = event.data
1185 if isinstance(data, SoloistAuthState):
1186 self._logged_in = data.logged_in
1187 if data.logged_in is False and not self._unpaired:
1188 await self._check_pairing_lost()
1189 return
1190 if isinstance(data, SoloistDeviceChanged):
1191 self._observe_device_active(active=data.is_active)
1192 return
1193 if isinstance(data, SoloistTrackChanged):
1194 if data.item is not None and data.item.uri:
1195 self._observe_item(data.item.uri, _decorated_duration_ms(data.item))
1196 return
1197 if isinstance(data, SoloistPositionSync):
1198 self._observe_position(data.position.position_ms)
1199 return
1200 if isinstance(data, SoloistPlaybackState):
1201 if data.item is not None and data.item.uri:
1202 self._observe_item(data.item.uri, _decorated_duration_ms(data.item))
1203 if data.position is not None:
1204 self._observe_position(data.position.position_ms)
1205 playing = data.status == "playing"
1206 self._engine_playing = playing
1207 if self._started.is_set() and self._reader is not None:
1208 # pause silence stays out of the delivered PCM; the cushion gate
1209 # takes priority over resuming
1210 if playing and not self._chunks.full():
1211 await self._set_sink(running=True)
1212 elif not playing and proc_running(self._proc):
1213 await self._set_sink(running=False)
1214
1215 def _observe_item(self, uri: str, duration_ms: int | None) -> None:
1216 """Record the engine reaching an item."""
1217 if uri != self.spotify_uri:
1218 if not self._started.is_set():
1219 # the engine started on something else entirely: whatever it is
1220 # playing, it is not what was asked
1221 self._fail(
1222 "Spotify is already streaming somewhere else - an account "
1223 "can only play in one place at a time"
1224 )
1225 return
1226 # The engine wanders into the next track (autoplay) in the instant
1227 # before a finished single-track run exits: this run's item is over,
1228 # and what renders now is not its audio. The daemon is put down
1229 # rather than left to play to nobody on the account's one stream.
1230 self._item_over = True
1231 self.logger.debug("Engine wandered to %s; %s has ended", uri, self.spotify_uri)
1232 self._finish_delivery()
1233 self.mass.create_task(self.backend.discard_run, self)
1234 return
1235 if duration_ms:
1236 self._duration_ms = duration_ms
1237 self._started.set()
1238
1239 def _observe_device_active(self, *, active: bool) -> None:
1240 """Fail the run when the account starts playing somewhere else."""
1241 # the engine reports itself active as each run starts, and a run whose item
1242 # is over or whose daemon has gone is only shedding the device on its way
1243 # out: another player took the account over when a LIVE item loses it. Auth
1244 # and playback states carry the same flag but report False during startup.
1245 if active or self._item_over or self._engine_exited or not self._started.is_set():
1246 return
1247 self._fail(
1248 "Spotify is already streaming somewhere else - an account "
1249 "can only play in one place at a time"
1250 )
1251
1252 def _observe_position(self, position_ms: int) -> None:
1253 """Confirm an armed seek once the engine reports at (or past) its target."""
1254 self._position_ms = position_ms
1255 if (
1256 not self._seek_confirmed.is_set()
1257 and self._seek_target_ms
1258 and position_ms >= self._seek_target_ms - _SEEK_CONFIRM_GRACE_MS
1259 ):
1260 self._seek_confirmed.set()
1261
1262
1263def proc_running(proc: AsyncProcess | None) -> bool:
1264 """Return whether the daemon process is still alive."""
1265 return proc is not None and proc.returncode is None
1266
1267
1268def _decorated_duration_ms(item: object) -> int | None:
1269 """Return the item duration the engine reports in its playback decorations."""
1270 decorations = getattr(item, "decorations", None)
1271 if not isinstance(decorations, dict):
1272 return None
1273 playback = decorations.get("playback")
1274 if not isinstance(playback, dict):
1275 return None
1276 duration_ms = playback.get("duration_ms")
1277 return int(duration_ms) if isinstance(duration_ms, int | float) else None
1278
1279
1280class _CaptureShaper:
1281 """
1282 Turns raw FIFO reads into whole sample frames of real audio.
1283
1284 Two jobs. It drops the infrastructure silence that precedes the session's
1285 first decoded sample (bounded, see ``_MAX_LEAD_TRIM_S``). And it carries a
1286 partial frame across reads: ``StreamReader.read`` returns whatever is
1287 available, which is not always a whole number of frames, so without this an
1288 item change between two reads would end one item mid-frame and start the
1289 next on the remainder - swapping its channels.
1290 """
1291
1292 def __init__(self) -> None:
1293 """Initialize the shaper for one session."""
1294 self._lead_skipped = 0
1295 self._first_audio_seen = False
1296 self._carry = b""
1297
1298 def shape(self, chunk: bytes) -> bytes:
1299 """
1300 Return the next whole frames of audio, or empty when there are none yet.
1301
1302 :param chunk: The bytes just read from the capture FIFO.
1303 """
1304 if self._carry:
1305 chunk = self._carry + chunk
1306 self._carry = b""
1307 if not self._first_audio_seen:
1308 chunk, skipped = _trim_lead_silence(chunk, self._lead_skipped)
1309 self._lead_skipped += skipped
1310 if not chunk.lstrip(b"\x00"):
1311 # still nothing but pre-roll: hold what was left of the frame so
1312 # the next read continues on the same grid
1313 self._carry = chunk
1314 return b""
1315 self._first_audio_seen = True
1316 if remainder := len(chunk) % _FRAME_BYTES:
1317 self._carry = chunk[-remainder:]
1318 return chunk[:-remainder]
1319 self._carry = b""
1320 return chunk
1321
1322
1323async def _cancel_and_join(tasks: list[asyncio.Task[None]]) -> None:
1324 """Cancel the given tasks and wait for them to finish."""
1325 for task in tasks:
1326 task.cancel()
1327 for task in tasks:
1328 with suppress(asyncio.CancelledError, Exception):
1329 await task
1330
1331
1332async def _open_fifo_reader(
1333 fifo_path: Path,
1334) -> tuple[asyncio.StreamReader, asyncio.ReadTransport]:
1335 """Open the sink's FIFO for non-blocking reads with a small buffer limit."""
1336 loop = asyncio.get_running_loop()
1337 fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK)
1338 try:
1339 pipe_file = os.fdopen(fd, "rb", buffering=0)
1340 except OSError:
1341 os.close(fd)
1342 raise
1343 reader = asyncio.StreamReader(limit=_READ_CHUNK_SIZE * 2)
1344 try:
1345 transport, _ = await loop.connect_read_pipe(
1346 partial(asyncio.StreamReaderProtocol, reader), pipe_file
1347 )
1348 except BaseException:
1349 pipe_file.close()
1350 raise
1351 return reader, transport
1352
1353
1354def _trim_lead_silence(chunk: bytes, already_skipped: int) -> tuple[bytes, int]:
1355 """
1356 Drop leading infrastructure silence from the head of the session (bounded).
1357
1358 :param chunk: The chunk read from the FIFO.
1359 :param already_skipped: Silence bytes dropped from earlier chunks.
1360 :return: The (possibly emptied/shortened) chunk and how many bytes were dropped.
1361 """
1362 max_lead_trim = int(_MAX_LEAD_TRIM_S * _BYTES_PER_SECOND)
1363 stripped = chunk.lstrip(b"\x00")
1364 if not stripped:
1365 if already_skipped + len(chunk) <= max_lead_trim:
1366 # whole frames only: the offset below places the first sample on the
1367 # session's frame grid, which holds only while everything dropped
1368 # before it was a whole number of frames. The leftover bytes go back
1369 # to the caller, which carries them into the next read.
1370 dropped = len(chunk) // _FRAME_BYTES * _FRAME_BYTES
1371 return chunk[dropped:], dropped
1372 # budget exhausted: this is genuine silence content, not infrastructure
1373 return chunk, 0
1374 # Keep sample-frame alignment when the audio starts mid-chunk, and never trim
1375 # past the budget: whatever silence is left by then is content, not pre-roll.
1376 remaining = max(0, max_lead_trim - already_skipped)
1377 offset = min(len(chunk) - len(stripped), remaining) // _FRAME_BYTES * _FRAME_BYTES
1378 return chunk[offset:], offset
1379