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