/
/
/
1"""Unified AirPlay/RAOP stream session logic for AirPlay devices."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections.abc import AsyncGenerator, Coroutine
8from contextlib import aclosing, suppress
9from typing import TYPE_CHECKING, Any
10
11from music_assistant_models.enums import ContentType, PlaybackState
12from music_assistant_models.errors import MusicAssistantError, PlayerCommandFailed
13
14from music_assistant.constants import CONF_SYNC_ADJUST
15from music_assistant.controllers.streams.audio_processing import get_media_session_id
16from music_assistant.helpers.ffmpeg import FFMpeg
17
18from .constants import (
19 AIRPLAY_CLOCK_READY_LEAD_MS,
20 AIRPLAY_CLOCK_READY_TIMEOUT_MS,
21 AIRPLAY_COLD_GROUP_START_LEAD_MS,
22 AIRPLAY_FEED_START_TIMEOUT,
23 AIRPLAY_GROUP_START_LEAD_MS,
24 AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS,
25 AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
26 AIRPLAY_LATE_JOIN_RING_MAX_BYTES,
27 AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
28 AIRPLAY_REPLACEMENT_EOF_TIMEOUT,
29 AIRPLAY_REPLACEMENT_POLL_INTERVAL,
30 AIRPLAY_SPLICE_LEAD_MARGIN_MS,
31 AIRPLAY_START_LEAD_MS,
32 ClockReadiness,
33 StreamingProtocol,
34)
35from .helpers import get_final_output_format
36from .stream import AirPlayStream
37
38if TYPE_CHECKING:
39 from music_assistant_models.media_items import AudioFormat
40
41 from music_assistant.models.player import PlayerMedia
42
43 from .player import AirPlayPlayer
44 from .provider import AirPlayProvider
45
46# What each readiness outcome means for the join anchor: a projection only moves
47# it when it clears the join floor, which otherwise carries the anchor alone, so
48# the note says which bound the outcome set. STALLED has no note because it never
49# reaches a join anchor - a stalled joiner is refused the join before that.
50_CLOCK_READINESS_NOTES: dict[ClockReadiness, str] = {
51 ClockReadiness.PROJECTED: "usable in {out:.2f}s; anchoring no earlier than that",
52 ClockReadiness.NOT_APPLICABLE: "runs on NTP timing, so there is none to wait for; "
53 "anchoring on the join floor",
54 ClockReadiness.UNREPORTED: "was not reported within {timeout:.1f}s (a slow device, or a "
55 "receiver that never answered); anchoring on the join floor",
56}
57# A locked clock projects an instant that has already passed - the common case,
58# since only a cold receiver is still probing - so its note reads back in time.
59_CLOCK_LOCKED_NOTE = "became usable {ago:.2f}s ago; anchoring on the join floor"
60
61
62class AirPlayStreamSession:
63 """Stream session (RAOP or AirPlay2) to one or more players."""
64
65 def __init__(
66 self,
67 airplay_provider: AirPlayProvider,
68 sync_clients: list[AirPlayPlayer],
69 pcm_format: AudioFormat,
70 media: PlayerMedia,
71 ) -> None:
72 """
73 Initialize AirPlayStreamSession.
74
75 :param airplay_provider: The AirPlay provider instance.
76 :param sync_clients: List of AirPlay players to stream to.
77 :param pcm_format: PCM format of the input stream.
78 :param media: Queue media that owns the stream session.
79 """
80 assert sync_clients
81 self.prov = airplay_provider
82 self.mass = airplay_provider.mass
83 self.pcm_format = pcm_format
84 self.media = media
85 self.sync_clients = sync_clients
86 self._audio_source_task: asyncio.Task[None] | None = None
87 self._player_ffmpeg: dict[str, FFMpeg] = {}
88 self._lock = asyncio.Lock()
89 self._ptp_lock = asyncio.Lock()
90 self.start_unix_ms: int = 0
91 self.start_time: float = 0.0
92 self.seconds_streamed: float = 0
93 # Parked in standby: the members stay connected but nothing is fed and
94 # their binaries hold until a START, so only a re-anchor (play_media)
95 # revives them. It outlives the group that parked it, which is why the
96 # session - not the group membership - owns this.
97 self.parked: bool = False
98 # Timing source for the whole session, decided once in start() and applied
99 # identically to every native AirPlay 2 member (and any late joiner) so a
100 # sync group can never mix shared-PTP and NTP members.
101 self.use_shared_ptp: bool = False
102 self._shared_ptp_resolved = False
103 self._ptp_degraded_warning_logged = False
104 # Raw PCM ring buffer for late joiners. When a late joiner arrives we
105 # send this buffer to prime its pipeline so it starts playing quickly
106 # instead of waiting for the full pipeline to fill from scratch.
107 # It must cover the write-head lead - how far the feed runs ahead of the
108 # audible position - because that is exactly the span a joiner's anchor
109 # maps into. The lead is not a constant of the protocol: it is the sum
110 # of every buffer downstream of this counter (see
111 # AIRPLAY_LATE_JOIN_RING_MIN_SECONDS), so it is measured per session and
112 # the ring is grown to match.
113 self._pcm_buffer = bytearray()
114 # Largest write-head lead this session has shown, and the ring size
115 # derived from it.
116 self._peak_lead_seconds: float = 0.0
117 # Raw byte sizes of the PCM actually on the wire. At 24-bit the binary
118 # is fed s32le carriers (bit_depth stays 24 for the ALAC encode), so
119 # sizes MUST come from the content type: bit_depth-derived sizes are
120 # 6 bytes/frame while the wire frames are 8 â slicing or timing the
121 # feed on those boundaries cuts mid-sample (loud noise on a late
122 # joiner's prime) and inflates the position clock by 4/3.
123 bytes_per_sample = {
124 ContentType.PCM_S16LE: 2,
125 ContentType.PCM_S24LE: 3,
126 ContentType.PCM_S32LE: 4,
127 ContentType.PCM_F32LE: 4,
128 }.get(pcm_format.content_type, pcm_format.bit_depth // 8)
129 self._pcm_frame_size = bytes_per_sample * pcm_format.channels
130 self._pcm_byte_rate = self._pcm_frame_size * pcm_format.sample_rate
131 self._pcm_buffer_max = self._ring_bytes_for(0.0)
132 # Bytes still to skip off the head of the live feed for a late joiner
133 # whose anchor lands ahead of the current write head, keyed by player id.
134 self._client_skip_bytes: dict[str, int] = {}
135 # Cumulative bytes fed to the members (absolute stream coordinate of
136 # the write head). Source chunking makes no frame-alignment promise,
137 # so any slice or skip of the live feed must be aligned against THIS
138 # counter â aligning lengths in isolation can still land mid-sample,
139 # which a joiner renders as pure static.
140 self._pcm_total_fed: int = 0
141 # Set once the first audio of a start cycle has been handed to the
142 # members, and also when the source ends without ever handing any over,
143 # so a waiter is never left holding on for a feed that is not coming.
144 self._feed_settled = asyncio.Event()
145
146 @property
147 def effective_start_time(self) -> float:
148 """
149 Return the session anchor adjusted for the reference member's clock shift.
150
151 ``start_time`` is the wall-clock anchor set at the last group (re)start.
152 A member's cliairplay can re-anchor its playout LATER after recovering
153 from a PCM starvation, shifting the group's real timeline; that shift is
154 tracked per member. The first sync client is taken as the reference and
155 its accumulated shift is added, so a late joiner maps its first sample to
156 where the group actually plays. Divergent per-member shifts make any
157 single reference imperfect; the current first client is the best
158 available choice and the reference transfers to the new first client when
159 the old one is removed.
160 """
161 if not self.sync_clients:
162 return self.start_time
163 reference_stream = self.sync_clients[0].stream
164 if reference_stream is None:
165 return self.start_time
166 return self.start_time + reference_stream.cumulative_shift_seconds
167
168 async def start(self, audio_source: AsyncGenerator[bytes]) -> None:
169 """
170 Connect every member and anchor synchronized playback.
171
172 Spawns and connects each member's CLI, wires the per-seek ffmpeg into its
173 persistent stdin and starts feeding audio, then commands one shared
174 audible start instant. On any failure the whole session is stopped so the
175 caller can fall back to a cold restart.
176 """
177 ap2_members = sum(1 for p in self.sync_clients if p.protocol == StreamingProtocol.AIRPLAY2)
178 if ap2_members:
179 # Resolve the timing source before calculating the audible anchor so
180 # a bounded daemon-readiness wait cannot consume the setup lead.
181 self.use_shared_ptp = await self._resolve_shared_ptp(ap2_members)
182 self._shared_ptp_resolved = True
183 position_ms = int((self.media.elapsed_time or 0) * 1000)
184 try:
185 async with asyncio.TaskGroup() as task_group:
186 for player in self.sync_clients:
187 task_group.create_task(
188 self._member_start_step(
189 player, "spawn its cli", self._start_client(player, self.use_shared_ptp)
190 )
191 )
192 await asyncio.gather(
193 *[
194 self._member_start_step(
195 player, "connect to its device", player.stream.wait_for_connection()
196 )
197 for player in self.sync_clients
198 if player.stream
199 ]
200 )
201 # The binary buffers stdin into its ring from process start; feed
202 # audio first and wait for every member to confirm it flowing, then
203 # anchor with a short lead. Readiness is fully event-driven
204 # (connected + audio), so no guessed setup time is needed; the
205 # binary bursts the receiver pre-fill after START.
206 self._audio_source_task = asyncio.create_task(self._audio_streamer(audio_source))
207 await self._wait_members_audio_present()
208 # Members of a group have to agree on one instant, and a freshly
209 # connected receiver needs about a second before it even starts
210 # probing â too late for the binary to raise its own commit floor,
211 # and a first start is the one case it will not correct afterwards.
212 # So a group anchor waits for the receivers to say when they can
213 # play, rather than trusting the lead to have covered it.
214 ready_at_unix_ms = await self._wait_members_clock_ready()
215 await self._start_members(
216 position_ms, self._anchor_start_unix_ms(ready_at_unix_ms=ready_at_unix_ms)
217 )
218 except asyncio.CancelledError:
219 await self.stop()
220 raise
221 except Exception as err:
222 # playback failed to start, cleanup. This runs for every failure. A
223 # per-member one has already been named where it happened, so here
224 # the line says the whole session went down with it; for a failure
225 # no single member owns, it is the only line there is.
226 self.prov.logger.warning(
227 "AirPlay start failed for a session of %d member(s): %s",
228 len(self.sync_clients),
229 err,
230 )
231 await self.stop()
232 # A member can fail for a specific, user-actionable reason (a device
233 # that needs its password configured, for example). That error must
234 # reach the caller intact instead of being flattened into the generic
235 # message; the TaskGroup above nests it inside an exception group.
236 if (specific := _first_music_assistant_error(err)) is not None:
237 raise specific from err
238 raise PlayerCommandFailed("Playback failed to start") from err
239
240 def can_replace(self, sync_clients: list[AirPlayPlayer], pcm_format: AudioFormat) -> bool:
241 """
242 Return whether this live session can absorb a new play_media warm.
243
244 A warm replacement needs the same member set, the same session PCM
245 format and a connected stream still taking audio on every member;
246 anything else takes the cold path.
247 """
248 if {p.player_id for p in sync_clients} != {p.player_id for p in self.sync_clients}:
249 return False
250 # the encoding matters as much as the depth here (a 24-bit session carries
251 # PCM_S32LE): replace() wires the new source into the session's declared format
252 if pcm_format != self.pcm_format:
253 return False
254 return all(
255 p.stream is not None and p.stream.accepts_audio and p.stream.connected
256 for p in self.sync_clients
257 )
258
259 async def replace(self, audio_source: AsyncGenerator[bytes], media: PlayerMedia) -> bool:
260 """
261 Warm-replace the playing media with a new source (seek/next-track).
262
263 Stops feeding old audio and kills each member's ffmpeg (never the
264 persistent cli stdin), flushes every member's live stream in place, then
265 feeds a fresh ffmpeg into the same stdin and anchors all members at one
266 shared instant. A group flushes every member and awaits all acks before
267 the shared start. Returns False when anything fails so the caller can
268 fall back to the cold path.
269 """
270 position_ms = int((media.elapsed_time or 0) * 1000)
271 try:
272 # Stop feeding old audio and drop each member's buffered ffmpeg output
273 # before flushing: the binary drains stdin to EAGAIN on FLUSH, so no
274 # bytes may be written between the old ffmpeg dying and the flush ack.
275 # Killing ffmpeg never closes the cli stdin (MA holds the write end),
276 # so the binary keeps its stdin reader alive across the seek.
277 if self._audio_source_task and not self._audio_source_task.done():
278 self._audio_source_task.cancel()
279 with suppress(asyncio.CancelledError):
280 await self._audio_source_task
281 for player in self.sync_clients:
282 if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
283 await ffmpeg.kill()
284 flushed = await asyncio.gather(
285 *[self._flush_member(player) for player in self.sync_clients]
286 )
287 if not all(flushed):
288 raise PlayerCommandFailed("warm flush was not acknowledged")
289 for player in self.sync_clients:
290 await self._start_player_ffmpeg(player, media)
291 self.media = media
292 # The stream position counter and the late-join prime buffer both
293 # describe the OLD timeline; restart them before the new source pumps.
294 self.seconds_streamed = 0
295 self._pcm_total_fed = 0
296 self._feed_settled.clear()
297 self._pcm_buffer.clear()
298 # The shared START below re-establishes start_time, so the per-client
299 # late-join skip counters and every member's accumulated starvation
300 # shift describe a timeline that no longer exists: reset them.
301 self._client_skip_bytes.clear()
302 self._reset_member_shifts()
303 self._audio_source_task = asyncio.create_task(self._audio_streamer(audio_source))
304 # Anchor only after every member confirms the new audio flowing;
305 # the live connection and clock survive the flush, so a short
306 # re-anchor lead replaces the full setup lead.
307 await self._wait_members_audio_present()
308 await self._start_members(position_ms, self._anchor_start_unix_ms(warm=True))
309 except asyncio.CancelledError:
310 raise
311 except Exception as err:
312 self.prov.logger.warning(
313 "Warm replacement failed (%r); falling back to a cold restart", err
314 )
315 return False
316 return True
317
318 async def standby(self) -> bool:
319 """
320 Park the session: stall every member but keep the connections alive.
321
322 The next play_media (resume or seek) replaces the media warm over the
323 live connections â the same coordinated flush-refill as seek/next.
324 Returns False when any member lacks a connected stream that still takes
325 audio, or its standby command cannot be delivered, so the caller can fall
326 back to a full stop.
327 """
328 if not all(
329 p.stream is not None and p.stream.accepts_audio and p.stream.connected
330 for p in self.sync_clients
331 ):
332 return False
333 if self._audio_source_task and not self._audio_source_task.done():
334 self._audio_source_task.cancel()
335 with suppress(asyncio.CancelledError):
336 await self._audio_source_task
337 for player in self.sync_clients:
338 if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
339 await ffmpeg.kill()
340 stream = player.stream
341 assert stream
342 try:
343 command_delivered = await stream.send_cli_command("ACTION=STANDBY")
344 except Exception as err:
345 self.prov.logger.warning(
346 "Could not park AirPlay player %s: %s", player.player_id, err
347 )
348 return False
349 if not command_delivered:
350 self.prov.logger.warning(
351 "Could not park AirPlay player %s: standby command was not delivered",
352 player.player_id,
353 )
354 return False
355 player.set_state_from_stream(state=PlaybackState.PAUSED, stream=stream)
356 # a parked session has no live timeline; the resume re-anchors it
357 self.seconds_streamed = 0
358 self._pcm_total_fed = 0
359 self._feed_settled.clear()
360 self._pcm_buffer.clear()
361 self._client_skip_bytes.clear()
362 self._reset_member_shifts()
363 self.parked = True
364 return True
365
366 async def stop(self) -> None:
367 """Stop playback and cleanup."""
368 if self._audio_source_task and not self._audio_source_task.done():
369 self._audio_source_task.cancel()
370 with suppress(asyncio.CancelledError):
371 await self._audio_source_task
372 await asyncio.gather(
373 *[self.remove_client(x, reason="session stop") for x in self.sync_clients],
374 )
375
376 async def remove_client(
377 self, airplay_player: AirPlayPlayer, reason: str = "client removed"
378 ) -> None:
379 """
380 Remove a sync client from the session.
381
382 :param airplay_player: The player to remove from the session.
383 :param reason: Short human-readable reason for the removal, used in teardown logs.
384 """
385 async with self._lock:
386 if airplay_player not in self.sync_clients:
387 return
388 self.sync_clients.remove(airplay_player)
389 await self._cleanup_after_removal(airplay_player, reason=reason)
390
391 async def stop_client(
392 self, airplay_player: AirPlayPlayer, reason: str = "stop_client called"
393 ) -> None:
394 """
395 Stop a client's stream and ffmpeg.
396
397 :param airplay_player: The player to stop.
398 :param reason: Short human-readable reason for the teardown, used in debug logs.
399 """
400 self.prov.logger.debug(
401 "AirPlay session teardown: session=%s client=%s reason=%s",
402 id(self),
403 airplay_player.player_id,
404 reason,
405 )
406 self._client_skip_bytes.pop(airplay_player.player_id, None)
407 ffmpeg = self._player_ffmpeg.pop(airplay_player.player_id, None)
408 # note that we use kill instead of graceful close here,
409 # because otherwise it can take a very long time for the process to exit.
410 if ffmpeg and not ffmpeg.closed:
411 await ffmpeg.kill()
412 if airplay_player.stream and airplay_player.stream.session == self:
413 airplay_player.stream.reset_reanchor_shift()
414 await airplay_player.stream.stop(force=True)
415
416 async def add_client(self, airplay_player: AirPlayPlayer) -> None: # noqa: PLR0915
417 """
418 Add a sync client to the session as a late joiner.
419
420 The joiner cannot honour an anchor in the past â cliairplay makes the
421 first post-START stdin byte audible exactly at the instant it acks and
422 then freezes the anchor, with no catch-up. So the anchor is commanded
423 just past the instant the receiver reports its clock becomes usable (and
424 never inside the join floor), the binary acks the instant it can truly
425 honour, and the stream position due at that instant is derived from the
426 group's effective (shift-adjusted) timeline. Depending on where that
427 position falls relative to the ring buffer the joiner is either primed
428 from the ring tail (position at or behind the write head) or has the
429 leading bytes of the live feed skipped (position ahead of the write
430 head), so its first audible sample lands exactly where the group is
431 playing.
432
433 Timing-source readiness, the receiver's clock projection and the START
434 ack are all awaited without the session lock so the rest of the group
435 keeps being fed meanwhile; all buffer and anchor math stays under the
436 lock so ``seconds_streamed``, the ring buffer and the per-client skip
437 counter stay consistent with the live feed.
438 """
439 await self._resolve_late_joiner_ptp(airplay_player)
440 async with self._lock:
441 if not self._session_is_live():
442 return
443 try:
444 await self._start_client(airplay_player, self.use_shared_ptp)
445 stream = airplay_player.stream
446 assert stream
447 await stream.wait_for_connection()
448 # A receiver starts probing its clock at connect, so how long it
449 # still needs is measurable before any anchor is announced. Waiting
450 # for that projection here â outside the session lock, so the rest
451 # of the group keeps being fed â lets the join anchor on the
452 # device's own readiness instead of a fixed guess.
453 readiness, ready_at_unix_ms = await stream.wait_clock_ready(
454 timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
455 )
456 except asyncio.CancelledError:
457 await self.stop_client(airplay_player, reason="late joiner start cancelled")
458 raise
459 except Exception as err:
460 self.prov.logger.warning(
461 "Late joiner %s: failed to connect pipeline: %s",
462 airplay_player.player_id,
463 err,
464 )
465 await self.stop_client(airplay_player, reason="late joiner connection failed")
466 return
467
468 if readiness is ClockReadiness.STALLED:
469 # The receiver never answered our clock, so it would render silence
470 # for as long as it stayed in the group while every other signal
471 # said it was playing. Better to leave it out: the group keeps
472 # playing and the player stays idle, which is the visible truth.
473 # The binary's own report names the device and the ports to check.
474 self.prov.logger.warning(
475 "Late joiner %s: not adding it to the group - its receiver never answered "
476 "the server's PTP clock, so it would render silence",
477 airplay_player.player_id,
478 )
479 await self.stop_client(airplay_player, reason="receiver clock stalled")
480 return
481
482 pcm_sample_size = self._pcm_byte_rate
483 frame_size = self._pcm_frame_size
484
485 def map_to(anchor_at: float, *, committed: bool) -> tuple[float, float, bytes, int]:
486 """
487 Map an anchor instant onto the live feed (call under the lock).
488
489 Snapshots the ring and derives which stream position the group
490 plays at ``anchor_at``, returning the anchor, the due feed
491 position, the prime slice and the live skip.
492
493 :param anchor_at: Instant at which the joiner's first delivered
494 sample becomes audible.
495 :param committed: True once the binary has acked the instant, which
496 fixes it. A due position the ring can no longer reach is then
497 covered with silence rather than by moving the anchor, because
498 moving an instant the binary already owns would offset the
499 joiner from the group by exactly that much.
500 """
501 # Snapshot the ring, which ends at the write head: it covers the
502 # stream positions [seconds_streamed - len(ring)/rate, seconds_streamed].
503 effective_start_time = self.effective_start_time
504 buffered_pcm = bytes(self._pcm_buffer)
505 due = anchor_at - effective_start_time
506 skip = 0
507 prime_slice = b""
508 if due <= self.seconds_streamed:
509 # The due position is at or behind the write head: prime
510 # the joiner from the ring so the live feed then continues
511 # seamlessly.
512 keep_bytes = int((self.seconds_streamed - due) * pcm_sample_size)
513 # Frame-align the prime START in ABSOLUTE stream
514 # coordinates. The prime must end exactly at the write head
515 # (the live feed continues byte-contiguously from there),
516 # so only the start may move: shift it forward to the next
517 # frame boundary (dropping under one frame, ~23 us).
518 start_abs = self._pcm_total_fed - keep_bytes
519 realign = (frame_size - start_abs % frame_size) % frame_size
520 keep_bytes -= realign
521 if realign:
522 self.prov.logger.debug(
523 "Late joiner %s: prime start realigned +%d bytes "
524 "(write head sits mid-frame)",
525 airplay_player.player_id,
526 realign,
527 )
528 # The ring's own head is frame-aligned only by accident - it is
529 # trimmed on byte overflow - so align it too before measuring
530 # how much of the requested prime it can really serve.
531 ring_head_abs = self._pcm_total_fed - len(buffered_pcm)
532 ring_realign = (frame_size - ring_head_abs % frame_size) % frame_size
533 servable = buffered_pcm[ring_realign:]
534 if keep_bytes > len(servable):
535 # The due position predates the ring: the write head ran
536 # further ahead of the audible position than the ring was
537 # sized for. Those samples are gone, so the join either
538 # starts a little later (anchor still free to move) or opens
539 # with that much silence (anchor already acked) - both keep
540 # the real content on the instant the group plays it, which
541 # dropping the missing head would not.
542 missing = keep_bytes - len(servable)
543 lead = self.seconds_streamed - (time.time() - effective_start_time)
544 if committed:
545 # One ring's worth is already far past any real
546 # shortfall, so it bounds what an implausible ack (one
547 # mapping back near the session start) can allocate.
548 pad = min(missing, self._pcm_buffer_max)
549 prime_slice = bytes(pad) + servable
550 # A clamped pad cannot reach the due position, so the
551 # joiner really does end up ahead of the group by the
552 # remainder. Say which of the two happened.
553 residual = (missing - pad) / pcm_sample_size
554 self.prov.logger.warning(
555 "Late joiner %s: the feed runs %.2fs ahead of the audible "
556 "position, past the %.2fs of audio kept for a join; opening "
557 "with %.2fs of silence to cover the missing head. %s Please "
558 "report this with a debug log.",
559 airplay_player.player_id,
560 lead,
561 len(servable) / pcm_sample_size,
562 pad / pcm_sample_size,
563 f"That still leaves it {residual:.2f}s ahead of the group."
564 if residual
565 else "The content itself still lands in sync; the joiner is "
566 "only audible that much late.",
567 )
568 else:
569 due += missing / pcm_sample_size
570 anchor_at = effective_start_time + due
571 prime_slice = servable
572 self.prov.logger.debug(
573 "Late joiner %s: due position predates the %.2fs ring "
574 "(feed runs %.2fs ahead); anchoring %.2fs later, on the "
575 "ring's oldest sample",
576 airplay_player.player_id,
577 len(servable) / pcm_sample_size,
578 lead,
579 missing / pcm_sample_size,
580 )
581 elif keep_bytes > 0:
582 prime_slice = servable[len(servable) - keep_bytes :]
583 else:
584 # The due position is ahead of the write head: skip that
585 # many bytes off the head of the live feed so the joiner's
586 # first delivered byte is the sample due at the anchor.
587 skip = int((due - self.seconds_streamed) * pcm_sample_size)
588 # The first delivered live byte must land on a frame
589 # boundary in ABSOLUTE stream coordinates (round the skip
590 # up, under one frame).
591 first_abs = self._pcm_total_fed + skip
592 skip += (frame_size - first_abs % frame_size) % frame_size
593 return anchor_at, due, prime_slice, skip
594
595 now = time.time()
596 clock_out = ready_at_unix_ms / 1000 - now
597 if readiness is ClockReadiness.PROJECTED and clock_out <= 0:
598 clock_note = _CLOCK_LOCKED_NOTE.format(ago=abs(clock_out))
599 else:
600 clock_note = _CLOCK_READINESS_NOTES.get(readiness, "").format(
601 out=clock_out,
602 timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000,
603 )
604 self.prov.logger.debug(
605 "Late joiner %s: receiver clock %s",
606 airplay_player.player_id,
607 clock_note,
608 )
609 async with self._lock:
610 if not self._session_is_live():
611 await self.stop_client(airplay_player, reason="session ended during late join")
612 return
613 # cliairplay makes the first post-START stdin byte audible exactly at
614 # the instant it acks, so the anchor is commanded first and the
615 # content mapped onto the ack afterwards. It sits just past the
616 # instant the receiver's clock becomes usable; the floor is only a
617 # lower bound, and carries the whole anchor when no projection came.
618 min_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
619 anchor_at = now + min_headroom
620 if ready_at_unix_ms:
621 anchor_at = max(
622 anchor_at,
623 (ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS) / 1000,
624 )
625 requested_at, fed_pos_due = map_to(anchor_at, committed=False)[:2]
626 start_unix_ms = int(requested_at * 1000)
627 sync_adjust = airplay_player.config.get_value(CONF_SYNC_ADJUST, 0)
628 adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
629 position_ms = int(((self.media.elapsed_time or 0) + fed_pos_due) * 1000)
630
631 try:
632 # Anchor the joiner BEFORE feeding the prime: pre-START the binary
633 # only buffers stdin into its bounded ring and sends nothing, so a
634 # prime longer than that ring would wedge that write â and, through
635 # the session lock, stall the whole group's feed. Anchored, the
636 # binary drains the prime as it streams in. The START does not
637 # re-anchor the session timeline (the group keeps playing). Its ack
638 # is awaited WITHOUT the session lock: the binary can hold that ack
639 # until its receiver clock verification resolves, which would
640 # otherwise starve every other member's feed for that whole wait.
641 actual = await stream.start(start_unix_ms + adjust_ms, position_ms, join=True)
642 except asyncio.CancelledError:
643 await self.stop_client(airplay_player, reason="late joiner start cancelled")
644 raise
645 except Exception as err:
646 self.prov.logger.warning(
647 "Late joiner %s: failed to start/prime pipeline: %r",
648 airplay_player.player_id,
649 err,
650 )
651 await self.stop_client(airplay_player, reason="late joiner start/prime failed")
652 return
653
654 try:
655 async with self._lock:
656 if not self._session_is_live():
657 await self.stop_client(airplay_player, reason="session ended during late join")
658 return
659 if airplay_player.stream is not stream or not stream.running:
660 await self.stop_client(
661 airplay_player, reason="late joiner stopped during start"
662 )
663 return
664 # Map the content onto the instant the binary acked â its
665 # verified truth. The ring is re-snapshotted here, so the mapping
666 # covers whatever the group was fed while the ack was outstanding.
667 acked_at = (actual - adjust_ms) / 1000
668 start_at, fed_pos_due, prime, skip_bytes = map_to(acked_at, committed=True)
669 self._client_skip_bytes[airplay_player.player_id] = skip_bytes
670 # An instant that moved carries the content mapped onto it
671 # further into the stream than the position sent with the
672 # command, so progress is reported against the sample that
673 # actually lands on the anchor.
674 acked_position_ms = int(((self.media.elapsed_time or 0) + fed_pos_due) * 1000)
675 if acked_position_ms != position_ms:
676 stream.rebase_position(acked_position_ms)
677
678 self.prov.logger.debug(
679 "Late joiner %s: priming %.2fs, skipping %.2fs, stream_pos=%.2fs, "
680 "fed_pos_due=%.2fs, start_at is %.2fs from now, acked %+d ms off the "
681 "commanded instant (min_headroom=%.2fs, effective_shift=%.2fs, "
682 "write_head_lead=%.2fs, peak=%.2fs, ring=%.2fs)",
683 airplay_player.player_id,
684 len(prime) / pcm_sample_size,
685 skip_bytes / pcm_sample_size,
686 self.seconds_streamed,
687 fed_pos_due,
688 start_at - time.time(),
689 int(acked_at * 1000) - start_unix_ms,
690 min_headroom,
691 self.effective_start_time - self.start_time,
692 self.seconds_streamed - (time.time() - self.effective_start_time),
693 self._peak_lead_seconds,
694 self._pcm_buffer_max / pcm_sample_size,
695 )
696
697 if prime:
698 try:
699 await self._write_chunk_to_player(airplay_player, prime)
700 except Exception as err:
701 self.prov.logger.warning(
702 "Late joiner %s: failed to start/prime pipeline: %r",
703 airplay_player.player_id,
704 err,
705 )
706 await self.stop_client(
707 airplay_player, reason="late joiner start/prime failed"
708 )
709 return
710
711 if airplay_player not in self.sync_clients:
712 self.sync_clients.append(airplay_player)
713 if (
714 airplay_player.protocol == StreamingProtocol.AIRPLAY2
715 and not self.use_shared_ptp
716 ):
717 ap2_members = sum(
718 1
719 for player in self.sync_clients
720 if player.protocol == StreamingProtocol.AIRPLAY2
721 )
722 self._warn_degraded_shared_ptp(ap2_members)
723 except asyncio.CancelledError:
724 # the joiner is anchored and audible from here on, so a cancellation
725 # before it is part of the session must still tear it down
726 await self.stop_client(airplay_player, reason="late joiner start cancelled")
727 raise
728
729 self.prov.logger.debug(
730 "Late joiner %s: started after %.2fs",
731 airplay_player.player_id,
732 time.time() - now,
733 )
734
735 def _ring_bytes_for(self, lead_seconds: float) -> int:
736 """
737 Return the late-join ring size that covers a given write-head lead.
738
739 :param lead_seconds: Lead the ring has to span, before margin.
740 """
741 seconds = max(
742 lead_seconds + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
743 AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
744 )
745 size = min(int(seconds * self._pcm_byte_rate), AIRPLAY_LATE_JOIN_RING_MAX_BYTES)
746 # Keep it a whole number of frames so it can bound a silence pad without
747 # knocking the prime off its frame boundaries.
748 return size - size % self._pcm_frame_size
749
750 def _observe_write_head_lead(self) -> None:
751 """
752 Track how far the feed runs ahead of the audible position (call under the lock).
753
754 A joiner's anchor maps into exactly this span, so the ring is grown to
755 the largest lead the session has shown. It only ever grows: shrinking it
756 mid-session would discard history a joiner still needs, and the lead is
757 at its largest right after the anchor is placed - the whole downstream
758 pipeline is full by then - so the ring is sized long before any late
759 join can arrive.
760 """
761 if self.start_time <= 0:
762 return # not anchored yet, so there is no audible position to measure against
763 now = time.time()
764 if now < self.effective_start_time:
765 # Still inside the start lead: everything fed is ahead of an anchor
766 # that has not arrived, which would read as a lead seconds larger
767 # than the pipeline really holds.
768 return
769 lead = self.seconds_streamed - (now - self.effective_start_time)
770 if lead <= self._peak_lead_seconds:
771 return
772 self._peak_lead_seconds = lead
773 self._pcm_buffer_max = self._ring_bytes_for(lead)
774
775 def _session_is_live(self) -> bool:
776 """Return whether the session still plays a joinable timeline (call under the lock)."""
777 if not self.sync_clients:
778 return False
779 reference = self.sync_clients[0]
780 reference_stream = reference.stream
781 # A stream that has been sent its audio EOF keeps running while it plays
782 # out, but it is on its way to exiting and can never be fed again, so a
783 # joiner would land in a session that is ending.
784 if reference_stream is None or not reference_stream.accepts_audio:
785 return False
786 # A parked (standby) session keeps every member's stream running while
787 # its timeline is gone - the anchor is stale and nothing is being fed -
788 # so only a member that is actually playing can absorb a joiner.
789 return reference.playback_state == PlaybackState.PLAYING
790
791 async def _resolve_shared_ptp(self, ap2_members: int | None = None) -> bool:
792 """
793 Decide, once per session, whether members attach to the shared PTP daemon.
794
795 The decision is session-wide and gated on the daemon actually being ready
796 (bound to 319/320 with its control channel open), not merely spawned, so a
797 group start cannot race the daemon and end up mixing PTP and NTP members.
798
799 :param ap2_members: Number of native AirPlay 2 members that will use the
800 decision. Defaults to the current session members.
801 :return: True if every native AirPlay 2 member should use the shared PTP
802 clock; False to degrade the whole session consistently (no member
803 attaches to the daemon).
804 """
805 if ap2_members is None:
806 ap2_members = sum(
807 1 for p in self.sync_clients if p.protocol == StreamingProtocol.AIRPLAY2
808 )
809 if not ap2_members:
810 return False
811 if await self.prov.wait_ptp_daemon_ready():
812 return True
813 # Daemon not ready: keep the group coherent by attaching no one to the
814 # shared clock. A lone native AP2 player can still self-bind its own PTP
815 # with no partner to drift against; only a real multi-room group loses
816 # tight sync, so warn just for that case.
817 self._warn_degraded_shared_ptp(ap2_members)
818 return False
819
820 async def _resolve_late_joiner_ptp(self, airplay_player: AirPlayPlayer) -> None:
821 """Resolve the session timing source before its first late AirPlay 2 join."""
822 if airplay_player.protocol != StreamingProtocol.AIRPLAY2:
823 return
824 async with self._ptp_lock:
825 if self._shared_ptp_resolved:
826 return
827 async with self._lock:
828 ap2_members = sum(
829 1
830 for player in self.sync_clients
831 if player.protocol == StreamingProtocol.AIRPLAY2
832 ) + (airplay_player not in self.sync_clients)
833 self.use_shared_ptp = await self._resolve_shared_ptp(ap2_members)
834 self._shared_ptp_resolved = True
835
836 def _warn_degraded_shared_ptp(self, ap2_members: int) -> None:
837 """Warn once when multiple AirPlay 2 members cannot share the PTP clock."""
838 if ap2_members <= 1 or self._ptp_degraded_warning_logged:
839 return
840 self._ptp_degraded_warning_logged = True
841 self.prov.logger.warning(
842 "Shared PTP clock daemon not ready - native AirPlay 2 multi-room sync "
843 "for this group of %d players is degraded and members may drift. The "
844 "server likely cannot bind the privileged PTP ports (UDP 319/320); "
845 "running without root or CAP_NET_BIND_SERVICE is the common cause.",
846 ap2_members,
847 )
848
849 async def _cleanup_after_removal(
850 self, airplay_player: AirPlayPlayer, reason: str = "client removed"
851 ) -> None:
852 """
853 Clean up processes and state after a client has been removed from sync_clients.
854
855 :param airplay_player: The player whose processes should be stopped.
856 :param reason: Short human-readable reason, forwarded to stop_client for logging.
857 """
858 stream = airplay_player.stream
859 if stream is not None and stream.session != self:
860 stream = None
861 await self.stop_client(airplay_player, reason=reason)
862 # Only set IDLE if the player's stream still belongs to this session,
863 # otherwise a re-add to a new session may have already set a new state.
864 if stream is not None:
865 airplay_player.set_state_from_stream(PlaybackState.IDLE, stream=stream)
866 # Re-check sync_clients under the lock to avoid racing with add_client.
867 async with self._lock:
868 should_stop = not self.sync_clients
869 if should_stop:
870 await self.stop()
871
872 async def _audio_streamer(self, audio_source: AsyncGenerator[bytes]) -> None:
873 """Stream audio to all players."""
874 stream_error: BaseException | None = None
875 try:
876 # the loop below leaves early once the clients are gone; closing the source
877 # from here releases its decoders instead of waiting on the garbage collector
878 async with aclosing(audio_source):
879 async for chunk in audio_source:
880 if not self.sync_clients:
881 break
882
883 has_running_clients = await self._write_chunk_to_all_players(chunk)
884 if not has_running_clients:
885 self.prov.logger.debug(
886 "No running clients remaining, stopping audio streamer"
887 )
888 break
889 except asyncio.CancelledError:
890 self.prov.logger.debug("Audio streamer cancelled after %.1fs", self.seconds_streamed)
891 raise
892 except Exception as err:
893 stream_error = err
894 self.prov.logger.error(
895 "Audio source error after %.1fs of streaming: %s",
896 self.seconds_streamed,
897 err,
898 exc_info=err,
899 )
900 finally:
901 # a source that ends - or is cancelled - without handing anything
902 # over settles the question for a start still waiting on the feed
903 self._feed_settled.set()
904 if stream_error:
905 self.prov.logger.warning(
906 "Stream ended prematurely due to error - notifying players"
907 )
908 # A source that ends is not the same thing as a stream that is over: a
909 # seek or a next-track ends this one while the queue is already loading
910 # the stream that takes over. Closing the binary's stdin there would end
911 # the stream for good - it cannot be reopened - and cost that replacement
912 # a full cold restart. Everywhere else the EOF is what ends playback: the
913 # binary plays out, reports eof and exits, and only that makes the player
914 # report idle, which a queue waiting to restart its flow depends on.
915 end_of_stream = not self._replacement_expected()
916 async with self._lock:
917 await asyncio.gather(
918 *[
919 self._retire_player_ffmpeg(x, end_of_stream=end_of_stream)
920 for x in self.sync_clients
921 if x.stream and x.stream.running
922 ],
923 return_exceptions=True,
924 )
925 if not end_of_stream:
926 await self._end_stream_if_no_replacement_lands()
927
928 async def _write_chunk_to_all_players(self, chunk: bytes) -> bool:
929 """
930 Write a chunk to all connected players.
931
932 :return: True if there are still running clients, False otherwise.
933 """
934 async with self._lock:
935 sync_clients = [x for x in self.sync_clients if x.stream and x.stream.running]
936 if not sync_clients:
937 return False
938
939 # Update seconds_streamed and ring buffer under the lock so
940 # add_client always reads consistent values.
941 self.seconds_streamed += len(chunk) / self._pcm_byte_rate
942 self._pcm_total_fed += len(chunk)
943 self._feed_settled.set()
944 self._observe_write_head_lead()
945 self._pcm_buffer.extend(chunk)
946 overflow = len(self._pcm_buffer) - self._pcm_buffer_max
947 if overflow > 0:
948 del self._pcm_buffer[:overflow]
949
950 # Write chunk to all players
951 write_tasks = [self._write_chunk_to_player(x, chunk) for x in sync_clients if x.stream]
952 results = await asyncio.gather(*write_tasks, return_exceptions=True)
953
954 # Check for write errors or timeouts
955 players_to_remove: list[tuple[AirPlayPlayer, str]] = []
956 for i, result in enumerate(results):
957 if i >= len(sync_clients):
958 continue
959 player = sync_clients[i]
960
961 if isinstance(result, TimeoutError):
962 self.prov.logger.warning(
963 "Removing player %s from session: stopped reading data (write timeout)",
964 player.player_id,
965 )
966 players_to_remove.append((player, "audio write timeout"))
967 elif isinstance(result, Exception):
968 self.prov.logger.warning(
969 "Removing player %s from session due to write error: %s",
970 player.player_id,
971 result,
972 )
973 players_to_remove.append((player, f"audio write error: {result}"))
974
975 # Remove failed players from sync_clients immediately under the lock
976 # so they are excluded from future write cycles. Only defer process
977 # cleanup (_cleanup_after_removal) â this prevents fire-and-forget
978 # remove_client calls from racing with a subsequent add_client when
979 # a player is being moved between groups.
980 for player, removal_reason in players_to_remove:
981 if player in self.sync_clients:
982 self.sync_clients.remove(player)
983 self.mass.create_task(self._cleanup_after_removal(player, reason=removal_reason))
984
985 remaining_clients = len(sync_clients) - len(players_to_remove)
986 return remaining_clients > 0
987
988 async def _write_chunk_to_player(self, airplay_player: AirPlayPlayer, chunk: bytes) -> None:
989 """Write audio chunk to a player's ffmpeg process."""
990 player_id = airplay_player.player_id
991 # Drain any pending late-join skip first: a joiner anchored ahead of the
992 # write head must drop that many leading bytes of the live feed so its
993 # first delivered byte is the sample due at its anchor.
994 if skip := self._client_skip_bytes.get(player_id, 0):
995 if skip >= len(chunk):
996 self._client_skip_bytes[player_id] = skip - len(chunk)
997 return
998 chunk = chunk[skip:]
999 self._client_skip_bytes[player_id] = 0
1000 if ffmpeg := self._player_ffmpeg.get(player_id):
1001 if ffmpeg.closed:
1002 return
1003 await asyncio.wait_for(ffmpeg.write(chunk), timeout=35.0)
1004
1005 def _replacement_expected(self) -> bool:
1006 """
1007 Return whether the queue is loading a stream that takes this session over.
1008
1009 A seek or a next-track starts the new stream while the old one is still
1010 playing: the queue rotates its stream session at the beginning of that,
1011 well before the play_media carrying it arrives, and the flow stream it
1012 supersedes ends as soon as it notices. A flow that ends on its own -
1013 the queue played out, the source failed, or it broke off to be restarted
1014 once the player reports idle - leaves the queue between transitions.
1015 """
1016 queue_id = self.media.source_id
1017 session_id = get_media_session_id(self.media)
1018 if not queue_id or not session_id:
1019 return False
1020 queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
1021 return (
1022 queue_data is not None
1023 and queue_data.transitioning
1024 and queue_data.session_id != session_id
1025 )
1026
1027 async def _retire_player_ffmpeg(
1028 self, airplay_player: AirPlayPlayer, *, end_of_stream: bool
1029 ) -> None:
1030 """
1031 Retire a member's ffmpeg now that the source feeding it has ended.
1032
1033 :param airplay_player: The member whose ffmpeg is retired.
1034 :param end_of_stream: Whether this ends the stream for good. The audio the
1035 ffmpeg still holds is then handed over and the binary's stdin closed
1036 behind it, which makes it play out, report eof and exit. A session a
1037 replacement is coming for drops that audio instead: the flush the
1038 replacement opens with discards it anyway, and no byte may reach the
1039 binary between the old ffmpeg dying and that flush.
1040 """
1041 if not (ffmpeg := self._player_ffmpeg.pop(airplay_player.player_id, None)):
1042 return
1043 if not end_of_stream:
1044 await ffmpeg.kill()
1045 return
1046 await ffmpeg.write_eof()
1047 await ffmpeg.wait_with_timeout(30)
1048 if airplay_player.stream:
1049 await airplay_player.stream.write_audio_eof()
1050
1051 async def _end_stream_if_no_replacement_lands(self) -> None:
1052 """
1053 Deliver a withheld end of stream once no replacement is coming for it.
1054
1055 This runs on the audio streamer's own task, which every route that takes
1056 the session over - a warm replace, a park, a stop - cancels before it
1057 touches the members, so getting past the wait below means no replacement
1058 ever claimed the session. The queue is watched rather than a fixed time
1059 waited out: it clears its transition on any failure between rotating its
1060 stream session and the play_media that carries the replacement, which
1061 says one is never coming, while a slow load keeps it set and must not be
1062 cut short. Nothing else can end this stream - without the EOF the binary
1063 never plays out, never reports eof, and the player keeps reporting
1064 playback until the user commands something else.
1065 """
1066 deadline = time.monotonic() + AIRPLAY_REPLACEMENT_EOF_TIMEOUT
1067 while self._replacement_expected() and (left := deadline - time.monotonic()) > 0:
1068 await asyncio.sleep(min(AIRPLAY_REPLACEMENT_POLL_INTERVAL, left))
1069 self.prov.logger.warning(
1070 "No replacement stream claimed the AirPlay session of %s - %s; "
1071 "ending it so the player can report idle",
1072 self.media.source_id,
1073 f"it is still loading one after {AIRPLAY_REPLACEMENT_EOF_TIMEOUT:.0f}s"
1074 if self._replacement_expected()
1075 else "the queue ended that transition without one",
1076 )
1077 async with self._lock:
1078 # A member that joined while the EOF was withheld holds a fresh
1079 # ffmpeg, and that process owns its own handle on the same cli
1080 # stdin: closing only this end would leave the pipe open and the
1081 # binary still waiting on it.
1082 for player in self.sync_clients:
1083 if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
1084 await ffmpeg.kill()
1085 await asyncio.gather(
1086 *[
1087 stream.write_audio_eof()
1088 for player in self.sync_clients
1089 if (stream := player.stream) and stream.accepts_audio
1090 ],
1091 return_exceptions=True,
1092 )
1093
1094 async def _member_start_step(
1095 self, airplay_player: AirPlayPlayer, step: str, awaitable: Coroutine[Any, Any, None]
1096 ) -> None:
1097 """
1098 Run one per-member step of a group start, naming the member if it fails.
1099
1100 A group start fans its members out over a task group and a gather, both
1101 of which collapse into a single exception at the caller - so with five
1102 speakers connecting, nothing in the log says which one failed. That,
1103 with whatever reason its binary reported, is the whole diagnostic.
1104
1105 :param airplay_player: The member the step belongs to.
1106 :param step: What the member was doing, for the failure message.
1107 :param awaitable: The step to run.
1108 """
1109 try:
1110 await awaitable
1111 except asyncio.CancelledError:
1112 raise
1113 except Exception as err:
1114 self.prov.logger.warning(
1115 "AirPlay group start: %s failed to %s: %s",
1116 airplay_player.display_name,
1117 step,
1118 err,
1119 )
1120 raise
1121
1122 async def _start_client(self, airplay_player: AirPlayPlayer, use_shared_ptp: bool) -> None:
1123 """
1124 Connect a CLI process and start its ffmpeg for a single client.
1125
1126 :param airplay_player: The player to start streaming to.
1127 :param use_shared_ptp: The session-wide shared-PTP decision applied to
1128 this member so the whole group shares one timing source.
1129 """
1130 # joining a session supersedes any pending automatic group re-join
1131 airplay_player.cancel_group_rejoin()
1132 airplay_player.release_foreign_mute_latch()
1133 # Held from the decision to displace whatever is published until the new
1134 # process is connected and published, so a Sendspin bridge start cannot
1135 # put a second cli process on the same receiver in between.
1136 async with airplay_player.stream_spawn_lock:
1137 if airplay_player.stream:
1138 # Stopped unconditionally, not just while it reads as running: a
1139 # stream stops reporting that the moment its own stop() starts,
1140 # while its process can still be on the receiver. stop() is
1141 # idempotent, so this joins a teardown already under way and
1142 # returns at once for one that finished.
1143 await airplay_player.stream.stop()
1144 stream_pcm_format = airplay_player.get_stream_pcm_format(self.pcm_format)
1145 airplay_player.stream = AirPlayStream(airplay_player, pcm_format=stream_pcm_format)
1146 airplay_player.stream.session = self
1147 await airplay_player.stream.connect(use_shared_ptp)
1148 # Wiring the audio producer to the cli stdin belongs to the same
1149 # claim: a displacement landing between the connect and this would
1150 # leave an ffmpeg feeding a process that is already gone, with
1151 # nothing tracking it to clean up.
1152 await self._start_player_ffmpeg(airplay_player, self.media)
1153
1154 def _anchor_start_unix_ms(self, *, warm: bool = False, ready_at_unix_ms: int = 0) -> int:
1155 """
1156 Return the shared audible-start instant for a readiness-confirmed start.
1157
1158 :param warm: True for a warm re-start over live connections (seek/next/
1159 resume-from-park). Members on the splice timeline report a
1160 minimum warm lead â their queued audio plays out before the new
1161 content can begin â and the shared anchor must sit beyond the
1162 largest member value so every member splices at the same instant.
1163 :param ready_at_unix_ms: Latest instant at which a member's receiver
1164 clock becomes usable, as the binaries reported it. The anchor never
1165 lands before it. 0 when no member reported a projection, leaving the
1166 lead below as the whole anchor.
1167 """
1168 if len(self.sync_clients) == 1:
1169 lead_ms = AIRPLAY_START_LEAD_MS
1170 elif warm:
1171 lead_ms = AIRPLAY_GROUP_START_LEAD_MS
1172 else:
1173 # Cold group start: cover the members' receiver-side clock
1174 # acquisition (see AIRPLAY_COLD_GROUP_START_LEAD_MS).
1175 lead_ms = AIRPLAY_COLD_GROUP_START_LEAD_MS
1176 anchor = int(time.time() * 1000) + lead_ms
1177 if ready_at_unix_ms:
1178 anchor = max(anchor, ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS)
1179 if not warm:
1180 return anchor
1181 # Splice-timeline members honor the commanded instant only when that
1182 # instant (plus their sync_adjust) lands beyond their queued audio.
1183 # A NEGATIVE sync_adjust moves a member's commanded instant earlier,
1184 # eating into the lead, so it must be added to that member's
1185 # requirement â otherwise the first round can never succeed for that
1186 # member and every group start pays a corrective round.
1187 member_requirement = 0
1188 for player in self.sync_clients:
1189 stream = player.stream
1190 if stream is None or stream.warm_lead_ms <= 0:
1191 continue
1192 sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1193 adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1194 member_requirement = max(member_requirement, stream.warm_lead_ms - min(0, adjust_ms))
1195 if member_requirement > 0:
1196 anchor = max(
1197 anchor,
1198 int(time.time() * 1000) + member_requirement + AIRPLAY_SPLICE_LEAD_MARGIN_MS,
1199 )
1200 for player in self.sync_clients:
1201 stream = player.stream
1202 if stream is None or stream.flushed_head_unix_ms <= 0:
1203 continue
1204 sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1205 adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1206 # The member's commanded instant is anchor + adjust; it must clear
1207 # the member's frozen head with margin for the command round-trip.
1208 anchor = max(
1209 anchor,
1210 stream.flushed_head_unix_ms - adjust_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS,
1211 )
1212 return anchor
1213
1214 async def _wait_members_audio_present(self) -> None:
1215 """
1216 Wait until every member's binary reports the new audio flowing.
1217
1218 A binary can only report audio once it has been handed some, and a seek
1219 may land seconds ahead of what the source has produced. So the feed is
1220 waited out first and the per-member budget below measures the binary
1221 alone; giving up on the source here would only restart the session into
1222 the very same wait.
1223 """
1224 await self._wait_feed_settled()
1225 members = [(p, p.stream) for p in self.sync_clients if p.stream]
1226 results = await asyncio.gather(*[stream.wait_audio_present() for _, stream in members])
1227 if all(results):
1228 return
1229 # Name the members that never reported audio: they are what has to be
1230 # looked at, and the group start below is abandoned for all of them.
1231 silent = [
1232 player.display_name
1233 for (player, _), present in zip(members, results, strict=True)
1234 if not present
1235 ]
1236 raise PlayerCommandFailed(f"audio feed was not confirmed by {', '.join(silent)}")
1237
1238 async def _wait_feed_settled(self) -> None:
1239 """Wait for the source to hand over its first audio, or to end without any."""
1240 task = self._audio_source_task
1241 if task is None or self._feed_settled.is_set():
1242 return
1243 settled = asyncio.create_task(self._feed_settled.wait())
1244 try:
1245 # The streamer settles the event itself, but watching the task too
1246 # means a feed that never even starts cannot hold this open: a task
1247 # cancelled before its first step never runs the finally that
1248 # settles the event. The timeout is the backstop for a producer that
1249 # neither delivers nor gives up: this runs under the player lock,
1250 # where every route that could stop the session waits behind it.
1251 await asyncio.wait(
1252 {settled, task},
1253 timeout=AIRPLAY_FEED_START_TIMEOUT,
1254 return_when=asyncio.FIRST_COMPLETED,
1255 )
1256 finally:
1257 settled.cancel()
1258
1259 async def _wait_members_clock_ready(self) -> int:
1260 """
1261 Return the latest receiver-clock readiness any member reported.
1262
1263 Members that report nothing contribute no instant to the maximum; the
1264 caller anchors those on its lead alone.
1265
1266 :return: Unix epoch ms of the latest projection any member reported, or 0
1267 when there is nothing to wait for â a receiver on NTP timing, or one
1268 that never answered. The caller then anchors on its lead alone.
1269 """
1270 # A solo start waits for the projection too: a receiver that has not
1271 # seated its clock renders silence at an anchor it cannot honor, and
1272 # enough of them need that time (WiiM, Edifier) that no start may assume
1273 # otherwise. It costs a warm clock nothing - the binary reports it ready
1274 # with a past instant right after connect - and a cold one anchors just
1275 # past its own projection instead of being corrected there by the binary
1276 # afterwards: the same instant, planned rather than repaired, with the
1277 # readiness lead's slack on top.
1278 results = await asyncio.gather(
1279 *[
1280 p.stream.wait_clock_ready(timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000)
1281 for p in self.sync_clients
1282 if p.stream
1283 ]
1284 )
1285 ready_at_unix_ms = max(
1286 (at for readiness, at in results if readiness is ClockReadiness.PROJECTED), default=0
1287 )
1288 # A group start does not drop a member that stalled - the rest of the
1289 # group would still be started, and the member is already warned about
1290 # by name, with the ports to check, where the binary reported it. Say
1291 # which outcomes were seen so the anchor decision is readable.
1292 unprojected = [
1293 readiness for readiness, _ in results if readiness is not ClockReadiness.PROJECTED
1294 ]
1295 if unprojected:
1296 self.prov.logger.debug(
1297 "AirPlay start: %d of %d member(s) reported no receiver clock projection (%s); "
1298 "anchoring those on the start lead alone",
1299 len(unprojected),
1300 len(results),
1301 ", ".join(sorted({readiness.value for readiness in unprojected})),
1302 )
1303 return ready_at_unix_ms
1304
1305 async def _start_members(self, position_ms: int, start_unix_ms: int) -> None:
1306 """
1307 Anchor every member's playback at one shared audible instant.
1308
1309 The binaries verify the instant: each ack carries the TRUE scheduled
1310 instant (an infeasible one is corrected forward, never silently
1311 misplaced). When any member was corrected, every member is re-STARTed
1312 at the largest reported instant so the group converges on one shared
1313 instant; the recorded session anchor is always the verified truth.
1314 A solo member is never re-STARTed: its corrected instant is simply
1315 adopted as the anchor, since there is no partner to converge with.
1316
1317 :param position_ms: Media position mapped to the first sample of the anchor.
1318 :param start_unix_ms: Shared audible-start instant in unix epoch ms.
1319 """
1320 target_ms = start_unix_ms
1321 corrected_ms = start_unix_ms
1322 for _ in range(4):
1323 member_tasks: list[tuple[int, asyncio.Task[int]]] = []
1324 async with asyncio.TaskGroup() as task_group:
1325 for player in self.sync_clients:
1326 stream = player.stream
1327 if stream is None:
1328 continue
1329 sync_adjust = player.config.get_value(CONF_SYNC_ADJUST, 0)
1330 adjust_ms = sync_adjust if isinstance(sync_adjust, int) else 0
1331 member_tasks.append(
1332 (
1333 adjust_ms,
1334 task_group.create_task(
1335 stream.start(target_ms + adjust_ms, position_ms)
1336 ),
1337 )
1338 )
1339 corrected_ms = target_ms
1340 for adjust_ms, task in member_tasks:
1341 corrected_ms = max(corrected_ms, task.result() - adjust_ms)
1342 if corrected_ms <= target_ms + 2:
1343 break
1344 if len(member_tasks) == 1:
1345 # A lone member has no partner to converge with and its binary
1346 # already scheduled the corrected instant exactly, so adopt that
1347 # as the anchor. Re-STARTing it only does damage: the command
1348 # re-bases reported position on the raw position_ms (start()
1349 # writes that base unconditionally), throwing away the
1350 # correction the anchor already folded into it, and if the
1351 # second ack times out the session records an instant the
1352 # binary never played on.
1353 target_ms = corrected_ms
1354 break
1355 self.prov.logger.warning(
1356 "AirPlay group start corrected: a member could not honor %d, "
1357 "re-anchoring all members at %d (+%d ms)",
1358 target_ms,
1359 corrected_ms,
1360 corrected_ms - target_ms,
1361 )
1362 # The corrected instant already carries the binary's own
1363 # command-latency slack; the extra margin covers fanning the
1364 # retry out to every member so the next round lands.
1365 target_ms = corrected_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS
1366 else:
1367 # No round landed, so the members sit on the instants they last
1368 # reported, not on the retry that was about to be commanded. The
1369 # anchor has to record where they actually are, or every later
1370 # joiner maps against a timeline the group never played.
1371 target_ms = corrected_ms
1372 self.prov.logger.error(
1373 "AirPlay group start did not converge after 4 rounds "
1374 "(members last reported %d) - they may be audibly out of sync; "
1375 "please report this with a debug log",
1376 target_ms,
1377 )
1378 self.start_unix_ms = target_ms
1379 self.start_time = target_ms / 1000
1380 # the only place a session (re)gains a live timeline, so any park ends here
1381 self.parked = False
1382
1383 async def _flush_member(self, player: AirPlayPlayer) -> bool:
1384 """Flush one member's live stream in place and report the binary's ack."""
1385 stream = player.stream
1386 if stream is None:
1387 return False
1388 return await stream.flush()
1389
1390 def _reset_member_shifts(self) -> None:
1391 """Zero every member's accumulated starvation shift for a fresh anchor."""
1392 for player in self.sync_clients:
1393 if player.stream is not None:
1394 player.stream.reset_reanchor_shift()
1395
1396 async def _start_player_ffmpeg(self, player: AirPlayPlayer, media: PlayerMedia) -> None:
1397 """
1398 Start the per-seek ffmpeg feeding a member's persistent cli stdin.
1399
1400 Retires any ffmpeg still tracked for the player and wires a fresh one to
1401 the same cli stdin fd. Killing ffmpeg never closes cli stdin (MA holds
1402 the write end via its process transport), so the binary's stdin reader
1403 survives a warm seek.
1404
1405 :param player: The member whose ffmpeg is (re)started.
1406 :param media: Media whose queue/session identify the output plan.
1407 """
1408 if ffmpeg := self._player_ffmpeg.pop(player.player_id, None):
1409 await ffmpeg.close()
1410 stream = player.stream
1411 assert stream
1412 handoff_format = stream.pcm_format
1413 output_plan = self.mass.streams.audio.get_player_output_plan(
1414 player.player_id,
1415 input_format=self.pcm_format,
1416 output_format=get_final_output_format(handoff_format),
1417 handoff_format=handoff_format,
1418 queue_id=media.source_id,
1419 session_id=get_media_session_id(media),
1420 )
1421 cli_proc = stream._cli_proc
1422 assert cli_proc
1423 assert cli_proc.proc
1424 assert cli_proc.proc.stdin
1425 stdin_transport = cli_proc.proc.stdin.transport
1426 audio_output: str | int = stdin_transport.get_extra_info("pipe").fileno()
1427 ffmpeg = FFMpeg(
1428 audio_input="-",
1429 input_format=self.pcm_format,
1430 output_format=handoff_format,
1431 filter_params=output_plan.filter_params,
1432 audio_output=audio_output,
1433 )
1434 await ffmpeg.start()
1435 self._player_ffmpeg[player.player_id] = ffmpeg
1436
1437
1438def _first_music_assistant_error(err: BaseException) -> MusicAssistantError | None:
1439 """
1440 Return the first MusicAssistantError inside an error or (nested) exception group.
1441
1442 :param err: The error to inspect.
1443 """
1444 if isinstance(err, MusicAssistantError):
1445 return err
1446 if isinstance(err, BaseExceptionGroup):
1447 for nested in err.exceptions:
1448 if (found := _first_music_assistant_error(nested)) is not None:
1449 return found
1450 return None
1451