music-assistant-server

1"""Playback session coordinator for Sendspin players."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections import deque
8from contextlib import suppress
9from dataclasses import dataclass, field
10from typing import TYPE_CHECKING, Any, cast
11from uuid import UUID, uuid4
12
13from aiosendspin.models.types import AudioCodec as SendspinAudioCodec
14from aiosendspin.server.audio import AudioFormat as SendspinAudioFormat
15from aiosendspin.server.push_stream import MAIN_CHANNEL, PushStream, StreamStoppedError
16from aiosendspin.server.roles.player.v1 import PlayerV1Role
17from music_assistant_models.enums import ContentType, MediaType
18from music_assistant_models.media_items.audio_format import AudioFormat
19
20from music_assistant.constants import CONF_OUTPUT_CHANNELS
21from music_assistant.controllers.streams.audio_processing import get_media_session_id
22from music_assistant.helpers.audio import iter_pcm_slices
23from music_assistant.helpers.ffmpeg import FFMpeg
24from music_assistant.helpers.util import import_module_in_thread
25from music_assistant.models.player import PlayerMedia
26from music_assistant.providers.sendspin.bridge_role import (
27    BRIDGE_BIT_DEPTH,
28    BRIDGE_CHANNELS,
29    BRIDGE_SAMPLE_RATE,
30    BridgePlayerRole,
31)
32
33if TYPE_CHECKING:
34    from music_assistant.helpers.dsp import ComplexFilter
35
36    from .player import SendspinPlayer
37    from .provider import SendspinProvider
38
39
40# Default session PCM format (MA-side and wire) used until _run_playback picks a
41# leader-driven rate. Same sample format expressed in both MA and Sendspin types.
42_DEFAULT_PCM_FORMAT = AudioFormat(
43    content_type=ContentType.PCM_F32LE,
44    sample_rate=48000,
45    bit_depth=32,
46    channels=2,
47)
48_DEFAULT_SENDSPIN_PCM_FORMAT = SendspinAudioFormat(
49    sample_rate=48000,
50    bit_depth=32,
51    channels=2,
52    sample_type="float",
53)
54# Media types whose upstream always feeds at realtime rate, so the Sendspin queue
55# cannot grow after playback begins and their send-ahead stays at the min_buffer_ms
56# floor. Buffered types (tracks, podcasts, etc.) race ahead and fill the queue
57# naturally, so their send-ahead may extend to a larger required_lead_time_ms without
58# lasting cost. Media type alone does not settle it: a track can come from a source
59# that also feeds just-in-time, which is what StreamDetails.is_realtime marks - see
60# _is_live_source().
61_LIVE_MEDIA_TYPES: frozenset[MediaType] = frozenset(
62    {
63        MediaType.RADIO,
64        MediaType.AUDIO_SOURCE,
65        MediaType.PLUGIN_SOURCE,
66    }
67)
68
69
70# Sample rate ceiling for lossy output codecs — anything above is wasted bandwidth.
71_LOSSY_MAX_SAMPLE_RATE = 48000
72# Max PCM slice fed to the producer per iteration.
73_PRODUCER_SLICE_US = 100_000
74# Max pending chunks between producer and committer before the producer blocks.
75_PRODUCER_BACKLOG_SIZE = 64
76# Backpressure threshold: push stream sleeps when buffered audio exceeds this.
77_PRODUCER_BUFFER_LIMIT_US = 30_000_000
78# Start join promotion once catchup processor lag is within this window of the history tail.
79_JOIN_PROMOTE_ARM_WINDOW_US = 2_000_000
80# Accept catchup output within this margin of the promotion target.
81_JOIN_PROMOTE_TOLERANCE_US = 50_000
82# Abort join catchup if promotion hasn't completed within this.
83_JOIN_PROMOTION_TIMEOUT_S = 15.0
84# Retain committed history this far behind real-time for late-join backfill.
85# This pre-history also warms up ffmpeg's internal filter buffers so the DSP
86# output has settled by the time the member's channel goes live.
87_HISTORY_KEEP_PAST_US = 1_000_000
88
89
90class _BufferedFfmpegProcessor:
91    """FFmpeg wrapper with small output carry-over buffer and duration-based reads."""
92
93    def __init__(self, ffmpeg: FFMpeg, audio_format: AudioFormat) -> None:
94        self._ffmpeg = ffmpeg
95        self._output_buffer = bytearray()
96        bytes_per_sample = max(1, int(audio_format.bit_depth // 8))
97        self._sample_rate = int(audio_format.sample_rate)
98        self._frame_size = bytes_per_sample * int(audio_format.channels)
99        self._bytes_per_second = self._sample_rate * self._frame_size
100        # ~25ms worth of audio per read syscall.
101        self._read_quantum_bytes = max(1, int(self._bytes_per_second * 0.025))
102        self._produced_output_us = 0
103        self._pending_skip_bytes = 0
104
105    async def start(self) -> None:
106        await self._ffmpeg.start()
107
108    async def close(self) -> None:
109        await self._ffmpeg.close()
110
111    async def push(self, pcm: bytes) -> None:
112        await self._ffmpeg.write(pcm)
113
114    async def write_eof(self) -> None:
115        """Signal no more input, causing ffmpeg to flush its internal buffers."""
116        await self._ffmpeg.write_eof()
117
118    @property
119    def produced_output_us(self) -> int:
120        """Return cumulative output duration currently drained from ffmpeg."""
121        return self._produced_output_us
122
123    async def read_duration_us(self, duration_us: int) -> bytes:
124        """Block-read exactly `duration_us` worth of processed PCM from ffmpeg."""
125        target_bytes = self._target_bytes_for_duration_us(duration_us)
126        if target_bytes == 0:
127            return b""
128
129        while len(self._output_buffer) < target_bytes:
130            missing = target_bytes - len(self._output_buffer)
131            read_size = max(self._read_quantum_bytes, missing)
132            chunk = await self._ffmpeg.readexactly(read_size)
133            if not chunk:
134                break
135            chunk = self._consume_pending_skip(chunk)
136            if chunk:
137                self._output_buffer.extend(chunk)
138
139        out = bytes(self._output_buffer[:target_bytes])
140        del self._output_buffer[:target_bytes]
141        return out
142
143    async def drain_available(self) -> int:
144        """
145        Non-blocking drain of ffmpeg stdout into internal buffer.
146
147        Returns cumulative produced output duration in microseconds.
148        """
149        while True:
150            try:
151                # 1ms timeout: non-blocking check for available data.
152                chunk = await asyncio.wait_for(
153                    self._ffmpeg.read(self._read_quantum_bytes),
154                    timeout=0.001,
155                )
156            except TimeoutError:
157                break
158            if not chunk:
159                break
160            self._produced_output_us += self._duration_us_for_bytes(len(chunk))
161            chunk = self._consume_pending_skip(chunk)
162            if chunk:
163                self._output_buffer.extend(chunk)
164            if len(chunk) < self._read_quantum_bytes:
165                break
166        return self._produced_output_us
167
168    async def drain_forever(self) -> None:
169        """Continuously drain ffmpeg stdout into internal buffer until EOF."""
170        while True:
171            chunk = await self._ffmpeg.read(self._read_quantum_bytes)
172            if not chunk:
173                break
174            self._produced_output_us += self._duration_us_for_bytes(len(chunk))
175            chunk = self._consume_pending_skip(chunk)
176            if chunk:
177                self._output_buffer.extend(chunk)
178
179    def pop_duration_us(self, duration_us: int) -> bytes | None:
180        """Pop exactly `duration_us` from already buffered output, or None if insufficient."""
181        target_bytes = self._target_bytes_for_duration_us(duration_us)
182        if target_bytes == 0:
183            return b""
184        if len(self._output_buffer) < target_bytes:
185            return None
186        out = bytes(self._output_buffer[:target_bytes])
187        del self._output_buffer[:target_bytes]
188        return out
189
190    def buffered_duration_us(self) -> int:
191        """Return buffered output duration currently available for immediate pop."""
192        return self._duration_us_for_bytes(len(self._output_buffer))
193
194    def pop_duration_us_or_pad(self, duration_us: int, pad_tolerance_us: int) -> bytes | None:
195        """Pop target duration; if short within tolerance, pad tail with silence."""
196        target_bytes = self._target_bytes_for_duration_us(duration_us)
197        if target_bytes == 0:
198            return b""
199        available = len(self._output_buffer)
200        if available >= target_bytes:
201            out = bytes(self._output_buffer[:target_bytes])
202            del self._output_buffer[:target_bytes]
203            return out
204        short_bytes = target_bytes - available
205        short_us = self._duration_us_for_bytes(short_bytes)
206        if short_us > max(0, pad_tolerance_us):
207            return None
208        out = bytes(self._output_buffer)
209        self._output_buffer.clear()
210        self._pending_skip_bytes += short_bytes
211        return out + (b"\x00" * short_bytes)
212
213    def pad_and_skip(self, duration_us: int) -> bytes:
214        """Return silence PCM and skip the equivalent from upcoming ffmpeg output."""
215        target_bytes = self._target_bytes_for_duration_us(duration_us)
216        if target_bytes <= 0:
217            return b""
218        # Drop buffered output with stale source positions.
219        leftover = len(self._output_buffer)
220        self._output_buffer.clear()
221        self._pending_skip_bytes += max(0, target_bytes - leftover)
222        return b"\x00" * target_bytes
223
224    def _consume_pending_skip(self, chunk: bytes) -> bytes:
225        # Drops bytes pop_duration_us_or_pad replaced with silence to keep timeline aligned.
226        if self._pending_skip_bytes <= 0 or not chunk:
227            return chunk
228        skip = min(self._pending_skip_bytes, len(chunk))
229        self._pending_skip_bytes -= skip
230        return chunk[skip:]
231
232    def _duration_us_for_bytes(self, byte_count: int) -> int:
233        if byte_count <= 0 or self._sample_rate <= 0 or self._frame_size <= 0:
234            return 0
235        frames = byte_count // self._frame_size
236        if frames <= 0:
237            return 0
238        return int((frames * 1_000_000) / self._sample_rate)
239
240    def _target_bytes_for_duration_us(self, duration_us: int) -> int:
241        """Convert duration to frame-aligned PCM byte count."""
242        if duration_us <= 0 or self._sample_rate <= 0 or self._frame_size <= 0:
243            return 0
244        samples = max(0, int((duration_us * self._sample_rate + 500_000) / 1_000_000))
245        return samples * self._frame_size
246
247
248@dataclass(slots=True)
249class _HistoryChunk:
250    start_time_us: int
251    duration_us: int
252    pcm: bytes
253
254
255@dataclass(slots=True)
256class _PendingChunk:
257    pcm: bytes
258    duration_us: int
259
260
261@dataclass(slots=True)
262class _JoinCatchupState:
263    """
264    Per-member state for a join-catchup processor replaying history through DSP.
265
266    The processor is fed historical + live PCM via ``input_queue``.  Once its
267    output catches up to the live stream (within tolerance), it is promoted to
268    the member's live pipeline.  See ``_inject_ready_join_historical`` for the
269    full promotion lifecycle.
270    """
271
272    processor: _BufferedFfmpegProcessor
273    input_queue: asyncio.Queue[bytes | None]
274    writer_task: asyncio.Task[None]
275    drainer_task: asyncio.Task[None]
276    snapshot_task: asyncio.Task[None] | None = None
277    # Timeline position of the first history chunk fed into the processor.
278    first_history_start_us: int | None = None
279    # Timeline position up to which PCM has been enqueued into the processor.
280    fed_until_us: int | None = None
281    # End of the history snapshot taken when catchup started.
282    history_end_us: int | None = None
283    # Locked target: once set, promotion fires when output reaches this point.
284    promotion_target_end_us: int | None = None
285    # Monotonic time when promotion was armed, used for timeout detection.
286    promotion_armed_monotonic_s: float | None = None
287    write_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
288
289
290@dataclass(slots=True)
291class _PipelineConfig:
292    requires_transform: bool
293    output_channels: str
294    filter_params: tuple[str | ComplexFilter, ...]
295
296    @property
297    def signature(self) -> tuple[bool, str, tuple[str | ComplexFilter, ...]]:
298        return (self.requires_transform, self.output_channels, self.filter_params)
299
300
301@dataclass(slots=True)
302class _MemberPipeline:
303    player_id: str
304    channel_id: UUID
305    config: _PipelineConfig
306    processor: _BufferedFfmpegProcessor | None = None
307    ready: bool = False
308
309
310class SendspinPlaybackSession:
311    """
312    Coordinates playback for a Sendspin player group leader.
313
314    The push stream supports multi-channel audio: members that need per-player
315    DSP (EQ, channel mixing, output routing) each get a dedicated ffmpeg
316    processor and a separate channel. Members without DSP share MAIN_CHANNEL
317    and receive the raw PCM directly.
318
319    Playback runs as two concurrent coroutines inside ``_run_playback``:
320
321    * **Producer** -- reads PCM from the MA stream, slices it into fixed-size
322      chunks, queues them, and writes each slice into per-member ffmpeg
323      processors (transform push) in parallel.
324    * **Consumer** -- dequeues chunks, reads the corresponding transformed
325      output from each processor (transform read), prepares all channels on
326      the push stream, commits audio, and applies backpressure via
327      ``sleep_to_limit_buffer``.
328
329    When a new member joins mid-playback, a *join-catchup* processor replays
330    committed history through the member's DSP chain so it can be promoted
331    to the live pipeline without an audible gap.
332    """
333
334    def __init__(self, player: SendspinPlayer) -> None:
335        """Initialize session coordinator bound to the owning player."""
336        self.player = player
337        self.playback_task: asyncio.Task[None] | None = None
338        self.pending_join_members: set[str] = set()
339        self._state_lock = asyncio.Lock()
340        self._members: set[str] = set()
341        self._member_pipelines: dict[str, _MemberPipeline] = {}
342        self._push_stream: PushStream | None = None
343        self._playback_running = False
344        self._producer_eof_sent = False
345        self._timeline_start_us: int | None = None
346        self._first_commit_monotonic_us: int | None = None
347        self._produced_audio_us = 0
348        self._history: deque[_HistoryChunk] = deque()
349        self._join_catchup: dict[str, _JoinCatchupState] = {}
350        self._pipeline_config_cache: dict[str, _PipelineConfig] = {}
351        self._preassigned_channels: dict[str, UUID] = {}
352        self._mapping_dirty = True
353        self._cancel_requested = False
354        # PCM formats are session-scoped and refreshed at the start of every
355        # _run_playback: the wire/MA-side rate is taken from the leader player's
356        # preferred format (capped at 48 kHz for lossy codecs), F32 is always used
357        # internally for DSP headroom.
358        self._pcm_format: AudioFormat = _DEFAULT_PCM_FORMAT
359        self._sendspin_pcm_format: SendspinAudioFormat = _DEFAULT_SENDSPIN_PCM_FORMAT
360        self._queue_id: str | None = None
361        self._queue_session_id: str | None = None
362
363    def flow_track_anchor_us(self, track_start_offset_us: int) -> int | None:
364        """
365        Server-clock time of the current flow track's file-position 0.
366
367        ``track_start_offset_us`` is the current track's start offset within the
368        flow stream (minus its file seek), so beats timed from the track file
369        map onto the shared audio timeline regardless of queue position. Returns
370        None until the first chunk commits and the timeline anchor is known.
371        """
372        if self._timeline_start_us is None:
373            return None
374        return self._timeline_start_us + track_start_offset_us
375
376    # -- Public API ------------------------------------------------------------
377
378    async def transfer_to(self, new_player: SendspinPlayer) -> None:
379        """
380        Transfer session ownership to a new player.
381
382        Used during dynamic leader switching to keep the push stream alive
383        while the old leader is removed from the sendspin group. The PushStream
384        and all internal state (pipelines, history, join-catchup) stay intact;
385        only the owning player reference is updated.
386
387        Cleans up the old leader's pipeline/channel state so its FFmpeg
388        processor is released.
389
390        :param new_player: The SendspinPlayer that will take over as session owner.
391        """
392        old_leader_id = self.player.player_id
393        self.player = new_player
394        # Release the old leader's DSP pipeline -- it's no longer in the group
395        # and _refresh_member_mappings won't touch it since it only iterates
396        # current members + the (new) leader.
397        async with self._state_lock:
398            pipeline = self._member_pipelines.pop(old_leader_id, None)
399            self._pipeline_config_cache.pop(old_leader_id, None)
400            self._preassigned_channels.pop(old_leader_id, None)
401            self._mapping_dirty = True
402        if pipeline is not None and pipeline.processor is not None:
403            await self._close_member_ffmpeg(pipeline.processor)
404
405    async def cancel(self, reason: str, *, keep_stream: bool = False) -> None:
406        """
407        Cancel and await the active playback task, if any.
408
409        :param reason: Why the task is being cancelled, for logging and the cancel message.
410        :param keep_stream: Keep the stream active for a track change and only have clients
411            clear their buffers. Ignored while legacy clients are allowed, since they might
412            mishandle stream/clear.
413        """
414        task = self.playback_task
415        if task is None:
416            return
417        if task.done():
418            if self.playback_task is task:
419                self.playback_task = None
420            return
421        provider = cast("SendspinProvider", self.player.provider)
422        if provider.server_api.allow_noncompliant_clients:
423            keep_stream = False
424        self.player.logger.debug("Cancelling playback task (%s)", reason)
425        self._cancel_requested = True
426        task.cancel(msg=reason)
427        if keep_stream:
428            with suppress(Exception):
429                self._stop_push_stream(keep_stream=True)
430        with suppress(asyncio.CancelledError, Exception):
431            await task
432        if self.playback_task is task:
433            self.playback_task = None
434
435    async def start(self, media: PlayerMedia, restart: bool = False) -> None:
436        """Start background playback for `media`."""
437        active_task = self.playback_task
438        if active_task is not None and not active_task.done():
439            if not restart:
440                raise RuntimeError("playback already active")
441            await self.cancel("restart requested", keep_stream=True)
442        self._cancel_requested = False
443        self.playback_task = asyncio.create_task(self._run_playback(media))
444        self._attach_task_exception_logger(self.playback_task, "playback")
445
446    async def close(self) -> None:
447        """Stop playback and release all managed resources."""
448        await self.cancel("session close")
449        self.pending_join_members.clear()
450        async with self._state_lock:
451            self._members.clear()
452            self._mapping_dirty = True
453        await self._clear_member_pipelines()
454        await self._clear_join_catchup()
455        async with self._state_lock:
456            self._history.clear()
457            self._produced_audio_us = 0
458            self._timeline_start_us = None
459            self._first_commit_monotonic_us = None
460            self._pipeline_config_cache.clear()
461            self._preassigned_channels.clear()
462
463    async def add_member(self, player_id: str) -> None:
464        """Add a member to the group with DSP-aware lifecycle handling."""
465        async with self._state_lock:
466            if player_id in self._members:
467                return
468            self.pending_join_members.add(player_id)
469            # Preserve any channel pre-resolved during add_client so join-time
470            # role requirements and prepared audio stay on the same channel.
471            self._preassigned_channels.setdefault(player_id, uuid4())
472        try:
473            await self._start_join_catchup(player_id)
474        except Exception:
475            async with self._state_lock:
476                self.pending_join_members.discard(player_id)
477            await self._release_player_channel(player_id)
478            raise
479        # Promote to full member even if already pending to avoid losing
480        # the join when a cancelled task clears our pending flag.
481        async with self._state_lock:
482            if player_id not in self.pending_join_members:
483                return
484            self._members.add(player_id)
485            self._mapping_dirty = True
486            self.pending_join_members.discard(player_id)
487
488    async def remove_member(self, player_id: str) -> None:
489        """Remove a member from the group and clean up per-member playback state."""
490        async with self._state_lock:
491            self.pending_join_members.discard(player_id)
492            self._members.discard(player_id)
493            self._mapping_dirty = True
494            self._pipeline_config_cache.pop(player_id, None)
495            self._preassigned_channels.pop(player_id, None)
496        await self._stop_join_catchup(player_id)
497        await self._release_player_channel(player_id)
498
499    async def sync_members(self, member_ids: set[str]) -> None:
500        """Reconcile session members to exactly the provided set."""
501        async with self._state_lock:
502            current_members = set(self._members)
503            stale_pending = self.pending_join_members - member_ids
504        for player_id in stale_pending:
505            await self.remove_member(player_id)
506        for player_id in current_members - member_ids:
507            await self.remove_member(player_id)
508        for player_id in member_ids - current_members:
509            await self.add_member(player_id)
510
511    # -- Helpers ---------------------------------------------------------------
512
513    def _attach_task_exception_logger(self, task: asyncio.Task[Any], name: str) -> None:
514        """Log unhandled exception from background task when it finishes."""
515
516        def _done_callback(done_task: asyncio.Task[Any]) -> None:
517            if done_task.cancelled():
518                return
519            with suppress(Exception):
520                exc = done_task.exception()
521                if exc is not None:
522                    self.player.logger.exception(
523                        "Background task failed: %s",
524                        name,
525                        exc_info=exc,
526                    )
527
528        task.add_done_callback(_done_callback)
529
530    def _get_join_readiness(self) -> tuple[bool, str | None]:
531        """Check whether live join DSP preparation can be performed right now."""
532        if self._playback_running and self._push_stream is not None:
533            return (True, None)
534        return (False, "no active stream context")
535
536    # -- Snapshot helper -------------------------------------------------------
537
538    async def _snapshot_active_pipelines(
539        self,
540    ) -> tuple[set[str], tuple[tuple[str, _MemberPipeline], ...]]:
541        """Return (join_pending_ids, active_pipelines) under lock."""
542        async with self._state_lock:
543            members = self._members
544            leader_id = self.player.player_id
545            return set(self._join_catchup), tuple(
546                (mid, p)
547                for mid, p in self._member_pipelines.items()
548                if mid in members or mid == leader_id
549            )
550
551    # -- Join catchup ----------------------------------------------------------
552
553    async def _start_join_catchup(self, player_id: str) -> None:  # noqa: PLR0915
554        """Start dedicated join catchup processor fed from committed history."""
555        async with self._state_lock:
556            playback_active = self._playback_running and self._push_stream is not None
557        if not playback_active:
558            return
559
560        pipeline = await self._sync_member_pipeline(player_id)
561        if not pipeline.config.requires_transform:
562            return
563
564        await self._stop_join_catchup(player_id)
565
566        ffmpeg_obj = self._create_member_ffmpeg(pipeline.config.filter_params)
567        processor = _BufferedFfmpegProcessor(ffmpeg_obj, self._pcm_format)
568        await processor.start()
569        # Bounded queue sized to hold the full buffer duration with some headroom.
570        queue_size = (_PRODUCER_BUFFER_LIMIT_US // _PRODUCER_SLICE_US) + _PRODUCER_BACKLOG_SIZE
571        input_queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=queue_size)
572
573        async with self._state_lock:
574            history_snapshot = list(self._history)
575        if not history_snapshot:
576            await processor.close()
577            return
578        history_end_us = history_snapshot[-1].start_time_us + history_snapshot[-1].duration_us
579        writer_task: asyncio.Task[None] | None = None
580        drainer_task: asyncio.Task[None] | None = None
581        snapshot_task: asyncio.Task[None] | None = None
582        state: _JoinCatchupState | None = None
583        registered = False
584
585        async def _writer() -> None:
586            while True:
587                chunk = await input_queue.get()
588                if chunk is None:
589                    return
590                await processor.push(chunk)
591
592        async def _drainer() -> None:
593            await processor.drain_forever()
594
595        try:
596            async with self._state_lock:
597                writer_task = asyncio.create_task(_writer())
598                drainer_task = asyncio.create_task(_drainer())
599                self._attach_task_exception_logger(writer_task, f"join_writer_{player_id}")
600                self._attach_task_exception_logger(drainer_task, f"join_drainer_{player_id}")
601
602                state = _JoinCatchupState(
603                    processor=processor,
604                    input_queue=input_queue,
605                    writer_task=writer_task,
606                    drainer_task=drainer_task,
607                    history_end_us=history_end_us,
608                )
609                self._join_catchup[player_id] = state
610                snapshot_task = asyncio.create_task(
611                    self._feed_join_history(player_id, processor, history_snapshot)
612                )
613                self._attach_task_exception_logger(snapshot_task, f"join_snapshot_{player_id}")
614                state.snapshot_task = snapshot_task
615                registered = True
616        except BaseException:
617            if not registered:
618                # If registered, cleanup is handled via _stop_join_catchup/_clear_join_catchup.
619                async with self._state_lock:
620                    current = self._join_catchup.get(player_id)
621                    if current is state:
622                        self._join_catchup.pop(player_id, None)
623                for task in (snapshot_task, drainer_task, writer_task):
624                    if task is None:
625                        continue
626                    task.cancel()
627                    with suppress(asyncio.CancelledError, Exception):
628                        await task
629                with suppress(Exception):
630                    await processor.close()
631            raise
632
633    async def _feed_join_history(
634        self,
635        player_id: str,
636        processor: _BufferedFfmpegProcessor,
637        history_snapshot: list[_HistoryChunk],
638    ) -> None:
639        """Feed historical PCM into a join-catchup processor."""
640        async with self._state_lock:
641            state = self._join_catchup.get(player_id)
642        if state is None or state.processor is not processor:
643            return
644        async with state.write_lock:
645            first_history_start_us: int | None = None
646            previous_end_us: int | None = None
647            for hist_chunk in history_snapshot:
648                if first_history_start_us is None:
649                    first_history_start_us = hist_chunk.start_time_us
650                    async with self._state_lock:
651                        current = self._join_catchup.get(player_id)
652                        if current is not None and current.processor is processor:
653                            current.first_history_start_us = first_history_start_us
654                            current.fed_until_us = first_history_start_us
655                if previous_end_us is not None and hist_chunk.start_time_us > previous_end_us:
656                    gap_us = hist_chunk.start_time_us - previous_end_us
657                    silence = self._silence_for_duration_us(gap_us)
658                    if silence:
659                        await self._enqueue_join_pcm(state, silence)
660                await self._enqueue_join_pcm(state, hist_chunk.pcm)
661                previous_end_us = hist_chunk.start_time_us + hist_chunk.duration_us
662                async with self._state_lock:
663                    current = self._join_catchup.get(player_id)
664                    if current is not None and current.processor is processor:
665                        current.fed_until_us = previous_end_us
666
667    async def _stop_join_catchup(self, player_id: str) -> None:
668        """Stop and remove dedicated join catchup processor for one player."""
669        async with self._state_lock:
670            state = self._join_catchup.pop(player_id, None)
671        if state is None:
672            return
673        if state.snapshot_task is not None:
674            state.snapshot_task.cancel()
675            with suppress(asyncio.CancelledError, Exception):
676                await state.snapshot_task
677        state.writer_task.cancel()
678        with suppress(asyncio.CancelledError, Exception):
679            await state.writer_task
680        state.drainer_task.cancel()
681        with suppress(asyncio.CancelledError, Exception):
682            await state.drainer_task
683        with suppress(Exception):
684            await state.processor.close()
685
686    async def _promote_join_catchup_processor(
687        self,
688        player_id: str,
689        pipeline: _MemberPipeline,
690        target_end_us: int,
691    ) -> None:
692        """Promote join catchup processor to the member's live DSP processor."""
693        old_processor: _BufferedFfmpegProcessor | None = None
694        async with self._state_lock:
695            state = self._join_catchup.pop(player_id, None)
696            if state is None:
697                return
698            old_processor = pipeline.processor
699            pipeline.processor = state.processor
700        if state.snapshot_task is not None:
701            state.snapshot_task.cancel()
702            with suppress(asyncio.CancelledError, Exception):
703                await state.snapshot_task
704        # Let writer flush queued PCM before handoff; cancel if queue is full.
705        try:
706            state.input_queue.put_nowait(None)
707        except asyncio.QueueFull:
708            state.writer_task.cancel()
709        with suppress(asyncio.CancelledError, Exception):
710            await state.writer_task
711        state.drainer_task.cancel()
712        with suppress(asyncio.CancelledError, Exception):
713            await state.drainer_task
714        if self._producer_eof_sent and pipeline.config.requires_transform:
715            with suppress(Exception):
716                await pipeline.processor.write_eof()
717        if old_processor is not None and old_processor is not state.processor:
718            with suppress(Exception):
719                await old_processor.close()
720
721    async def _clear_join_catchup(self) -> None:
722        """Stop and remove all dedicated join catchup processors."""
723        async with self._state_lock:
724            player_ids = list(self._join_catchup.keys())
725        for player_id in player_ids:
726            await self._stop_join_catchup(player_id)
727
728    async def _release_player_channel(self, player_id: str) -> None:
729        """Release per-member channel/DSP state for a removed member."""
730        async with self._state_lock:
731            pipeline = self._member_pipelines.pop(player_id, None)
732            self._preassigned_channels.pop(player_id, None)
733        if pipeline is None or pipeline.processor is None:
734            return
735        await self._close_member_ffmpeg(pipeline.processor)
736
737    # -- Playback pipeline -----------------------------------------------------
738
739    async def _run_playback(self, media: PlayerMedia) -> None:  # noqa: PLR0915
740        """
741        Run the playback pipeline for a single media session.
742
743        Pulls PCM from the MA stream, feeds main + per-member DSP channels into the
744        Sendspin push stream, and commits audio continuously. Supports dynamic group
745        membership changes and late-join historical backfill while running.
746        """
747        # aiosendspin resamples and encodes with PyAV, which it imports lazily on first use -
748        # from inside commit_audio(), on the event loop. Pull that import forward to a thread,
749        # before the play timeline exists, so its cost can neither stall audio production nor
750        # push the timeline into a forward rebase.
751        await import_module_in_thread("av")
752        push_stream: PushStream | None = None
753        try:
754            # refresh the session PCM format from the leader's preferred output before
755            # building any pipelines; member ffmpeg pipelines and pre-computed filter
756            # params depend on this rate so the cache must also be cleared
757            self._pcm_format, self._sendspin_pcm_format = self._select_session_pcm_formats()
758            self._queue_id = media.source_id
759            self._queue_session_id = get_media_session_id(media)
760            self._pipeline_config_cache.clear()
761            self.player.logger.debug(
762                "Sendspin session PCM format: %d Hz / F32",
763                self._pcm_format.sample_rate,
764            )
765            push_stream = self._create_push_stream()
766            push_stream.set_live_source(self._is_live_source(media))
767            async with self._state_lock:
768                self._push_stream = push_stream
769                self._playback_running = True
770                self._producer_eof_sent = False
771                self._history.clear()
772                self._produced_audio_us = 0
773                self._timeline_start_us = None
774                self._first_commit_monotonic_us = None
775                self._mapping_dirty = True
776        except Exception:
777            # A track change stops the previous stream without stream/end, so a failed
778            # setup has to end this one. Cancellation propagates untouched, since there
779            # the successor keeps the stream.
780            if push_stream is not None:
781                with suppress(Exception):
782                    push_stream.stop()
783            await self._reset_session_state()
784            raise
785        # Bounded queue between producer (stream reader) and consumer (committer).
786        pending_chunks: asyncio.Queue[_PendingChunk | None] = asyncio.Queue(
787            maxsize=_PRODUCER_BACKLOG_SIZE
788        )
789        # Shadow deque mirroring pending_chunks for join-catchup backlog peeking.
790        pending_backlog: deque[_PendingChunk] = deque()
791        pending_duration_us = 0
792        last_elapsed_update_s = 0.0
793
794        async def _produce_pending_chunks() -> None:
795            nonlocal pending_duration_us
796            audio_source = self.player.mass.streams.get_stream(
797                media, self._pcm_format, self.player.player_id
798            )
799            completed = False
800            try:
801                async for chunk in audio_source:
802                    if not chunk:
803                        continue
804                    for slice_chunk in iter_pcm_slices(
805                        chunk, self._pcm_format, target_duration_ms=_PRODUCER_SLICE_US // 1000
806                    ):
807                        if not slice_chunk:
808                            continue
809                        duration_us = self._duration_us(slice_chunk, self._pcm_format)
810                        if duration_us <= 0:
811                            continue
812                        await self._refresh_member_mappings()
813                        pending = _PendingChunk(pcm=slice_chunk, duration_us=duration_us)
814                        await pending_chunks.put(pending)
815                        pending_backlog.append(pending)
816                        pending_duration_us += duration_us
817                        join_pending_ids, pipelines = await self._snapshot_active_pipelines()
818                        transform_pipelines: list[_MemberPipeline] = []
819                        for member_id, pipeline in pipelines:
820                            if not pipeline.config.requires_transform:
821                                continue
822                            if member_id in join_pending_ids:
823                                continue
824                            transform_pipelines.append(pipeline)
825                        results = await asyncio.gather(
826                            *(
827                                self._transform_member_chunk(pipeline, slice_chunk)
828                                for pipeline in transform_pipelines
829                            ),
830                            return_exceptions=True,
831                        )
832                        for pipeline, result in zip(transform_pipelines, results, strict=True):
833                            if isinstance(result, BaseException):
834                                self.player.logger.warning(
835                                    "Transform push failed for channel %s: %s",
836                                    pipeline.channel_id,
837                                    result,
838                                )
839                completed = True
840            finally:
841                if not completed:
842                    close_task = asyncio.create_task(audio_source.aclose())
843                    try:
844                        await asyncio.shield(close_task)
845                    except asyncio.CancelledError:
846                        await close_task
847                        raise
848
849        async def _commit_pending_chunks() -> None:
850            nonlocal pending_duration_us, last_elapsed_update_s
851            while True:
852                pending = await pending_chunks.get()
853                if pending is None:
854                    break
855                pending_backlog.popleft()
856                pending_duration_us = max(0, pending_duration_us - pending.duration_us)
857                await self._inject_ready_join_historical(push_stream, pending_backlog, pending.pcm)
858                push_stream.prepare_audio(
859                    pending.pcm, self._sendspin_pcm_format, channel_id=MAIN_CHANNEL
860                )
861                join_pending_ids, pipelines = await self._snapshot_active_pipelines()
862                transform_pipelines: list[_MemberPipeline] = []
863                for member_id, pipeline in pipelines:
864                    if not pipeline.config.requires_transform:
865                        continue
866                    if member_id in join_pending_ids:
867                        continue
868                    transform_pipelines.append(pipeline)
869                transformed_chunks = await asyncio.gather(
870                    *(
871                        self._read_member_chunk(pipeline, pending.duration_us)
872                        for pipeline in transform_pipelines
873                    ),
874                    return_exceptions=True,
875                )
876                for pipeline, transformed_chunk in zip(
877                    transform_pipelines, transformed_chunks, strict=True
878                ):
879                    if isinstance(transformed_chunk, BaseException):
880                        self.player.logger.warning(
881                            "Transform read failed for channel %s: %s",
882                            pipeline.channel_id,
883                            transformed_chunk,
884                        )
885                        continue
886                    if transformed_chunk is None:
887                        continue
888                    push_stream.prepare_audio(
889                        transformed_chunk,
890                        self._sendspin_pcm_format,
891                        channel_id=pipeline.channel_id,
892                    )
893                try:
894                    commit_start_us = await push_stream.commit_audio()
895                except StreamStoppedError:
896                    # Stream stopped since it was replaced by another stream
897                    self.player.logger.debug("Stopping commit loop due to stopped push stream")
898                    break
899                await push_stream.sleep_to_limit_buffer(_PRODUCER_BUFFER_LIMIT_US)
900                commit_now_us = push_stream.now_us()
901                committed_history_chunk = _HistoryChunk(
902                    start_time_us=int(commit_start_us),
903                    duration_us=pending.duration_us,
904                    pcm=pending.pcm,
905                )
906                async with self._state_lock:
907                    if self._timeline_start_us is None:
908                        self._timeline_start_us = int(commit_start_us)
909                    if self._first_commit_monotonic_us is None:
910                        self._first_commit_monotonic_us = commit_now_us
911                    self._history.append(committed_history_chunk)
912                    self._produced_audio_us += pending.duration_us
913                    self._prune_history_locked(commit_now_us)
914                await self._fanout_history_chunk_to_join_processors(committed_history_chunk)
915                if self._timeline_start_us is not None:
916                    elapsed_real_s = max(0.0, (commit_now_us - self._timeline_start_us) / 1_000_000)
917                    if elapsed_real_s - last_elapsed_update_s >= 1.0:
918                        last_elapsed_update_s = elapsed_real_s
919                        self.player._attr_elapsed_time = elapsed_real_s
920                        self.player._attr_elapsed_time_last_updated = time.time()
921                        self.player.update_state()
922
923        commit_task = asyncio.create_task(_commit_pending_chunks())
924        self._attach_task_exception_logger(commit_task, "commit_pending_chunks")
925        producer_stopped_cleanly = False
926        try:
927            await _produce_pending_chunks()
928            producer_stopped_cleanly = True
929        finally:
930            if producer_stopped_cleanly and not self._cancel_requested and not commit_task.done():
931                # Mark EOF so that catchup processors promoted after this
932                # point also get flushed (see _promote_join_catchup_processor).
933                self._producer_eof_sent = True
934                # Signal EOF to transform pipelines so ffmpeg flushes its
935                # internal buffers instead of blocking on the last read.
936                _, pipelines = await self._snapshot_active_pipelines()
937                for _, pipeline in pipelines:
938                    if pipeline.processor is not None and pipeline.config.requires_transform:
939                        with suppress(Exception):
940                            await pipeline.processor.write_eof()
941                # Producer finished normally; send a None sentinel so the
942                # consumer exits cleanly.  The queue may be full, so retry
943                # with a deadline before falling back to cancellation.
944                sentinel_sent = False
945                deadline = time.monotonic() + 1.0
946                while not sentinel_sent and not commit_task.done():
947                    try:
948                        pending_chunks.put_nowait(None)
949                        sentinel_sent = True
950                    except asyncio.QueueFull:
951                        if time.monotonic() >= deadline:
952                            break
953                        await asyncio.sleep(0.01)
954                if not sentinel_sent:
955                    commit_task.cancel()
956            else:
957                commit_task.cancel()
958            with suppress(asyncio.CancelledError, Exception):
959                await commit_task
960            # On clean EOF, wait for clients to finish playing their
961            # buffered audio before sending stream/end (which clears
962            # client buffers per the Sendspin spec). Skip this when a
963            # new playback is superseding this one, so skipping tracks is still fast.
964            if producer_stopped_cleanly and not self._cancel_requested:
965                try:
966                    await self._wait_for_buffer_drain()
967                except asyncio.CancelledError:
968                    # New playback interrupted the drain — treat as
969                    # non-clean stop so we skip group.stop() below
970                    # and let the new playback handle the transition.
971                    producer_stopped_cleanly = False
972            with suppress(Exception):
973                # Same condition as the group.stop() below, so we snapshot on exactly the
974                # paths where a group STOP - and therefore a freeze - is already emitted.
975                self._stop_push_stream(
976                    snapshot_progress=producer_stopped_cleanly and not self._cancel_requested,
977                )
978            await self._clear_join_catchup()
979            await self._clear_member_pipelines()
980            await self._reset_session_state()
981            # Only emit a group STOP when MA stream playback reached natural EOF.
982            # Skip this on cancellation/error paths to avoid stop-event races with transitions.
983            if producer_stopped_cleanly and not self._cancel_requested:
984                with suppress(Exception):
985                    await self.player.api.group.stop()
986
987    # -- Join injection --------------------------------------------------------
988
989    async def _inject_ready_join_historical(
990        self,
991        push_stream: PushStream,
992        pending_backlog: deque[_PendingChunk],
993        current_pcm: bytes,
994    ) -> bool:
995        """
996        Inject join-catchup historical audio once processor output reaches history end.
997
998        Join promotion lifecycle:
999        1. A catchup processor is fed historical PCM and new commits in parallel.
1000        2. Once the processor's output lag falls within _JOIN_PROMOTE_ARM_WINDOW_US
1001           of the history tail, promotion is "armed" and a target end timestamp is locked.
1002        3. Once output reaches the target (within _JOIN_PROMOTE_TOLERANCE_US), the
1003           catchup processor is promoted to the member's live DSP pipeline.
1004        4. If promotion doesn't complete within _JOIN_PROMOTION_TIMEOUT_S, it's aborted.
1005        """
1006        injected_any = False
1007        async with self._state_lock:
1008            items = list(self._join_catchup.items())
1009        for player_id, state in items:
1010            produced_output_us = state.processor.produced_output_us
1011            async with self._state_lock:
1012                current = self._join_catchup.get(player_id)
1013                if current is None or current.processor is not state.processor:
1014                    continue
1015                first_history_start_us = current.first_history_start_us
1016                fed_until_us = current.fed_until_us
1017                history_end_us = current.history_end_us
1018                promotion_target_end_us = current.promotion_target_end_us
1019                promotion_armed_monotonic_s = current.promotion_armed_monotonic_s
1020            if first_history_start_us is None or fed_until_us is None or history_end_us is None:
1021                continue
1022            max_ready_end_us = min(
1023                fed_until_us,
1024                first_history_start_us + max(0, produced_output_us),
1025            )
1026            if promotion_target_end_us is None:
1027                lag_to_tail_us = history_end_us - max_ready_end_us
1028                if lag_to_tail_us > _JOIN_PROMOTE_ARM_WINDOW_US:
1029                    continue
1030                async with self._state_lock:
1031                    current = self._join_catchup.get(player_id)
1032                    if current is None or current.processor is not state.processor:
1033                        continue
1034                    if current.promotion_target_end_us is None:
1035                        current.promotion_target_end_us = history_end_us
1036                        current.promotion_armed_monotonic_s = time.monotonic()
1037                    promotion_target_end_us = current.promotion_target_end_us
1038                    promotion_armed_monotonic_s = current.promotion_armed_monotonic_s
1039            target_end_us = promotion_target_end_us
1040            if (
1041                promotion_armed_monotonic_s is not None
1042                and time.monotonic() - promotion_armed_monotonic_s > _JOIN_PROMOTION_TIMEOUT_S
1043            ):
1044                self.player.logger.error(
1045                    "Join promotion timed out for %s after %.1fs; dropping join catchup",
1046                    player_id,
1047                    _JOIN_PROMOTION_TIMEOUT_S,
1048                )
1049                await self._stop_join_catchup(player_id)
1050                continue
1051            if max_ready_end_us + _JOIN_PROMOTE_TOLERANCE_US < target_end_us:
1052                continue
1053            inject_duration_us = target_end_us - first_history_start_us
1054            transformed_history = state.processor.pop_duration_us_or_pad(
1055                inject_duration_us, _JOIN_PROMOTE_TOLERANCE_US
1056            )
1057            if transformed_history is None:
1058                continue
1059            transformed_history = await self._pad_history_to_live_tail(
1060                state, target_end_us, transformed_history
1061            )
1062            pipeline = await self._sync_member_pipeline(player_id)
1063            # Split the blob into slices so push_stream can yield between encodes.
1064            frame_stride = (
1065                self._sendspin_pcm_format.bit_depth // 8
1066            ) * self._sendspin_pcm_format.channels
1067            slice_bytes = (
1068                int(self._sendspin_pcm_format.sample_rate * _PRODUCER_SLICE_US / 1_000_000)
1069                * frame_stride
1070            )
1071            for offset in range(0, len(transformed_history), slice_bytes):
1072                push_stream.prepare_historical_audio(
1073                    transformed_history[offset : offset + slice_bytes],
1074                    self._sendspin_pcm_format,
1075                    channel_id=pipeline.channel_id,
1076                    start_time_us=first_history_start_us if offset == 0 else None,
1077                )
1078            await self._prefeed_pending_backlog_for_join(state, current_pcm, pending_backlog)
1079            await self._promote_join_catchup_processor(player_id, pipeline, target_end_us)
1080            injected_any = True
1081        return injected_any
1082
1083    async def _pad_history_to_live_tail(
1084        self,
1085        state: _JoinCatchupState,
1086        target_end_us: int,
1087        transformed_history: bytes,
1088    ) -> bytes:
1089        """Append silence so joiner's channel_timing aligns with the live tail at promotion."""
1090        # target_end_us is locked from earlier and may lag the live tail by seconds.
1091        async with self._state_lock:
1092            live_tail_us = (
1093                self._history[-1].start_time_us + self._history[-1].duration_us
1094                if self._history
1095                else target_end_us
1096            )
1097        promotion_lag_us = max(0, live_tail_us - target_end_us)
1098        if promotion_lag_us <= 0:
1099            return transformed_history
1100        return transformed_history + state.processor.pad_and_skip(promotion_lag_us)
1101
1102    async def _prefeed_pending_backlog_for_join(
1103        self,
1104        state: _JoinCatchupState,
1105        current_pcm: bytes,
1106        pending_backlog: deque[_PendingChunk],
1107    ) -> None:
1108        """
1109        Push current chunk + queued pending chunks into join processor before promotion.
1110
1111        Between the last committed chunk and the next commit, there may be
1112        chunks already queued by the producer that the catchup processor hasn't
1113        seen yet.  Feeding them now avoids a gap in transformed audio after
1114        promotion.
1115        """
1116        await self._enqueue_join_pcm(state, current_pcm)
1117        for item in list(pending_backlog):
1118            await self._enqueue_join_pcm(state, item.pcm)
1119
1120    async def _fanout_history_chunk_to_join_processors(self, hist_chunk: _HistoryChunk) -> None:
1121        """Feed newly committed history chunk into all active join-catchup processors."""
1122        async with self._state_lock:
1123            items = list(self._join_catchup.items())
1124        for player_id, state in items:
1125            async with state.write_lock:
1126                # Read current state under lock.
1127                async with self._state_lock:
1128                    current = self._join_catchup.get(player_id)
1129                    if current is None or current.processor is not state.processor:
1130                        continue
1131                    previous_end_us = current.fed_until_us
1132                    first_history_start_us = current.first_history_start_us
1133                # Initialize first_history_start_us if this is the first chunk.
1134                if first_history_start_us is None:
1135                    first_history_start_us = hist_chunk.start_time_us
1136                    previous_end_us = first_history_start_us
1137                # Fill timeline gaps with silence.
1138                if previous_end_us is not None and hist_chunk.start_time_us > previous_end_us:
1139                    gap_us = hist_chunk.start_time_us - previous_end_us
1140                    silence = self._silence_for_duration_us(gap_us)
1141                    if silence:
1142                        await self._enqueue_join_pcm(state, silence)
1143                await self._enqueue_join_pcm(state, hist_chunk.pcm)
1144                # Write updated state back under lock.
1145                new_end_us = hist_chunk.start_time_us + hist_chunk.duration_us
1146                async with self._state_lock:
1147                    current = self._join_catchup.get(player_id)
1148                    if current is not None and current.processor is state.processor:
1149                        if current.first_history_start_us is None:
1150                            current.first_history_start_us = first_history_start_us
1151                        if current.fed_until_us is None:
1152                            current.fed_until_us = first_history_start_us
1153                        current.fed_until_us = new_end_us
1154                        current.history_end_us = new_end_us
1155
1156    async def _enqueue_join_pcm(
1157        self,
1158        state: _JoinCatchupState,
1159        pcm: bytes,
1160    ) -> None:
1161        """
1162        Enqueue PCM into a joining member writer queue.
1163
1164        Bails out immediately if the writer task is dead to avoid blocking
1165        the commit loop on a queue with no consumer.
1166        """
1167        if state.writer_task.done():
1168            return
1169        try:
1170            state.input_queue.put_nowait(pcm)
1171        except asyncio.QueueFull:
1172            if state.writer_task.done():
1173                return
1174            await state.input_queue.put(pcm)
1175
1176    # -- Member pipeline management --------------------------------------------
1177
1178    async def _refresh_member_mappings(self) -> None:
1179        """Re-evaluate per-member channel mapping and DSP requirements."""
1180        async with self._state_lock:
1181            if not self._mapping_dirty:
1182                return
1183            member_ids = tuple(self._members)
1184            self._mapping_dirty = False
1185        for member_id in member_ids:
1186            await self._sync_member_pipeline(member_id)
1187        # Keep leader pipeline in sync so leader DSP can be applied when required.
1188        await self._sync_member_pipeline(self.player.player_id)
1189
1190    async def _sync_member_pipeline(self, player_id: str) -> _MemberPipeline:
1191        """Create/update pipeline state for one member from current MA config."""
1192        config = self._get_pipeline_config_cached(player_id)
1193        release_processor: _BufferedFfmpegProcessor | None = None
1194        start_processor: _BufferedFfmpegProcessor | None = None
1195        async with self._state_lock:
1196            current = self._member_pipelines.get(player_id)
1197            if current is not None and current.config.signature == config.signature:
1198                return current
1199            if current and current.config.requires_transform:
1200                channel_id = current.channel_id if config.requires_transform else MAIN_CHANNEL
1201                release_processor = current.processor
1202            elif config.requires_transform:
1203                channel_id = self._get_or_create_preassigned_channel(player_id)
1204            else:
1205                channel_id = MAIN_CHANNEL
1206                self._preassigned_channels.pop(player_id, None)
1207            processor: _BufferedFfmpegProcessor | None = None
1208            if config.requires_transform:
1209                ffmpeg_obj = self._create_member_ffmpeg(config.filter_params)
1210                processor = _BufferedFfmpegProcessor(ffmpeg_obj, self._pcm_format)
1211                start_processor = processor
1212            pipeline = _MemberPipeline(
1213                player_id=player_id,
1214                channel_id=channel_id,
1215                config=config,
1216                processor=processor,
1217            )
1218            self._member_pipelines[player_id] = pipeline
1219        if start_processor is not None:
1220            try:
1221                await start_processor.start()
1222            except Exception as err:
1223                async with self._state_lock:
1224                    if (
1225                        self._member_pipelines.get(player_id) is not None
1226                        and self._member_pipelines[player_id].processor is start_processor
1227                    ):
1228                        self._member_pipelines.pop(player_id, None)
1229                with suppress(Exception):
1230                    await self._close_member_ffmpeg(start_processor)
1231                raise RuntimeError(f"Failed to start member DSP ffmpeg for {player_id}") from err
1232        if release_processor is not None:
1233            await self._close_member_ffmpeg(release_processor)
1234        return pipeline
1235
1236    def _get_pipeline_config_cached(
1237        self,
1238        player_id: str,
1239        *,
1240        force_refresh: bool = False,
1241    ) -> _PipelineConfig:
1242        """Return cached pipeline config for a player, calculating on cache miss."""
1243        if not force_refresh and (cached := self._pipeline_config_cache.get(player_id)) is not None:
1244            return cached
1245        config = self._read_pipeline_config(player_id)
1246        self._pipeline_config_cache[player_id] = config
1247        return config
1248
1249    def _read_pipeline_config(self, player_id: str) -> _PipelineConfig:
1250        """Read MA config and determine if member needs a dedicated DSP channel."""
1251        dsp_config = self.player.mass.config.get_player_dsp_config(player_id)
1252        dsp_enabled = bool(dsp_config.enabled)
1253        raw_output_channels = self.player.mass.config.get_raw_player_config_value(
1254            player_id,
1255            CONF_OUTPUT_CHANNELS,
1256            "stereo",
1257        )
1258        output_channels = str(raw_output_channels or "stereo").strip().lower()
1259        if output_channels not in {"stereo", "left", "right", "mono"}:
1260            output_channels = "stereo"
1261        try:
1262            output_format = self._get_member_output_format(player_id)
1263            output_plan = self.player.mass.streams.audio.get_player_output_plan(
1264                player_id,
1265                self._pcm_format,
1266                output_format,
1267                handoff_format=self._pcm_format,
1268            )
1269            filter_params = tuple(output_plan.filter_params)
1270        except Exception:
1271            filter_params = ()
1272            output_plan = None
1273        # a ComplexFilter (e.g. convolution) is never a plain string, so it always counts
1274        custom_filter_graph = any(
1275            not isinstance(param, str) or param.strip() for param in filter_params
1276        )
1277        requires_transform = dsp_enabled or output_channels != "stereo" or custom_filter_graph
1278        if (
1279            output_plan is not None
1280            and self._queue_id is not None
1281            and self._queue_session_id is not None
1282        ):
1283            self.player.mass.streams.audio_processing.update_output(
1284                output_plan.output_details.player_ids[0],
1285                output_plan,
1286                queue_id=self._queue_id,
1287                session_id=self._queue_session_id,
1288            )
1289        return _PipelineConfig(
1290            requires_transform=requires_transform,
1291            output_channels=output_channels,
1292            filter_params=filter_params,
1293        )
1294
1295    def _select_session_pcm_formats(self) -> tuple[AudioFormat, SendspinAudioFormat]:
1296        """
1297        Pick the session PCM format (MA-side + wire) from the leader's preferred format.
1298
1299        F32 is always used for DSP headroom. The sample rate follows the leader's
1300        preferred rate but is capped at 48 kHz when the leader's output codec is
1301        lossy — higher rates yield no perceivable quality gain there. Member
1302        clients with a different preferred rate up/down-sample in their own DSP
1303        step on the receiving side.
1304        """
1305        leader_output = self._get_member_output_format(self.player.player_id)
1306        sample_rate = int(leader_output.sample_rate) or _DEFAULT_PCM_FORMAT.sample_rate
1307        if leader_output.content_type in (ContentType.OPUS, ContentType.MP3, ContentType.AAC):
1308            sample_rate = min(sample_rate, _LOSSY_MAX_SAMPLE_RATE)
1309        pcm_format = AudioFormat(
1310            content_type=ContentType.PCM_F32LE,
1311            sample_rate=sample_rate,
1312            bit_depth=32,
1313            channels=2,
1314        )
1315        sendspin_pcm_format = SendspinAudioFormat(
1316            sample_rate=sample_rate,
1317            bit_depth=32,
1318            channels=2,
1319            sample_type="float",
1320        )
1321        return pcm_format, sendspin_pcm_format
1322
1323    def _get_member_output_format(self, player_id: str) -> AudioFormat:
1324        """
1325        Return the actual output AudioFormat for a group member.
1326
1327        Derives the format from the member's sendspin player role (preferred codec
1328        and format), falling back to the internal PCM format if unavailable.
1329        """
1330        provider = cast("SendspinProvider", self.player.provider)
1331        client = provider.server_api.get_client(player_id)
1332        if client is not None:
1333            for role in client.roles_by_family("player"):
1334                if isinstance(role, PlayerV1Role):
1335                    preferred_fmt = role.preferred_format
1336                    preferred_codec = role.preferred_codec
1337                    if preferred_fmt is not None and preferred_codec is not None:
1338                        if preferred_codec == SendspinAudioCodec.FLAC:
1339                            content_type = ContentType.FLAC
1340                        elif preferred_codec == SendspinAudioCodec.OPUS:
1341                            content_type = ContentType.OPUS
1342                        else:
1343                            content_type = ContentType.from_bit_depth(preferred_fmt.bit_depth)
1344                        return AudioFormat(
1345                            content_type=content_type,
1346                            sample_rate=preferred_fmt.sample_rate,
1347                            bit_depth=preferred_fmt.bit_depth,
1348                            channels=preferred_fmt.channels,
1349                        )
1350                elif isinstance(role, BridgePlayerRole):
1351                    fmt = role.preferred_format
1352                    if fmt is not None:
1353                        return AudioFormat(
1354                            content_type=ContentType.from_bit_depth(fmt.bit_depth),
1355                            sample_rate=fmt.sample_rate,
1356                            bit_depth=fmt.bit_depth,
1357                            channels=fmt.channels,
1358                        )
1359                    return AudioFormat(
1360                        content_type=ContentType.from_bit_depth(BRIDGE_BIT_DEPTH),
1361                        sample_rate=BRIDGE_SAMPLE_RATE,
1362                        bit_depth=BRIDGE_BIT_DEPTH,
1363                        channels=BRIDGE_CHANNELS,
1364                    )
1365        return self._pcm_format
1366
1367    def _get_or_create_preassigned_channel(self, player_id: str) -> UUID:
1368        """Return stable dedicated channel id for transform-required player."""
1369        if (channel_id := self._preassigned_channels.get(player_id)) is not None:
1370            return channel_id
1371        channel_id = uuid4()
1372        self._preassigned_channels[player_id] = channel_id
1373        return channel_id
1374
1375    # -- FFmpeg lifecycle ------------------------------------------------------
1376
1377    def _create_member_ffmpeg(self, filter_params: tuple[str | ComplexFilter, ...]) -> FFMpeg:
1378        """Create per-member FFMpeg for DSP pipeline."""
1379        return FFMpeg(
1380            audio_input="-",
1381            input_format=self._pcm_format,
1382            output_format=self._pcm_format,
1383            filter_params=list(filter_params),
1384        )
1385
1386    async def _transform_member_chunk(self, pipeline: _MemberPipeline, chunk: bytes) -> None:
1387        """Push one PCM chunk into a member DSP pipeline."""
1388        processor = pipeline.processor
1389        if processor is None:
1390            return
1391        await processor.push(chunk)
1392
1393    async def _read_member_chunk(
1394        self,
1395        pipeline: _MemberPipeline,
1396        duration_us: int,
1397    ) -> bytes | None:
1398        """Read one transformed chunk from a member DSP pipeline."""
1399        processor = pipeline.processor
1400        if processor is None or duration_us <= 0:
1401            return b""
1402        transformed = await processor.read_duration_us(duration_us)
1403        if not transformed:
1404            return None
1405        pipeline.ready = True
1406        return bytes(transformed)
1407
1408    async def _close_member_ffmpeg(self, processor: _BufferedFfmpegProcessor) -> None:
1409        """Close an ffmpeg processor, suppressing errors."""
1410        with suppress(Exception):
1411            await processor.close()
1412
1413    async def _clear_member_pipelines(self) -> None:
1414        """Release all member pipeline resources."""
1415        async with self._state_lock:
1416            pipelines = list(self._member_pipelines.values())
1417            self._member_pipelines.clear()
1418        for pipeline in pipelines:
1419            if pipeline.processor is not None:
1420                await self._close_member_ffmpeg(pipeline.processor)
1421
1422    # -- Push stream -----------------------------------------------------------
1423
1424    def _create_push_stream(self) -> PushStream:
1425        """Create PushStream with channel resolver for per-member routing."""
1426        return self.player.api.group.start_stream(channel_resolver=self._resolve_channel_for_player)
1427
1428    async def _wait_for_buffer_drain(self) -> None:
1429        """
1430        Wait for clients to finish playing buffered audio.
1431
1432        Called before stopping the push stream on natural EOF to prevent
1433        stream/end from clearing client buffers while audio is still playing.
1434
1435        Uses the push stream's public backpressure API with a zero buffer
1436        target to sleep until the clock catches up with all committed audio.
1437        Each internal sleep is capped at 1 second, so we loop until drained.
1438
1439        Raises asyncio.CancelledError if a new playback request interrupts.
1440        """
1441        ps = self._push_stream
1442        if ps is None or ps.is_stopped:
1443            return
1444        self.player.logger.debug("Waiting for client buffer drain before stream/end")
1445        # Safety timeout: never wait longer than the max buffer depth.
1446        deadline = time.monotonic() + (_PRODUCER_BUFFER_LIMIT_US / 1_000_000)
1447        while time.monotonic() < deadline:
1448            t0 = time.monotonic()
1449            await ps.sleep_to_limit_buffer(0)
1450            # sleep_to_limit_buffer returns immediately when the clock has
1451            # caught up with all committed audio (nothing left to drain).
1452            if time.monotonic() - t0 < 0.05:
1453                break
1454        self.player.logger.debug("Client buffer drain complete")
1455
1456    def _stop_push_stream(
1457        self, *, snapshot_progress: bool = False, keep_stream: bool = False
1458    ) -> None:
1459        """
1460        Stop the active PushStream.
1461
1462        :param snapshot_progress: Freeze the group's playback progress first. Pass this
1463            only for a natural end of stream, never for one being superseded.
1464        :param keep_stream: Clear buffered audio without ending the client stream.
1465        """
1466        ps = self._push_stream
1467        if ps is None or ps.is_stopped:
1468            return
1469        if snapshot_progress and (metadata_role := self.player._metadata_role) is not None:
1470            # The group can only resolve the live position while the stream is up; once
1471            # it is down the freeze can just re-emit the last anchor that was pushed.
1472            metadata_role.freeze_progress()
1473        if keep_stream:
1474            ps.clear()
1475        ps.stop(keep_stream=keep_stream)
1476
1477    async def _reset_session_state(self) -> None:
1478        """Drop all per-session playback state so the next session starts clean."""
1479        async with self._state_lock:
1480            self._push_stream = None
1481            self._playback_running = False
1482            self._timeline_start_us = None
1483            self._first_commit_monotonic_us = None
1484            self._produced_audio_us = 0
1485            self._history.clear()
1486            # Drop cached DSP decisions so next playback reflects latest config.
1487            self._pipeline_config_cache.clear()
1488
1489    def _resolve_channel_for_player(self, player_id: str) -> UUID:
1490        """Channel resolver callback for per-player routing."""
1491        pipeline = self._member_pipelines.get(player_id)
1492        if pipeline is not None:
1493            return pipeline.channel_id
1494        # Force a fresh config read for pending/unknown joiners so the very
1495        # first resolution (triggered by add_client) uses up-to-date DSP settings.
1496        force = player_id not in self._members and player_id != self.player.player_id
1497        config = self._get_pipeline_config_cached(player_id, force_refresh=force)
1498        if not config.requires_transform:
1499            return MAIN_CHANNEL
1500        return self._get_or_create_preassigned_channel(player_id)
1501
1502    # -- History ---------------------------------------------------------------
1503
1504    def _prune_history_locked(self, now_monotonic_us: int) -> None:
1505        """Drop old history chunks that are fully in the past."""
1506        if self._timeline_start_us is None or self._first_commit_monotonic_us is None:
1507            return
1508        elapsed_real_us = max(0, now_monotonic_us - self._first_commit_monotonic_us)
1509        source_now_us = self._timeline_start_us + elapsed_real_us
1510        cutoff_us = source_now_us - _HISTORY_KEEP_PAST_US
1511        while self._history and (
1512            self._history[0].start_time_us + self._history[0].duration_us <= cutoff_us
1513        ):
1514            self._history.popleft()
1515
1516    # -- PCM utilities ---------------------------------------------------------
1517
1518    @staticmethod
1519    def _duration_us(audio: bytes, audio_format: AudioFormat) -> int:
1520        """Compute chunk duration from PCM payload size."""
1521        bytes_per_sample = max(1, int(audio_format.bit_depth // 8))
1522        bytes_per_second = (
1523            int(audio_format.sample_rate) * bytes_per_sample * int(audio_format.channels)
1524        )
1525        if bytes_per_second <= 0:
1526            return 0
1527        return int((len(audio) / bytes_per_second) * 1_000_000)
1528
1529    def _silence_for_duration_us(self, duration_us: int) -> bytes:
1530        """Generate silent PCM with frame-aligned duration for the current session format."""
1531        if duration_us <= 0:
1532            return b""
1533        bytes_per_sample = max(1, int(self._pcm_format.bit_depth // 8))
1534        frame_size = bytes_per_sample * int(self._pcm_format.channels)
1535        samples = max(0, round((duration_us / 1_000_000) * int(self._pcm_format.sample_rate)))
1536        return b"\x00" * (samples * frame_size)
1537
1538    def _is_live_source(self, media: PlayerMedia) -> bool:
1539        """
1540        Return whether this media is fed to the server at playback pace.
1541
1542        Radio and live sources always are; a track is when its provider hands
1543        over the audio just-in-time (a queue flow of such items included, which
1544        the media type cannot express because it names the first item).
1545
1546        :param media: The media about to be played.
1547        """
1548        if media.media_type in _LIVE_MEDIA_TYPES:
1549            return True
1550        if not media.source_id or not media.queue_item_id:
1551            return False
1552        queue_item = self.player.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1553        return bool(
1554            queue_item and queue_item.streamdetails and queue_item.streamdetails.is_realtime
1555        )
1556