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