music-assistant-server

1"""
2AirPlay audio streaming using the cliairplay binary.
3
4Handles both RAOP (AirPlay 1) and AirPlay 2 protocols through a single
5unified binary. Audio is fed via stdin, commands via a named pipe,
6status is reported on stderr in normalized [STATUS] format.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import logging
13import re
14import time
15from contextlib import suppress
16from dataclasses import dataclass
17from http import HTTPStatus
18from typing import TYPE_CHECKING, Any, Final, cast
19from uuid import uuid4
20
21from music_assistant_models.enums import PlaybackState
22from music_assistant_models.errors import PlayerCommandFailed
23
24from music_assistant.constants import VERBOSE_LOG_LEVEL
25from music_assistant.helpers.images import get_image_thumb_path
26from music_assistant.helpers.named_pipe import AsyncNamedPipeWriter
27from music_assistant.helpers.process import AsyncProcess
28from music_assistant.providers.airplay.constants import (
29    AIRPLAY_ARTWORK_RENDER_TIMEOUT,
30    AIRPLAY_ARTWORK_SIZE,
31    AIRPLAY_CONTENT_CUT_TOLERANCE_MS,
32    AIRPLAY_JOIN_START_ACK_TIMEOUT_MS,
33    AIRPLAY_PCM_FORMAT,
34    AIRPLAY_START_ACK_TIMEOUT_MS,
35    CLI_PROBLEM_MARKERS,
36    CONF_AIRPLAY_CREDENTIALS,
37    CONF_BUFFER_DEPTH,
38    CONF_ENCRYPTION,
39    CONF_PASSWORD,
40    CONF_RAOP_CREDENTIALS,
41    AirPlayRemoteCommand,
42    ClockReadiness,
43    StreamingProtocol,
44)
45from music_assistant.providers.airplay.helpers import (
46    default_buffer_depth,
47    generate_active_remote_id,
48    get_cli_binary,
49    get_decoded_property,
50    serialize_txt_records,
51)
52
53if TYPE_CHECKING:
54    from collections.abc import Mapping
55
56    from music_assistant_models.media_items import AudioFormat
57    from music_assistant_models.player import PlayerMedia
58
59    from music_assistant.providers.airplay.player import AirPlayPlayer
60    from music_assistant.providers.airplay.provider import AirPlayProvider
61    from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
62
63# Slugs of the machine-readable failure line the binary emits:
64#   [STATUS] error code=<slug> http=<int> detail="<short text>"
65# The auth/connect slugs are terminal - the binary gives up and exits. The
66# command slugs are not: the connection survives a rejected transport command
67# and only the pending ack is answered with a failure.
68CLI_ERROR_AUTH_REQUIRED: Final[str] = "auth_required"
69CLI_ERROR_AUTH_FAILED: Final[str] = "auth_failed"
70CLI_ERROR_START_FAILED: Final[str] = "start_failed"
71CLI_ERROR_FLUSH_FAILED: Final[str] = "flush_failed"
72
73_CLI_ERROR_CODE_RE = re.compile(r"\bcode=(\S+)")
74_CLI_ERROR_HTTP_RE = re.compile(r"\bhttp=(\d+)")
75_CLI_ERROR_DETAIL_RE = re.compile(r'\bdetail="([^"]*)"')
76
77# Seconds to wait for the binary's command pipe reader. It attaches right after
78# the connection is reported, so this only bridges the reader thread starting up.
79_COMMAND_PIPE_READER_TIMEOUT: Final[float] = 2.0
80
81
82@dataclass
83class CliError:
84    """Structured failure as reported by the cliairplay binary."""
85
86    code: str
87    http_status: int = 0
88    detail: str = ""
89
90
91class AirPlayStream:
92    """AirPlay audio streamer using the unified cliairplay binary."""
93
94    _cli_proc: AsyncProcess | None
95    session: AirPlayStreamSession | None = None
96
97    def __init__(self, player: AirPlayPlayer, pcm_format: AudioFormat | None = None) -> None:
98        """
99        Initialize AirPlay stream.
100
101        :param player: The player to stream to.
102        :param pcm_format: The PCM format fed to the binary's stdin
103            (defaults to 44.1kHz/16-bit).
104        """
105        self.prov = player.provider
106        self.mass = player.provider.mass
107        self.player = player
108        self.pcm_format = pcm_format or AIRPLAY_PCM_FORMAT
109        mac_address = self.player.device_info.mac_address or self.player.player_id
110        self.active_remote_id: str = generate_active_remote_id(mac_address)
111        self._stream_id = uuid4().hex
112        self.prevent_playback: bool = False
113        self._cli_proc: AsyncProcess | None = None
114        self.commands_pipe = AsyncNamedPipeWriter(
115            f"/tmp/{self.player.protocol.value}-{self.player.player_id}-"  # noqa: S108
116            f"{self.active_remote_id}-{self._stream_id}-cmd",
117        )
118        self._stopped = False
119        self._stopping = False
120        self._cleanup_complete = False
121        self._stop_lock = asyncio.Lock()
122        # Both wake a track-change metadata push waiting out its artwork render
123        # budget inside the metadata lock: the first is set permanently the
124        # moment stop() begins, the second while start() is waiting on the lock
125        # (a pending START is time-critical — the anchor lead is a few hundred
126        # ms). Either makes the bounded wait yield so the teardown or START
127        # proceeds and the artwork follows asynchronously.
128        self._teardown_started, self._start_waiting = asyncio.Event(), asyncio.Event()
129        self._connected = asyncio.Event()
130        # Set when the stderr reader ends, i.e. the binary is gone. A connect
131        # wait watches it so a process that died (for example on a rejected
132        # password) fails right away instead of running out its timeout.
133        self._process_ended = asyncio.Event()
134        # Whether the binary reported the end of the stream itself ([STATUS] eof
135        # or its idle timeout) instead of dying. Both leave the process gone and
136        # `running` reading False, so this is what tells the two apart.
137        self.ended_cleanly: bool = False
138        # Structured fatal failure the binary reported before exiting; stays
139        # None when it exited without reporting one.
140        self._connect_error: CliError | None = None
141        # Set when the binary answers an in-place FLUSH, either acknowledging it
142        # ([STATUS] flushed) or reporting that it rejected the command, which
143        # fills _flush_error. Each answer clears the other.
144        self._flushed = asyncio.Event()
145        self._flush_error: CliError | None = None
146        # Set when the binary acks a START ([STATUS] started); carries the
147        # (requested, actual) unix-ms pair. The binary corrects an infeasible
148        # instant FORWARD and always reports the truth, so a mismatch here is
149        # the self-verification signal for the start contract. A reported start
150        # failure also sets it, filling _start_error instead of the ack; each
151        # answer clears the other.
152        self._started = asyncio.Event()
153        self._start_ack: tuple[int, int] | None = None
154        self._start_error: CliError | None = None
155        # Whether the last commanded START was a late-join start: routes the
156        # post-commit correction log level (a corrected join is the routine
157        # landing path, a corrected origin start is a loud signal).
158        self._start_was_join = False
159        # Receiver storm guard state: last-honored monotonic time per remote
160        # transport command, plus a counter of suppressed repeats.
161        self._remote_command_last: dict[str, float] = {}
162        self._remote_commands_suppressed: int = 0
163        # Set when the binary reports the first audio bytes of the current
164        # start cycle arriving on its stdin ([STATUS] audio). Together with
165        # `connected` this makes readiness fully event-driven, so START can
166        # use a short re-anchor lead instead of a guessed setup time.
167        self._audio_present = asyncio.Event()
168        # Set once the binary settled the receiver's clock readiness ([STATUS]
169        # clock_ready): either it projected when the clock becomes usable, or it
170        # reported that there is nothing to wait for. The projected instant
171        # (unix ms) stays 0 in the latter case, and the readiness below says
172        # which of the reasons it was.
173        self._clock_ready = asyncio.Event()
174        self._clock_ready_at_unix_ms: int = 0
175        self._clock_readiness: ClockReadiness = ClockReadiness.UNREPORTED
176        self._metadata_text_checksum = ""
177        # Artwork identity (the source image url) whose rendered bytes were
178        # last delivered to the binary. Settles on the first successful
179        # delivery, independent of the metadata generation: media updates
180        # around a track transition keep bumping the generation, and a settle
181        # tied to it would re-render and re-send the same art on every update
182        # until the churn stops.
183        self._metadata_artwork_checksum = ""
184        self._pending_metadata_checksum = ""
185        self._metadata_generation = 0
186        self._metadata_lock = asyncio.Lock()
187        self._artwork_render_generations: set[int] = set()
188        self._last_progress_sent: int | None = None
189        # Media position (seconds) mapped to the first sample of the current
190        # START anchor. The binary reports "playing elapsed_ms" relative to that
191        # anchor (resetting to ~0 at each START), so elapsed is this base plus
192        # the reported delta.
193        self._start_position: float = 0.0
194        # Content cut (ms) a post-commit anchor correction asked for and that is
195        # already folded into the base above, until the binary reports what it
196        # actually managed to take. 0 when no cut is outstanding.
197        self._pending_content_cut_ms: int = 0
198        self._stdout_reader_task: asyncio.Task[None] | None = None
199        # Device latency info reported by the binary after connect (0 = unreported)
200        self.latency_lead_ms: int = 0
201        self.device_min_frames: int = 0
202        self.device_max_frames: int = 0
203        # Minimum lead (ms) a warm commanded START needs for exact placement.
204        # Nonzero on the Apple splice timeline, where the receiver's queued
205        # audio plays out before the new content can begin; a warm group
206        # anchor must sit beyond the largest member value. 0 = no constraint.
207        self.warm_lead_ms: int = 0
208        # Audible instant (unix ms) of the delivery head frozen by the latest
209        # warm flush, from the flushed ack (0 = none/no constraint). The warm
210        # START anchor must land beyond it for the splice skip to engage.
211        self.flushed_head_unix_ms: int = 0
212        # Route the binary resolved for this stream (empty until reported),
213        # e.g. "AirPlay 2 (native, PTP)" or "RAOP"
214        self.active_route: str = ""
215        # Cumulative playout shift (seconds) this process reported after PCM
216        # starvation re-anchors (AP2 only). The stream session adds the
217        # reference member's shift so a late joiner anchors to the group's real
218        # timeline. Reset per process (and on every re-anchoring START/resume);
219        # a new cliairplay re-anchors from scratch.
220        self.cumulative_shift_seconds: float = 0.0
221        # Set once a stalled receiver clock has been reported, so the warning
222        # stays a single support signal instead of repeating with every
223        # clock_ready update of this stream session.
224        self._clock_stall_warned: bool = False
225
226    @property
227    def running(self) -> bool:
228        """Return boolean if this stream is running."""
229        return (
230            not self._stopped
231            and not self._stopping
232            and self._cli_proc is not None
233            and not self._cli_proc.closed
234        )
235
236    @property
237    def connected(self) -> bool:
238        """Return boolean if the device connection has been established."""
239        return self._connected.is_set()
240
241    async def connect(
242        self,
243        use_shared_ptp: bool | None = None,
244    ) -> None:
245        """
246        Spawn cliairplay and connect to the receiver.
247
248        Establishes the process, command pipe and device connection that persist
249        for the whole stream lifetime. Playback itself is anchored separately with
250        :meth:`start` once audio is being fed on the persistent stdin.
251
252        :param use_shared_ptp: Session-wide decision on whether native AirPlay 2
253            members attach to the shared PTP clock daemon. The stream session
254            passes the same value to every member so a group never mixes PTP and
255            NTP timing. None lets the stream decide from the daemon's readiness.
256        """
257        self._check_password_preflight()
258        # A fresh cliairplay process re-anchors from scratch, so drop any shift
259        # carried on this stream object.
260        self.reset_reanchor_shift()
261        args = await self._build_cli_args(use_shared_ptp)
262        self.player.logger.debug("Starting cliairplay for player %s", self.player.player_id)
263        self._cli_proc = AsyncProcess(args, stdin=True, stdout=True, stderr=True, name="cliairplay")
264        try:
265            await self.commands_pipe.create()
266            await self._cli_proc.start()
267            self._cli_proc.attach_stderr_reader(self.mass.create_task(self._stderr_reader()))
268            self._stdout_reader_task = self.mass.create_task(self._stdout_reader())
269        except BaseException:
270            try:
271                await self._cleanup_failed_start()
272            except Exception as err:
273                self.player.logger.warning("Failed to clean up cliairplay startup: %s", err)
274            raise
275
276    async def wait_for_connection(self) -> None:
277        """
278        Wait for device connection to be established.
279
280        Also waits for the binary's command pipe to open, so the first commands
281        are not dropped.
282
283        :raises PlayerCommandFailed: If the binary reported that the device needs
284            a password, rejected the configured one, or never opened the command
285            pipe that carries playback commands.
286        :raises TimeoutError: If the connection was not established for any other
287            reason (including an unreported one).
288        """
289        if not self._cli_proc:
290            raise RuntimeError("cliairplay process is not running")
291        await self._await_connected()
292        # The binary attaches to its command pipe only once it is connected, so
293        # the first command waits for that reader instead of being dropped. A
294        # pipe that never opens leaves the stream unable to be anchored at all,
295        # so it fails the connection rather than letting the caller command a
296        # START nothing can receive. A stream torn down while connecting takes
297        # its pipe along, which is not a fault worth reporting.
298        if not await self.commands_pipe.wait_for_reader(_COMMAND_PIPE_READER_TIMEOUT):
299            if self.running:
300                raise PlayerCommandFailed(
301                    f"cliairplay did not open its command pipe for "
302                    f"{self.player.display_name}; playback commands cannot be delivered"
303                )
304        # Nothing has reached this binary yet, so clear the delivery state to
305        # make sure the pushes below are really sent.
306        async with self._metadata_lock:
307            self._metadata_text_checksum = ""
308            self._metadata_artwork_checksum = ""
309            self._pending_metadata_checksum = ""
310            self._metadata_generation += 1
311        # Push track metadata before START. Some receivers (notably Sonos) hold
312        # back audio rendering until they receive track metadata; deferring it
313        # can keep them silent past the commanded start.
314        await self._send_current_metadata(send_artwork=False)
315        # Send the mute-aware volume right away — audio can start within a
316        # second now that metadata goes out immediately — and repeat it after
317        # 2 seconds because some players ignore the first volume command
318        # (https://github.com/music-assistant/support/issues/3330). The repeat reads
319        # the level when it fires, so it never replays a value that changed since.
320        await self._send_current_volume()
321        self.mass.call_later(2, self._send_current_volume)
322        # settle artwork and the position on top of the identity push above
323        self.player._on_player_media_updated()
324
325    async def stop(self, force: bool = False) -> None:
326        """
327        Stop playback and cleanup.
328
329        :param force: If True, immediately kill the process without graceful shutdown.
330        """
331        async with self._stop_lock:
332            if self._cleanup_complete:
333                return
334            self._stopping = True
335            self._teardown_started.set()
336            async with self._metadata_lock:
337                self._metadata_generation += 1
338                try:
339                    await self._write_cli_command("ACTION=STOP")
340                finally:
341                    self._stopped = True
342                    try:
343                        await self.commands_pipe.remove()
344                    finally:
345                        # stop the stdout reader first so process close can drain the pipe
346                        stdout_reader_task = self._stdout_reader_task
347                        if stdout_reader_task and not stdout_reader_task.done():
348                            stdout_reader_task.cancel()
349                            with suppress(asyncio.CancelledError):
350                                await stdout_reader_task
351                        try:
352                            if force:
353                                if self._cli_proc and not self._cli_proc.closed:
354                                    await self._cli_proc.kill()
355                            else:
356                                if self._cli_proc:
357                                    await self._cli_proc.write_eof()
358                                if self._cli_proc and not self._cli_proc.closed:
359                                    await self._cli_proc.close()
360                        finally:
361                            self.player.set_state_from_stream(
362                                state=PlaybackState.IDLE,
363                                elapsed_time=0,
364                                stream=self,
365                            )
366                            self._cleanup_complete = True
367
368    async def write_audio(self, data: bytes) -> None:
369        """
370        Write raw audio data to the CLI process stdin.
371
372        :param data: Raw audio bytes to send to the streaming process.
373        """
374        if self._stopped or self._stopping or not self._cli_proc or self._cli_proc.closed:
375            return
376        await self._cli_proc.write(data)
377
378    async def write_audio_eof(self) -> None:
379        """Signal end-of-stream to the CLI process stdin."""
380        if self._stopped or self._stopping or not self._cli_proc or self._cli_proc.closed:
381            return
382        await self._cli_proc.write_eof()
383
384    async def send_cli_command(self, command: str) -> bool:
385        """
386        Send an interactive command to the running CLI binary.
387
388        :param command: Command to send.
389        :return: True when the complete command is delivered, False when the
390            command is ignored, the CLI is unavailable, or the write is dropped.
391        """
392        if self._stopped or self._stopping:
393            return False
394        return await self._write_cli_command(command)
395
396    async def flush(self, timeout: float = 2.0) -> bool:
397        """
398        Flush the live stream in place and wait for the binary's acknowledgement.
399
400        Sends ``ACTION=FLUSH`` — the binary stops sending audio, flushes the
401        receiver, discards its input ring and drains stdin, then reports
402        ``[STATUS] flushed`` while keeping the connection and stdin reader alive.
403        The caller must have stopped feeding old audio before calling this so the
404        drain removes exactly the pre-flush bytes.
405
406        :param timeout: Seconds to wait for the flushed acknowledgement.
407        :return: True once the flush is acknowledged; False on a delivery
408            failure, a flush the binary reports it rejected, or a timeout, so
409            the caller can fall back to a cold restart.
410        """
411        if not self.running or not self.connected:
412            return False
413        self._arm_flush_answer()
414        # The flush drain re-arms the binary's one-shot audio signal; the next
415        # [STATUS] audio belongs to the new track.
416        self._audio_present.clear()
417        if not await self._write_cli_command("ACTION=FLUSH"):
418            return False
419        try:
420            await asyncio.wait_for(self._flushed.wait(), timeout)
421        except TimeoutError:
422            return False
423        if (error := self._flush_error) is not None:
424            self.player.logger.warning(
425                "cliairplay rejected the flush for %s (%s); falling back to a cold restart",
426                self.player.display_name,
427                error.detail or error.code,
428            )
429            return False
430        return True
431
432    async def wait_audio_present(self, timeout: float = 5.0) -> bool:
433        """
434        Wait until the binary reports the current start cycle's audio arriving.
435
436        The binary emits a one-shot ``[STATUS] audio`` when the first bytes of
437        a start cycle land on its stdin (re-armed by each flush). Waiting for
438        it before commanding START removes source/transcoder spin-up from the
439        start lead.
440
441        :param timeout: Seconds to wait for the signal.
442        :return: True once audio is flowing; False on timeout.
443        """
444        try:
445            await asyncio.wait_for(self._audio_present.wait(), timeout)
446        except TimeoutError:
447            return False
448        return True
449
450    async def wait_clock_ready(self, timeout: float = 2.5) -> tuple[ClockReadiness, int]:
451        """
452        Wait for the binary to project when the receiver's clock becomes usable.
453
454        A receiver starts probing its clock as soon as it is connected, so the
455        projection is available well before any anchor is announced and resolves
456        from the receiver's first probe rather than from its full servo lock.
457
458        :param timeout: Seconds to wait for the projection.
459        :return: How the readiness resolved, and the projected instant (unix ms)
460            at which the receiver's clock is usable — possibly already in the
461            past when it is locked. Only :attr:`ClockReadiness.PROJECTED` carries
462            an instant; every other outcome pairs with 0 and leaves the caller
463            anchoring on its lead alone, but for reasons that differ enough to
464            act on: a stalled receiver will render silence, NTP timing has no
465            clock to wait for, and an unreported one is worth retrying.
466        """
467        try:
468            await asyncio.wait_for(self._clock_ready.wait(), timeout)
469        except TimeoutError:
470            return (ClockReadiness.UNREPORTED, 0)
471        return (self._clock_readiness, self._clock_ready_at_unix_ms)
472
473    async def start(
474        self, start_unix_ms: int = 0, position_ms: int = 0, *, join: bool = False
475    ) -> int:
476        """
477        Anchor playback so the first pending stdin sample is audible at an instant.
478
479        The first call begins playback (connection already established); a call
480        after :meth:`flush` re-bases the frozen anchor and resumes from the ring.
481
482        :param start_unix_ms: Unix-epoch milliseconds at which the first pending
483            stdin sample must be audible. 0 means as soon as possible (the binary
484            clamps to its minimum lead).
485        :param position_ms: Media position mapped to that first sample, used as
486            the base for elapsed reporting.
487        :param join: This start must land on an already-live group timeline (a
488            late joiner): the binary then enforces receiver clock readiness and
489            holds its ack until that resolves, so the returned instant is the one
490            the caller must map the joiner's content onto. Group/solo origin
491            starts leave it False.
492        :return: The true scheduled audible instant (unix ms) from the binary's
493            started ack — the commanded instant when it was feasible, the
494            corrected-forward one otherwise.
495        :raises PlayerCommandFailed: If the START command cannot be delivered,
496            the binary reports that it scheduled no instant, or it never
497            acknowledged the start.
498        """
499        if not self.running or not self.connected:
500            raise RuntimeError("Cannot start playback without a connected cliairplay process")
501        # A START re-anchors playout from scratch — the binary zeroes its own
502        # re-anchor total on start/resume — so drop any shift accumulated against
503        # the previous anchor (this also covers the warm-seek FLUSH->refill->START
504        # path) to keep the server and binary baselines aligned.
505        self.reset_reanchor_shift()
506        self._start_position = position_ms / 1000
507        # This base is absolute, so a cut still outstanding against the previous
508        # anchor is no longer part of it and must not be reconciled into it.
509        self._pending_content_cut_ms = 0
510        # Stamp the player's elapsed onto the new anchor's base right away: until
511        # the binary's first status arrives, interpolation would otherwise keep
512        # extending the previous anchor's clock, briefly mapping a bogus position.
513        self.player.set_state_from_stream(elapsed_time=self._start_position, stream=self)
514        self._arm_start_answer()
515        self._start_was_join = join
516        start_cmd = f"START_UNIX_MS={start_unix_ms}\nACTION=START"
517        if join:
518            start_cmd = f"START_UNIX_MS={start_unix_ms}\nSTART_JOIN=1\nACTION=START"
519        self._start_waiting.set()
520        try:
521            await self._metadata_lock.acquire()
522        finally:
523            self._start_waiting.clear()
524        try:
525            if not await self._write_cli_command(start_cmd):
526                # Surfacing the dropped delivery lets the session fall back to a
527                # cold restart instead of waiting on an anchor that never happens.
528                raise PlayerCommandFailed(
529                    f"Could not deliver START to AirPlay player {self.player.player_id}"
530                )
531            # Supersede an in-flight pre-transition artwork render; the task
532            # below then sends only what actually changed (a track change's
533            # text/artwork). Unchanged title/artwork stay deduped — re-pushing
534            # identical metadata around every anchor makes an Apple TV
535            # re-render its Now Playing popup on each seek, and a re-anchor
536            # cannot lose artwork anyway now that the binary carries the
537            # artwork bytes in every now-playing push. The progress correction
538            # is deliberately NOT sent here: every now-playing push visibly
539            # refreshes the Apple TV screen, so the single settled correction
540            # from the post-anchor media-updated nudge (+1s) does the job with
541            # one refresh instead of two.
542            self._metadata_generation += 1
543        finally:
544            self._metadata_lock.release()
545        self.mass.create_task(
546            self._send_current_metadata_without_progress,
547            task_id=f"airplay_metadata_after_start_{self._stream_id}",
548            abort_existing=True,
549        )
550        # The binary always acks with the TRUE scheduled instant (correcting an
551        # infeasible one forward), so the caller can verify the contract and
552        # re-align a group. A join's ack is held back until the receiver clock
553        # verification resolves and therefore gets a much wider window than a
554        # plain start, which acks within the command round-trip. A reported
555        # failure answers the wait immediately.
556        ack_timeout = (
557            AIRPLAY_JOIN_START_ACK_TIMEOUT_MS if join else AIRPLAY_START_ACK_TIMEOUT_MS
558        ) / 1000
559        try:
560            await asyncio.wait_for(self._started.wait(), ack_timeout)
561        except TimeoutError as err:
562            # Nothing confirmed the instant, so nothing may be mapped onto it:
563            # an anchor the caller records but the receiver never played is a
564            # timeline every later joiner then aligns itself against.
565            raise PlayerCommandFailed(
566                f"AirPlay player {self.player.display_name} did not acknowledge its "
567                f"start within {ack_timeout:.1f}s (commanded instant {start_unix_ms})"
568            ) from err
569        if (error := self._start_error) is not None:
570            # Nothing was anchored, so there is no instant to map content onto.
571            # Failing here costs the caller the round-trip instead of the whole
572            # ack timeout, and tells it apart from an unacknowledged start.
573            raise PlayerCommandFailed(
574                f"cliairplay could not start playback on {self.player.display_name}"
575                + (f": {error.detail}" if error.detail else "")
576            )
577        # A malformed ack still answered the START, so the commanded instant is
578        # what the binary applied (see the parse fallback in _handle_status_line).
579        return self._start_ack[1] if self._start_ack else start_unix_ms
580
581    def rebase_position(self, position_ms: int) -> None:
582        """
583        Re-map reported progress onto a start instant that moved after the command.
584
585        A join's START is acked with the instant the receiver can actually seat,
586        which may be later than the commanded one. The caller maps its content
587        onto that instant and reports the position that lands there.
588
589        :param position_ms: Media position of the first sample the binary
590            renders at the acked instant.
591        """
592        self._start_position = position_ms / 1000
593        self._pending_content_cut_ms = 0
594        self.player.set_state_from_stream(elapsed_time=self._start_position, stream=self)
595
596    def reset_reanchor_shift(self) -> None:
597        """Clear the accumulated re-anchor shift."""
598        self.cumulative_shift_seconds = 0.0
599
600    async def send_metadata(  # noqa: PLR0915
601        self,
602        progress: int | None,
603        metadata: PlayerMedia | None,
604        send_artwork: bool = True,
605    ) -> None:
606        """
607        Send metadata to player.
608
609        :param progress: Current playback position in seconds.
610        :param metadata: Media metadata to send.
611        :param send_artwork: Whether artwork should be rendered and sent.
612        """
613        metadata_checksum: str | None = None
614        text_checksum: str | None = None
615        artwork_checksum = ""
616        duration = 0
617        title = ""
618        artist = ""
619        album = ""
620        item_id = ""
621        if metadata:
622            duration = self._full_media_duration(metadata)
623            title = metadata.title or ""
624            artist = metadata.artist or ""
625            album = metadata.album or ""
626            item_id = metadata.queue_item_id or ""
627            # The identity deliberately excludes duration and image url: a
628            # value that shifts per seek or per media-update would re-send the
629            # full metadata each time — which makes an Apple TV re-render its
630            # Now Playing popup.
631            text_checksum = f"{item_id}|{title}|{artist}|{album}"
632            artwork_checksum = metadata.image_url or ""
633            metadata_checksum = f"{text_checksum}|{metadata.image_url}"
634
635        artwork_url: str | None = None
636        artwork_render: asyncio.Task[str | None] | None = None
637        metadata_generation = 0
638        async with self._metadata_lock:
639            if self._stopped or self._stopping:
640                return
641            if metadata_checksum is not None:
642                if metadata_checksum != self._pending_metadata_checksum:
643                    self._pending_metadata_checksum = metadata_checksum
644                    self._metadata_generation += 1
645                metadata_generation = self._metadata_generation
646            needs_artwork = artwork_checksum != self._metadata_artwork_checksum
647            if (
648                metadata
649                and metadata_checksum is not None
650                and text_checksum is not None
651                and (needs_artwork or text_checksum != self._metadata_text_checksum)
652            ):
653                if text_checksum != self._metadata_text_checksum:
654                    artwork_file: str | None = None
655                    if (
656                        send_artwork
657                        and metadata.image_url
658                        and needs_artwork
659                        and metadata_generation not in self._artwork_render_generations
660                    ):
661                        # Budgeted pre-render so metadata and artwork ride ONE
662                        # SENDMETA push: back-to-back now-playing rewrites (a
663                        # bare replace followed by the artwork) intermittently
664                        # wedge the Apple TV now-playing screen. A render that
665                        # misses the budget keeps running and delivers through
666                        # the ARTWORK command instead.
667                        self._artwork_render_generations.add(metadata_generation)
668                        artwork_url = metadata.image_url
669                        artwork_file, artwork_render = await self._render_artwork_bounded(
670                            artwork_url, metadata_generation
671                        )
672                    # ITEMID gives the binary a stable per-track identity, so
673                    # a later tag refinement for the same queue item (library
674                    # enrichment can settle after playback starts) updates the
675                    # receiver's now-playing item in place instead of
676                    # presenting as a new track.
677                    cmd = f"TITLE={title}\nARTIST={artist}\nALBUM={album}\n"
678                    cmd += f"DURATION={duration}\nITEMID={item_id}\n"
679                    if artwork_file:
680                        cmd += f"ARTWORKFILE={artwork_file}\n"
681                    cmd += "ACTION=SENDMETA\n"
682                    if not await self.send_cli_command(cmd):
683                        if artwork_render is not None:
684                            # the identity push never went out, so drop the
685                            # render and let the next update retry from scratch
686                            artwork_render.cancel()
687                            self._artwork_render_generations.discard(metadata_generation)
688                        return
689                    self._metadata_text_checksum = text_checksum
690                    # the push resets a changed track to position zero (a
691                    # same-item refinement carries the current position), so
692                    # the correction below only follows when playback is
693                    # actually elsewhere (mid-track start, tag refinement)
694                    self._last_progress_sent = 0
695                    if artwork_file:
696                        # the bundle delivered the artwork: settle it and stand
697                        # down the ARTWORK follow-up
698                        self._metadata_artwork_checksum = artwork_checksum
699                        self._artwork_render_generations.discard(metadata_generation)
700                        needs_artwork = False
701                        artwork_url = None
702                        artwork_render = None
703                if metadata_generation != self._metadata_generation:
704                    return
705                if (
706                    send_artwork
707                    and metadata.image_url
708                    and needs_artwork
709                    and metadata_generation not in self._artwork_render_generations
710                ):
711                    self._artwork_render_generations.add(metadata_generation)
712                    artwork_url = metadata.image_url
713                elif not metadata.image_url or not needs_artwork:
714                    self._metadata_artwork_checksum = artwork_checksum
715            if progress is not None and (
716                self._last_progress_sent is None or abs(progress - self._last_progress_sent) >= 2
717            ):
718                # duration rides along so the seek-rebased remaining time
719                # reaches the device without re-sending the metadata identity
720                duration_cmd = f"DURATION={duration}\n" if metadata else ""
721                if await self.send_cli_command(f"{duration_cmd}PROGRESS={progress}"):
722                    self._last_progress_sent = progress
723
724        if artwork_url is not None:
725            await self._render_and_send_artwork(artwork_url, metadata_generation, artwork_render)
726
727    def _full_media_duration(self, metadata: PlayerMedia) -> int:
728        """
729        Return the media's full track duration in seconds.
730
731        The queue rewrites ``PlayerMedia.duration`` to the REMAINING time on a
732        seek (the stream-restart convention for players whose position resets
733        to zero). The spliced AirPlay timeline reports absolute positions, so
734        the device needs the real total — and a total that changes on every
735        seek also makes an Apple TV re-lay-out its Now Playing screen each
736        time, which shows as a brief artwork/screen flash.
737
738        :param metadata: Media whose duration is resolved.
739        """
740        if metadata.source_id and metadata.queue_item_id:
741            queue_item = self.mass.player_queues.get_item(
742                metadata.source_id, metadata.queue_item_id
743            )
744            if queue_item:
745                streamdetails = queue_item.streamdetails
746                full_duration = (
747                    streamdetails.duration if streamdetails else None
748                ) or queue_item.duration
749                if full_duration:
750                    return min(int(full_duration), 3600)
751        return min(metadata.duration or 0, 3600)
752
753    async def _render_artwork_bounded(
754        self, artwork_url: str, metadata_generation: int
755    ) -> tuple[str | None, asyncio.Task[str | None]]:
756        """
757        Start the artwork render and wait for it within the bundling budget.
758
759        Called with the metadata lock held. A render that misses the budget is
760        never cancelled: the returned task keeps running so the caller can hand
761        it to :meth:`_render_and_send_artwork` for the ARTWORK delivery. The
762        wait also yields early to a teardown or a pending START.
763
764        :param artwork_url: The cover-art URL to render.
765        :param metadata_generation: Generation the render belongs to.
766        :return: The rendered artwork path (None when the budget was missed or
767            the render failed) and the render task itself.
768        """
769        artwork_render = asyncio.create_task(
770            self._prepare_artwork(artwork_url, metadata_generation)
771        )
772        # Neither a teardown nor a time-critical START must sit out the render
773        # budget behind the metadata lock, so both release this wait early: a
774        # teardown's doomed bundle write is then dropped by send_cli_command,
775        # and a START's bundle goes out without artwork (the render delivers
776        # through the ARTWORK command once it completes).
777        teardown = asyncio.ensure_future(self._teardown_started.wait())
778        start_waiting = asyncio.ensure_future(self._start_waiting.wait())
779        waiters: set[asyncio.Future[Any]] = {artwork_render, teardown, start_waiting}
780        try:
781            await asyncio.wait(
782                waiters,
783                timeout=AIRPLAY_ARTWORK_RENDER_TIMEOUT,
784                return_when=asyncio.FIRST_COMPLETED,
785            )
786        except asyncio.CancelledError:
787            artwork_render.cancel()
788            self._artwork_render_generations.discard(metadata_generation)
789            raise
790        finally:
791            teardown.cancel()
792            start_waiting.cancel()
793        artwork_file = artwork_render.result() if artwork_render.done() else None
794        return artwork_file, artwork_render
795
796    async def _render_and_send_artwork(
797        self,
798        artwork_url: str,
799        metadata_generation: int,
800        render: asyncio.Task[str | None] | None = None,
801    ) -> None:
802        """
803        Render and apply artwork for the current metadata generation.
804
805        :param artwork_url: Source URL for the artwork; settles as the
806            delivered artwork identity once the binary accepts the command.
807        :param metadata_generation: Generation that must still be current before apply.
808        :param render: Render already under way for this generation to deliver,
809            instead of starting a new one.
810        """
811        try:
812            if render is not None:
813                artwork = await render
814            else:
815                artwork = await self._prepare_artwork(artwork_url, metadata_generation)
816        except asyncio.CancelledError:
817            if render is not None:
818                render.cancel()
819            async with self._metadata_lock:
820                self._artwork_render_generations.discard(metadata_generation)
821            raise
822        async with self._metadata_lock:
823            self._artwork_render_generations.discard(metadata_generation)
824            if (
825                artwork
826                and not self._stopped
827                and not self._stopping
828                and metadata_generation == self._metadata_generation
829                and await self.send_cli_command(f"ARTWORK={artwork}")
830            ):
831                self._metadata_artwork_checksum = artwork_url
832
833    async def _build_cli_args(  # noqa: PLR0915
834        self,
835        use_shared_ptp: bool | None = None,
836    ) -> list[str]:
837        """
838        Assemble the cliairplay argument list for this stream.
839
840        :param use_shared_ptp: Whether a native AirPlay 2 stream attaches to the
841            shared PTP clock daemon. The stream session passes an explicit
842            group-wide decision so members never mix PTP and NTP timing; None
843            reads the daemon's current readiness and decides from that.
844        """
845        cli_binary = await get_cli_binary()
846        prov = cast("AirPlayProvider", self.prov)
847        airplay_info = self.player.airplay_discovery_info
848        raop_info = self.player.raop_discovery_info
849        target_protocol = self.player.protocol_override or self.player.protocol
850        if self.player.protocol_override == StreamingProtocol.RAOP:
851            protocol_arg = "raop"
852        elif target_protocol == StreamingProtocol.AIRPLAY2 and not raop_info:
853            # With no RAOP fallback, force AirPlay 2 because featureless AP2-only
854            # receivers cannot be identified by the binary's TXT-bit test.
855            protocol_arg = "airplay2"
856        else:
857            protocol_arg = "auto"
858
859        args: list[str] = [
860            cli_binary,
861            "--protocol",
862            protocol_arg,
863            "--volume",
864            str(self.player.volume_level),
865            "--dacp",
866            prov.dacp_id,
867            "--activeremote",
868            self.active_remote_id,
869            "--cmdpipe",
870            self.commands_pipe.path,
871            "--samplerate",
872            str(self.pcm_format.sample_rate),
873            "--bitdepth",
874            str(self.pcm_format.bit_depth),
875        ]
876
877        # The binary owns the playback lead/buffer (2000 ms default, clamped to
878        # the device-reported window); there is no user override for it.
879
880        # The endpoint must follow the same capability decision as the binary:
881        # legacy RAOP uses _raop, while native and RAOP-compatible AP2 use _airplay.
882        if target_protocol == StreamingProtocol.AIRPLAY2 and airplay_info:
883            args += ["--port", str(airplay_info.port)]
884            args += ["--name", self.player.display_name]
885            args += ["--hostname", str(airplay_info.server)]
886        elif raop_info:
887            args += ["--port", str(raop_info.port)]
888
889        # mDNS properties from the RAOP service (needed by the RAOP-based flows)
890        if raop_info:
891            args += ["--udn", raop_info.name]
892            for prop in ("et", "md", "am", "pk", "pw", "cn"):
893                if prop_value := raop_info.decoded_properties.get(prop):
894                    args += [f"--{prop}", prop_value]
895        if target_protocol == StreamingProtocol.RAOP and self.player.config.get_value(
896            CONF_ENCRYPTION, True
897        ):
898            args += ["--encrypt"]
899
900        # Full _airplay._tcp TXT for the binary's automatic route selection.
901        # Some receivers advertise their AP2 feature bits only on _raop.ft.
902        txt_records = serialize_txt_records(airplay_info) if airplay_info else ""
903        if (
904            airplay_info
905            and not (
906                airplay_info.decoded_properties.get("features")
907                or airplay_info.decoded_properties.get("ft")
908            )
909            and raop_info
910            and (raop_features := raop_info.decoded_properties.get("ft"))
911        ):
912            txt_records = f"{txt_records} ft={raop_features}".strip()
913        if txt_records:
914            args += ["--txt", txt_records]
915
916        # HAP credentials (triggers native AP2 flow when present)
917        if creds := self.player.get_setup_value(CONF_AIRPLAY_CREDENTIALS):
918            creds_str = str(creds)
919            if len(creds_str) == 192:
920                args += ["--auth", creds_str]
921            else:
922                self.player.logger.warning(
923                    "Invalid credentials length: %d (expected 192)", len(creds_str)
924                )
925
926        # Legacy Apple TV RAOP pairing secret
927        if raop_creds := self.player.get_setup_value(CONF_RAOP_CREDENTIALS):
928            # Credentials format is "client_id:auth_secret", the binary expects the secret
929            creds_str = str(raop_creds)
930            auth_secret = creds_str.split(":", 1)[1] if ":" in creds_str else creds_str
931            args += ["--secret", auth_secret]
932
933        # Device password
934        if password := self.player.config.get_value(CONF_PASSWORD):
935            args += ["--password", str(password)]
936
937        # Shared PTP daemon clock (multi-room sync for native AP2 streams). The
938        # decision is made once per session and passed in, so every native AP2
939        # member of a sync group uses the same timing source and cannot drift.
940        # Without a caller-supplied decision (use_shared_ptp is None) the stream
941        # gates on the daemon serving rather than merely running: between spawn and
942        # the daemon publishing its clock there is nothing to attach to, and a
943        # stream that asks anyway silently takes its own timing instead.
944        if target_protocol == StreamingProtocol.AIRPLAY2:
945            # Deeper receiver queue for devices whose pipeline starves at the
946            # stock depth. The stored value wins; Automatic (0) resolves
947            # through the same device-family table that seeds the config
948            # entry default, so selecting it never downgrades a device.
949            depth_ms = cast("int", self.player.config.get_value(CONF_BUFFER_DEPTH) or 0)
950            if not depth_ms:
951                depth_ms = default_buffer_depth(
952                    self.player.device_info.manufacturer or "",
953                    self.player.device_info.model or "",
954                    get_decoded_property(airplay_info, "fv") if airplay_info else None,
955                )
956            if depth_ms:
957                args += ["--latency", str(depth_ms)]
958            shared_ptp = prov.ptp_daemon_ready if use_shared_ptp is None else use_shared_ptp
959            if shared_ptp:
960                args += ["--ptp-shared"]
961
962        # Local interface binding
963        target_ip = str(self.player.device_info.ip_address)
964        if_arg = await self.mass.streams.get_source_ip(target_ip)
965        if if_arg:
966            args += ["--if", if_arg]
967
968        # Address advertised inside the protocol (timing peers) for hosts where the
969        # reachable address differs from the bind address (e.g. containers). The binary
970        # treats this as authoritative for the receiver's clock-source filter and it
971        # outranks its own connection-derived fallback, so only pass an address the user
972        # actually configured: an auto-detected one would silence a receiver whenever it
973        # names an interface the timing packets do not leave from.
974        publish_arg = self.mass.streams.get_publish_ip(target_ip)
975        if publish_arg and publish_arg != if_arg:
976            args += ["--publish-ip", publish_arg]
977
978        # The addressing the stream ends up with is the first thing needed when triaging
979        # a connection or timing issue from a user's log, and it is invisible otherwise:
980        # both flags are dropped silently when no value applies here.
981        self.player.logger.debug(
982            "cliairplay network binding for player %s: if=%s publish_ip=%s",
983            self.player.player_id,
984            if_arg or "<all interfaces>",
985            publish_arg or "<not configured>",
986        )
987
988        # Debug level
989        if self.prov.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
990            args += ["--debug", "10"]
991        elif self.prov.logger.isEnabledFor(logging.DEBUG):
992            args += ["--debug", "5"]
993
994        # Audio is fed continuously on the process stdin; the binary reads it into
995        # a single ring buffer for the whole session lifetime (flushed and
996        # refilled in place on a seek, never reconnected).
997        args.append(self.player.address)
998        return args
999
1000    async def _stdout_reader(self) -> None:
1001        """
1002        Monitor stdout for the running cliairplay process.
1003
1004        The binary reports its resolved route at startup, the effective lead
1005        plus receiver-reported buffering window after connect, the audio formats
1006        the receiver advertises, and the result of MediaRemote now-playing
1007        pushes (Apple devices):
1008          [STATUS] route protocol=<raop|airplay2> flow=<...> timing=<ntp|ptp> buffered=<0|1>
1009          [STATUS] latency lead_ms=<int> device_min_frames=<int> device_max_frames=<int>
1010          [STATUS] capabilities requested=<hex> realtime_formats=<hex> realtime_known=<0|1>
1011            buffered_formats=<hex> buffered_known=<0|1>
1012          [STATUS] mrp path=<command> status=<http status>
1013          [EVENT] remote command=<play|pause|play_pause|next|previous>
1014        """
1015        if not self._cli_proc:
1016            return
1017        buffer = b""
1018        while chunk := await self._cli_proc.read(1024):
1019            buffer += chunk
1020            while b"\n" in buffer:
1021                raw_line, buffer = buffer.split(b"\n", 1)
1022                line = raw_line.decode("utf-8", errors="ignore").strip()
1023                if not line:
1024                    continue
1025                if "[STATUS] route" in line:
1026                    self._parse_route_status(line)
1027                elif "[STATUS] mrp" in line:
1028                    self._parse_mrp_status(line)
1029                elif "[STATUS] latency" in line:
1030                    self._parse_latency_status(line)
1031                elif "[STATUS] capabilities" in line:
1032                    self._parse_capabilities_status(line)
1033                elif line.startswith("[EVENT] remote command="):
1034                    self._parse_remote_event(line)
1035                self.player.logger.log(
1036                    VERBOSE_LOG_LEVEL, "cliairplay for %s: %s", self.player.display_name, line
1037                )
1038
1039    def _parse_remote_event(self, line: str) -> None:
1040        """Dispatch a normalized remote command reported by cliairplay."""
1041        command_value = line.removeprefix("[EVENT] remote command=").strip()
1042        try:
1043            command = AirPlayRemoteCommand(command_value)
1044        except ValueError:
1045            self.player.logger.warning(
1046                "Ignoring unknown cliairplay remote command: %s", command_value
1047            )
1048            return
1049        # Receiver storm guard: MediaRemote re-sends an unfulfilled transport
1050        # command at ~10 Hz (measured on tvOS: a next-track storm at end of
1051        # item drowned the whole server in seeks until playback collapsed).
1052        # No human intends repeats that fast, so only one command per window
1053        # is honored; the suppressed repeats are counted and reported loudly
1054        # as the support signal.
1055        window = (
1056            2.0 if command in (AirPlayRemoteCommand.NEXT, AirPlayRemoteCommand.PREVIOUS) else 0.5
1057        )
1058        now = time.monotonic()
1059        last = self._remote_command_last.get(command_value, 0.0)
1060        if now - last < window:
1061            self._remote_commands_suppressed += 1
1062            suppressed = self._remote_commands_suppressed
1063            if suppressed in (1, 10) or suppressed % 100 == 0:
1064                self.player.logger.warning(
1065                    "Storm guard: suppressed %d repeated remote '%s' "
1066                    "command(s) from %s within %.1fs",
1067                    suppressed,
1068                    command_value,
1069                    self.player.display_name,
1070                    window,
1071                )
1072            return
1073        self._remote_command_last[command_value] = now
1074        self._remote_commands_suppressed = 0
1075        prov = cast("AirPlayProvider", self.prov)
1076        prov.handle_remote_command(self.player, command)
1077
1078    def _parse_mrp_status(self, line: str) -> None:
1079        """
1080        Parse a [STATUS] mrp line and report how the now-playing push landed.
1081
1082        A push the device accepted is routine bookkeeping and stays at debug.
1083        A rejection is not: it is why a now-playing screen stays blank or keeps
1084        the previous track's art, with nothing else about the session looking
1085        wrong, so it is reported.
1086
1087        :param line: The status line, in one of the shapes
1088            ``[STATUS] mrp path=<command|channel> status=<int>`` or
1089            ``[STATUS] mrp artwork=<posted|rejected> ...``.
1090        """
1091        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1092        display_name = self.player.display_name
1093        if artwork := fields.get("artwork"):
1094            # The artwork variants carry no path=, and a rejection reports its
1095            # reason plus a clear_status rather than a status of its own.
1096            if artwork == "rejected":
1097                self.player.logger.warning(
1098                    "%s rejected the now-playing artwork (%s, %s bytes); its screen keeps "
1099                    "whatever art it had",
1100                    display_name,
1101                    fields.get("reason", "no reason given"),
1102                    fields.get("bytes", "?"),
1103                )
1104            else:
1105                self.player.logger.debug(
1106                    "MRP now-playing artwork accepted by %s (%s bytes, HTTP %s)",
1107                    display_name,
1108                    fields.get("bytes", "?"),
1109                    fields.get("status", "?"),
1110                )
1111            return
1112        if fields.get("path") == "channel":
1113            # Not an HTTP status: 0 = the opt-in data channel was attempted and
1114            # did not come up, 1 = established.
1115            self.player.logger.debug(
1116                "MRP data channel for %s: %s",
1117                display_name,
1118                "established" if fields.get("status") == "1" else "not established",
1119            )
1120            return
1121        # Only a 2xx means the device took the push. Nothing else on this
1122        # control channel does - a redirect is as much "not accepted" as a 4xx.
1123        # A status of 0 is the "unreported" reading (the field is missing or
1124        # unusable), which says nothing about how the push landed, so only a
1125        # reported status is judged - warning about a field the binary never
1126        # sent would report a rejection that nothing observed.
1127        status = _status_int(fields, "status")
1128        rejected = bool(status) and not (HTTPStatus.OK <= status < HTTPStatus.MULTIPLE_CHOICES)
1129        self.player.logger.log(
1130            logging.WARNING if rejected else logging.DEBUG,
1131            "MRP now-playing push (%s path) for %s: HTTP %s",
1132            fields.get("path", "?"),
1133            display_name,
1134            fields.get("status", "?"),
1135        )
1136
1137    def _parse_capabilities_status(self, line: str) -> None:
1138        """Parse the [STATUS] capabilities line and refresh the player's audio formats."""
1139        # The binary reports the format tables it read from the receiver's /info,
1140        # which corrects a device that was unreachable when it was discovered.
1141        # Only the native AirPlay 2 flow reads them; the other routes report
1142        # zeroes with the known flags unset. A change applies to the next stream.
1143        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1144        formats = 0
1145        for mask_field, known_field in (
1146            ("realtime_formats", "realtime_known"),
1147            ("buffered_formats", "buffered_known"),
1148        ):
1149            if fields.get(known_field) != "1":
1150                continue
1151            try:
1152                formats |= int(fields[mask_field], 16)
1153            except KeyError, ValueError:
1154                continue
1155        if formats and formats != self.player.advertised_audio_formats:
1156            self.player.logger.debug(
1157                "Audio formats advertised by %s changed to 0x%x",
1158                self.player.display_name,
1159                formats,
1160            )
1161            self.player.advertised_audio_formats = formats
1162
1163    def _parse_route_status(self, line: str) -> None:
1164        """Parse the [STATUS] route line and log which route this stream took."""
1165        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1166        protocol = fields.get("protocol", "")
1167        if protocol == "airplay2":
1168            flow = fields.get("flow", "")
1169            timing = fields.get("timing", "")
1170            details = "buffered" if fields.get("buffered") == "1" else flow
1171            self.active_route = f"AirPlay 2 ({details}, {timing.upper()})"
1172        else:
1173            self.active_route = "RAOP"
1174        self.player.logger.info(
1175            "Streaming to %s via %s", self.player.display_name, self.active_route
1176        )
1177
1178    def _parse_latency_status(self, line: str) -> None:
1179        """Parse and store the [STATUS] latency line reported by the binary."""
1180        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1181        # Read field by field: one unusable value used to abandon the rest of
1182        # the line, leaving whatever came after it stale from the previous
1183        # report while the values before it had already moved. Every field here
1184        # means "unreported" at 0, so a missing or malformed one lands there.
1185        self.latency_lead_ms = _status_int(fields, "lead_ms")
1186        self.device_min_frames = _status_int(fields, "device_min_frames")
1187        self.device_max_frames = _status_int(fields, "device_max_frames")
1188        self.warm_lead_ms = _status_int(fields, "warm_lead_ms")
1189        self.player.logger.debug(
1190            "Device latency for %s: lead=%dms, warm lead=%dms, "
1191            "buffer window=%d-%d frames (0=unreported)",
1192            self.player.display_name,
1193            self.latency_lead_ms,
1194            self.warm_lead_ms,
1195            self.device_min_frames,
1196            self.device_max_frames,
1197        )
1198
1199    async def _stderr_reader(self) -> None:
1200        """
1201        Monitor stderr for the running cliairplay process.
1202
1203        The binary emits normalized [STATUS] messages:
1204          [STATUS] connected
1205          [STATUS] playing elapsed_ms=<ms>
1206          [STATUS] paused
1207          [STATUS] eof
1208          [STATUS] error code=<slug> http=<int> detail="<short text>"
1209            (auth_required/auth_failed/connect_failed are terminal; the
1210             start_failed/flush_failed command slugs answer a pending ack)
1211          [ERROR] <message>
1212        """
1213        player = self.player
1214        logger = player.logger
1215        if not self._cli_proc:
1216            return
1217        async for line in self._cli_proc.iter_stderr():
1218            if self._stopped:
1219                break
1220            if self._handle_status_line(line):
1221                self.ended_cleanly = True
1222                break
1223            # Routine binary output is verbose-only so it never floods a user's log, but
1224            # its own diagnostics (a failed socket bind, a missing receiver clock) are the
1225            # first thing needed when triaging silent playback, so those stay visible.
1226            # Every line names its speaker: the provider logger is shared, so the output
1227            # of several concurrent streams is otherwise impossible to tell apart.
1228            level = (
1229                logging.WARNING
1230                if any(marker in line.lower() for marker in CLI_PROBLEM_MARKERS)
1231                else VERBOSE_LOG_LEVEL
1232            )
1233            logger.log(level, "cliairplay for %s: %s", player.display_name, line)
1234            await asyncio.sleep(0)
1235
1236        logger.debug("cliairplay stderr reader ended for %s", player.display_name)
1237        self._process_ended.set()
1238        if not self._stopped and not self._stopping:
1239            self._stopped = True
1240            try:
1241                if not self.ended_cleanly:
1242                    logger.warning(
1243                        "cliairplay process stopped unexpectedly for %s", player.display_name
1244                    )
1245                    # Candidates for the automatic re-join: the leader this member
1246                    # was synced to (plus its other members, in case leadership
1247                    # transfers while the backoff runs), or - when this was the
1248                    # leader itself - the members that survive it. Captured before
1249                    # the ungroup below mutates the group state (create_task
1250                    # starts eagerly).
1251                    was_leader = bool(player.group_members)
1252                    if player.synced_to:
1253                        rejoin_candidates = [player.synced_to]
1254                        if leader := self.mass.players.get_player(player.synced_to):
1255                            rejoin_candidates += [
1256                                member_id
1257                                for member_id in leader.group_members
1258                                if member_id not in (player.player_id, player.synced_to)
1259                            ]
1260                    else:
1261                        rejoin_candidates = [
1262                            m for m in player.group_members if m != player.player_id
1263                        ]
1264                    # Hand off to the player controller so it drops just this member, or
1265                    # transfers leadership to a healthy member, instead of dissolving the
1266                    # whole group over a single dead transport. A sync leader is left in
1267                    # its current state here on purpose: the controller only transfers
1268                    # leadership while the queue still looks active, and transfer_queue or
1269                    # dissolve sets the final state.
1270                    # One exception: a member that is a STATIC member of an active group
1271                    # player must not go through cmd_ungroup - the controller interprets
1272                    # unjoining a static member as releasing the whole group (HA unjoin
1273                    # semantics), which would silence every room over one dead transport.
1274                    # Its membership is configuration; drop only this member from the
1275                    # leader's live session instead. The set_members call cannot bounce
1276                    # back to the group player: the controller only redirects it when
1277                    # the group advertises SET_MEMBERS, which a static group never does.
1278                    static_member_of = self._static_group_membership(player)
1279                    if static_member_of and player.synced_to:
1280                        self.mass.create_task(
1281                            self.mass.players.cmd_set_members(
1282                                player.synced_to, player_ids_to_remove=[player.player_id]
1283                            )
1284                        )
1285                    else:
1286                        self.mass.create_task(self.mass.players.cmd_ungroup(player.player_id))
1287                    if rejoin_candidates:
1288                        # the group (or its successor) may still be playing:
1289                        # schedule bounded attempts to re-join it
1290                        player.schedule_group_rejoin(rejoin_candidates)
1291                    if was_leader:
1292                        return
1293                player.set_state_from_stream(state=PlaybackState.IDLE, elapsed_time=0, stream=self)
1294            finally:
1295                await self.commands_pipe.remove()
1296
1297    def _static_group_membership(self, player: AirPlayPlayer) -> str | None:
1298        """Return the active group player id the player is a static member of, if any."""
1299        active_group_id = player.state.active_group
1300        if not active_group_id:
1301            return None
1302        group_player = self.mass.players.get_player(active_group_id)
1303        if group_player and player.player_id in group_player.static_group_members:
1304            return active_group_id
1305        return None
1306
1307    def _handle_status_line(self, line: str) -> bool:  # noqa: PLR0915
1308        """Dispatch one cliairplay status line; True ends the stderr loop."""
1309        player = self.player
1310        if "[STATUS] connected" in line:
1311            self._connected.set()
1312            # whatever the device accepted just now is a working password
1313            player.set_password_invalid(False)
1314        elif "[STATUS] playing elapsed_ms=" in line:
1315            try:
1316                millis = int(line.split("elapsed_ms=")[1])
1317            except ValueError, IndexError:
1318                pass
1319            else:
1320                self._update_elapsed(millis / 1000)
1321        elif "[STATUS] paused" in line:
1322            player.set_state_from_stream(state=PlaybackState.PAUSED, stream=self)
1323        elif "[STATUS] started " in line:
1324            try:
1325                fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1326                requested = int(fields.get("requested_unix_ms", 0))
1327                actual = int(fields.get("at_unix_ms", 0))
1328            except ValueError, IndexError:
1329                # Malformed ack: leave _start_ack unset so the caller falls
1330                # back to trusting the commanded instant, without waiting out
1331                # the ack timeout.
1332                pass
1333            else:
1334                # A line missing at_unix_ms parses as 0, which is never a real
1335                # instant: treat it like the malformed ack above rather than
1336                # handing the caller 0 as the scheduled instant.
1337                if actual:
1338                    self._start_ack = (requested, actual)
1339                if requested and actual and abs(actual - requested) > 2:
1340                    # A correction on its own is self-healed: a join lands on it
1341                    # by design, and a solo start simply adopts it as the anchor.
1342                    # Only the session knows when one costs something - a group
1343                    # that has to be re-anchored to converge - and it raises that
1344                    # to a warning itself.
1345                    player.logger.info(
1346                        "AirPlay start corrected by %+d ms on %s (requested %d, scheduled %d)",
1347                        actual - requested,
1348                        player.display_name,
1349                        requested,
1350                        actual,
1351                    )
1352            # An ack and a failure are the two mutually exclusive answers to one
1353            # START, so each clears the other: the caller then reads whichever
1354            # answer released its wait, with nothing left from the previous one.
1355            self._start_error = None
1356            self._started.set()
1357        elif "[STATUS] anchor_corrected " in line:
1358            self._parse_anchor_corrected(line)
1359        elif "[STATUS] content_cut " in line:
1360            self._parse_content_cut(line)
1361        elif "[STATUS] clock_ready " in line:
1362            self._parse_clock_ready(line)
1363        elif "[STATUS] clock_verified" in line:
1364            self._parse_clock_verified(line)
1365        elif "[STATUS] flushed" in line:
1366            # A splice-timeline member reports the audible instant of its
1367            # frozen delivery head; the warm START must anchor beyond it (a
1368            # commanded instant at or behind the head splices at the head,
1369            # silently breaking the shared instant).
1370            if "head_unix_ms=" in line:
1371                try:
1372                    self.flushed_head_unix_ms = int(
1373                        line.split("head_unix_ms=")[1].split(maxsplit=1)[0]
1374                    )
1375                except ValueError, IndexError:
1376                    self.flushed_head_unix_ms = 0
1377            else:
1378                self.flushed_head_unix_ms = 0
1379            self._flush_error = None
1380            self._flushed.set()
1381        elif "[STATUS] audio " in line:
1382            self._audio_present.set()
1383        elif "[STATUS] mrp" in line:
1384            # The artwork reports arrive on stderr; the now-playing push
1385            # status arrives on stdout. One parser serves both shapes, so
1386            # each reader dispatches the mrp lines its own pipe carries.
1387            self._parse_mrp_status(line)
1388        elif "[STATUS] idle_timeout" in line:
1389            # a parked (paused) session outlived the binary's idle cap;
1390            # treat it as a normal end of stream
1391            player.logger.debug("cliairplay idle timeout reached")
1392            return True
1393        elif "[STATUS] eof" in line:
1394            player.logger.debug("End of stream reached")
1395            return True
1396        elif "[STATUS] REANCHOR" in line:
1397            self._parse_reanchor_status(line)
1398        elif "[STATUS] error " in line:
1399            self._parse_error_status(line)
1400        elif "[ERROR]" in line:
1401            player.logger.error("cliairplay: %s", line.strip())
1402        return False
1403
1404    def _update_elapsed(self, elapsed_time: float) -> None:
1405        """Update elapsed time against the current start anchor's media position."""
1406        # the binary's elapsed restarts at each START; report against its base
1407        elapsed_time += self._start_position
1408        # The binary only emits this status while actually playing, so it is
1409        # also the signal that drives the player into the PLAYING state.
1410        self.player.set_state_from_stream(
1411            state=PlaybackState.PLAYING, elapsed_time=elapsed_time, stream=self
1412        )
1413
1414    def _parse_reanchor_status(self, line: str) -> None:
1415        """
1416        Parse the machine-readable [STATUS] REANCHOR line.
1417
1418        The binary reports the shift cumulative since the last start/resume
1419        directly, so this SETS the tracked shift, using the sample rate carried
1420        on the line when present.
1421
1422        :param line: The status line, e.g. ``[STATUS] REANCHOR shifted_frames=67870
1423            total_shifted_frames=135740 sample_rate=44100``.
1424        """
1425        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1426        try:
1427            total_frames = int(fields["total_shifted_frames"])
1428        except KeyError, ValueError:
1429            return
1430        self.cumulative_shift_seconds = total_frames / self._reanchor_sample_rate(
1431            fields.get("sample_rate")
1432        )
1433        self.player.logger.debug(
1434            "cliairplay re-anchored %s after PCM starvation: cumulative shift %.3fs",
1435            self.player.display_name,
1436            self.cumulative_shift_seconds,
1437        )
1438
1439    def _reanchor_sample_rate(self, reported: str | None) -> int:
1440        """Return the frame->seconds rate, preferring a valid rate reported on the line."""
1441        if reported is not None:
1442            try:
1443                rate = int(reported)
1444            except ValueError:
1445                rate = 0
1446            if rate > 0:
1447                return rate
1448        return self.pcm_format.sample_rate or 44100
1449
1450    async def _prepare_artwork(self, image_url: str, _generation: int) -> str | None:
1451        """
1452        Return a cached JPEG path for the binary to embed.
1453
1454        The binary consumes artwork as a local file only; it does not fetch
1455        URLs. The image is flattened to JPEG and stored in the shared thumbnail
1456        cache.
1457
1458        :param image_url: The (imageproxy or remote) cover-art URL.
1459        :param _generation: Metadata generation associated with the render request.
1460        """
1461        try:
1462            return await get_image_thumb_path(
1463                self.mass,
1464                image_url,
1465                AIRPLAY_ARTWORK_SIZE,
1466                "",
1467                image_format="JPEG",
1468                flatten_transparency=True,
1469            )
1470        except Exception as err:
1471            self.player.logger.debug("Could not prepare artwork: %s", err)
1472            return None
1473
1474    async def _send_current_metadata(self, send_artwork: bool = True) -> None:
1475        """
1476        Send metadata for the media owned by the active stream.
1477
1478        :param send_artwork: Whether artwork should be rendered and sent.
1479        """
1480        metadata = self.session.media if self.session else self.player.current_media
1481        if not metadata:
1482            return
1483        progress = int(metadata.corrected_elapsed_time or 0)
1484        await self.send_metadata(progress, metadata, send_artwork=send_artwork)
1485
1486    async def _send_current_metadata_without_progress(self) -> None:
1487        """
1488        Send only the identity metadata for the active stream's media.
1489
1490        Used right after a commanded START: the anchor may still be settling,
1491        so the position correction is left to the post-anchor media-updated
1492        nudge — one settled now-playing refresh on the device instead of two.
1493        """
1494        metadata = self.session.media if self.session else self.player.current_media
1495        if not metadata:
1496            return
1497        await self.send_metadata(None, metadata)
1498
1499    async def _send_current_volume(self) -> None:
1500        """Send the player's current volume level to the device, muted as zero."""
1501        volume = 0 if self.player.volume_muted else self.player.volume_level
1502        await self.send_cli_command(f"VOLUME={volume}")
1503
1504    async def _cleanup_failed_start(self) -> None:
1505        """Release all resources owned by a cliairplay process that failed to start."""
1506        self._stopping = True
1507        self._stopped = True
1508        stdout_reader_task = self._stdout_reader_task
1509        if stdout_reader_task and not stdout_reader_task.done():
1510            stdout_reader_task.cancel()
1511            try:
1512                await stdout_reader_task
1513            except asyncio.CancelledError:
1514                pass
1515            except Exception as err:
1516                self.player.logger.debug("cliairplay stdout reader cleanup failed: %s", err)
1517        try:
1518            if self._cli_proc and not self._cli_proc.closed:
1519                await self._cli_proc.kill()
1520        finally:
1521            await self.commands_pipe.remove()
1522            self._cleanup_complete = True
1523            self._cli_proc = None
1524
1525    def _arm_start_answer(self) -> None:
1526        """Clear the slots the binary answers a START in, so only this one's answer is read."""
1527        # The ack and the failure are filled by the stderr reader while start()
1528        # waits, so both slots have to be emptied at the command itself: a
1529        # rejection left over from the previous START would otherwise be read
1530        # as this one's answer the moment an ack releases the wait.
1531        self._started.clear()
1532        self._start_ack = None
1533        self._start_error = None
1534
1535    def _arm_flush_answer(self) -> None:
1536        """Clear the slots the binary answers a FLUSH in, so only this one's answer is read."""
1537        self._flushed.clear()
1538        self._flush_error = None
1539
1540    async def _write_cli_command(self, command: str) -> bool:
1541        """Write an interactive command regardless of stream teardown state."""
1542        if not self._cli_proc or self._cli_proc.closed:
1543            return False
1544        if not command.endswith("\n"):
1545            command += "\n"
1546        command_delivered = await self.commands_pipe.write(command.encode("utf-8"))
1547        if command_delivered:
1548            self.player.last_command_sent = time.time()
1549        return command_delivered
1550
1551    def _check_password_preflight(self) -> None:
1552        """
1553        Refuse a native AirPlay 2 connect that has nothing to authenticate with.
1554
1555        A password-protected receiver answers the RTSP setup with a 401 unless the
1556        binary can present the device password or stored pairing credentials, so
1557        without either there is no point in spawning the process at all. A player
1558        in this state is already blocked from playback by ``needs_setup``, leaving
1559        this as the backstop for a device that only announced its password
1560        protection after the player was resolved as a playback target.
1561
1562        :raises PlayerCommandFailed: If the device password is missing.
1563        """
1564        target_protocol = self.player.protocol_override or self.player.protocol
1565        if target_protocol != StreamingProtocol.AIRPLAY2 or not self.player.password_required:
1566            return
1567        if self.player.config.get_value(CONF_PASSWORD):
1568            return
1569        # stored credentials keep the binary's pair-verify leg viable, and its own
1570        # failure report guides the user when that leg is rejected after all
1571        if self.player.get_setup_value(CONF_AIRPLAY_CREDENTIALS) or self.player.get_setup_value(
1572            CONF_RAOP_CREDENTIALS
1573        ):
1574            return
1575        raise self._password_required_error()
1576
1577    async def _await_connected(self, timeout: float = 10) -> None:
1578        """
1579        Wait for the binary to confirm the device connection.
1580
1581        :param timeout: Seconds to wait for the confirmation.
1582        """
1583        waiters = [
1584            asyncio.ensure_future(self._connected.wait()),
1585            asyncio.ensure_future(self._process_ended.wait()),
1586        ]
1587        try:
1588            await asyncio.wait(waiters, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
1589        finally:
1590            for waiter in waiters:
1591                waiter.cancel()
1592        if self._connected.is_set():
1593            return
1594        raise self._connect_failed_error()
1595
1596    def _connect_failed_error(self) -> Exception:
1597        """
1598        Return the error for a connection that was never established.
1599
1600        A binary that reported why it gave up produces a specific, actionable
1601        error. Everything else - including an unreported reason - keeps the plain
1602        timeout the callers already handle.
1603        """
1604        error = self._connect_error
1605        if error and error.code == CLI_ERROR_AUTH_REQUIRED:
1606            return self._password_required_error()
1607        if error and error.code == CLI_ERROR_AUTH_FAILED:
1608            return PlayerCommandFailed(
1609                f"{self.player.display_name} rejected the saved password. "
1610                "Run the setup for this player to enter it again.",
1611                translation_key="authentication_failed",
1612                translation_owner=self.player.translation_owner,
1613            )
1614        reason = f": {error.detail}" if error and error.detail else ""
1615        return TimeoutError(f"cliairplay did not connect to {self.player.display_name}{reason}")
1616
1617    def _password_required_error(self) -> PlayerCommandFailed:
1618        """Return the error that points the user at the player's setup flow."""
1619        return PlayerCommandFailed(
1620            f"{self.player.display_name} requires a password. "
1621            "Run the setup for this player to enter it.",
1622            translation_key="password_required",
1623        )
1624
1625    def _parse_error_status(self, line: str) -> None:
1626        """Parse the structured failure the binary reports and route it to its waiter."""
1627        payload = line.split("[STATUS] error ", 1)[-1]
1628        code_match = _CLI_ERROR_CODE_RE.search(payload)
1629        http_match = _CLI_ERROR_HTTP_RE.search(payload)
1630        detail_match = _CLI_ERROR_DETAIL_RE.search(payload)
1631        error = CliError(
1632            code=code_match.group(1) if code_match else "",
1633            http_status=int(http_match.group(1)) if http_match else 0,
1634            detail=detail_match.group(1) if detail_match else "",
1635        )
1636        self.player.logger.debug(
1637            "cliairplay reported an error for %s: code=%s http=%s detail=%s",
1638            self.player.display_name,
1639            error.code,
1640            error.http_status,
1641            error.detail,
1642        )
1643        # A rejected transport command leaves the connection alive, so it
1644        # answers only the ack it failed - never the connect error, which
1645        # decides how a NEW connection is reported to the user.
1646        if error.code == CLI_ERROR_START_FAILED:
1647            self._start_error = error
1648            self._started.set()
1649            return
1650        if error.code == CLI_ERROR_FLUSH_FAILED:
1651            self._flush_error = error
1652            self._flushed.set()
1653            return
1654        self._connect_error = error
1655        if error.code in (CLI_ERROR_AUTH_FAILED, CLI_ERROR_AUTH_REQUIRED):
1656            # The stored password is wrong, or the device demanded one we could
1657            # not supply (devices can enforce a password without announcing it -
1658            # e.g. an Apple TV with stale TXT records after the password was
1659            # enabled). Persist that so the player keeps offering its setup
1660            # action (across restarts) until a working password is entered,
1661            # instead of only failing at the next play attempt.
1662            self.player.set_password_invalid(True)
1663
1664    def _parse_anchor_corrected(self, line: str) -> None:
1665        """
1666        Parse a post-commit [STATUS] anchor_corrected line and re-base the position.
1667
1668        The binary emits this at most once per START, when a receiver clock
1669        exchange that only resumed after the START ack finds the committed
1670        instant infeasible: it moves the anchor forward and advances the queued
1671        content by the same amount (``content_cut_ms``), so the member still
1672        lands on the group timeline — only the reported media position shifts.
1673        That amount is what the correction ASKED for; :meth:`_parse_content_cut`
1674        reconciles it with what the cut managed to take.
1675
1676        :param line: The status line, e.g. ``[STATUS] anchor_corrected
1677            requested_unix_ms=1750000000000 from_unix_ms=1750000000400
1678            at_unix_ms=1750000000900 content_cut_ms=500``.
1679        """
1680        try:
1681            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1682            requested_unix_ms = int(fields.get("requested_unix_ms", 0))
1683            from_unix_ms = int(fields.get("from_unix_ms", 0))
1684            at_unix_ms = int(fields.get("at_unix_ms", 0))
1685            content_cut_ms = int(fields.get("content_cut_ms", 0))
1686        except ValueError, IndexError:
1687            # Malformed line: drop it rather than react to bogus numbers.
1688            return
1689        # The binary's elapsed counts only the retained content, so the
1690        # position base moves by the cut to keep reported progress exact.
1691        self._start_position += content_cut_ms / 1000
1692        # Track the cut owed the same way the base above tracks it: both
1693        # accumulate over an anchor and both are zeroed at every anchor
1694        # boundary. Overwriting here would settle only the newest correction
1695        # against a base that carries all of them.
1696        self._pending_content_cut_ms += content_cut_ms
1697        # Routine for a join start: the low join headroom defers to this
1698        # correction, which lands the anchor at exact receiver readiness. A
1699        # post-commit correction on any other START stays loud.
1700        self.player.logger.log(
1701            logging.INFO if self._start_was_join else logging.WARNING,
1702            "AirPlay anchor for %s corrected %+d ms after commit "
1703            "(requested %d, at %d, content advanced %d ms to stay in sync)",
1704            self.player.display_name,
1705            at_unix_ms - from_unix_ms,
1706            requested_unix_ms,
1707            at_unix_ms,
1708            content_cut_ms,
1709        )
1710
1711    def _parse_content_cut(self, line: str) -> None:
1712        """
1713        Settle a corrected anchor's content cut against the cut it asked for.
1714
1715        ``anchor_corrected`` reports the cut arithmetic on two instants demands,
1716        which :meth:`_parse_anchor_corrected` folds into the position base right
1717        away. This line reports what the cut actually took once the last byte is
1718        discarded, and the two disagree when the cut ended short — the input ran
1719        out inside it, or a teardown settled it. Every ms it fell short is a ms
1720        the reported position stays over-advanced by for the rest of the anchor,
1721        so the base is corrected back and the shortfall reported.
1722
1723        :param line: The status line, e.g. ``[STATUS] content_cut
1724            requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210``.
1725        """
1726        try:
1727            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1728            cut_ms = int(fields["cut_ms"])
1729            requested_ms = int(fields.get("requested_ms", 0))
1730            cut_bytes = int(fields.get("cut_bytes", 0))
1731            drain_ms = int(fields.get("drain_ms", 0))
1732        except KeyError, ValueError, IndexError:
1733            # Malformed line: drop it rather than react to bogus numbers.
1734            return
1735        applied_ms = self._pending_content_cut_ms
1736        self._pending_content_cut_ms = 0
1737        if not applied_ms:
1738            # The cut settled against an anchor whose base this stream no longer
1739            # reports on (a START or a join re-base landed in between), so there
1740            # is nothing of it left to correct.
1741            self.player.logger.debug(
1742                "AirPlay content cut on %s settled after the anchor it belonged to "
1743                "(requested %d ms, cut %d ms)",
1744                self.player.display_name,
1745                requested_ms,
1746                cut_ms,
1747            )
1748            return
1749        # A cut can miss the amount it was asked for in either direction, and
1750        # both leave the base wrong by the difference, so the magnitude decides
1751        # whether it settled cleanly. Re-basing by the signed difference then
1752        # corrects either one; only the report differs.
1753        shortfall_ms = applied_ms - cut_ms
1754        if abs(shortfall_ms) < AIRPLAY_CONTENT_CUT_TOLERANCE_MS:
1755            self.player.logger.debug(
1756                "AirPlay content cut on %s took the full %d ms (%d bytes in %d ms)",
1757                self.player.display_name,
1758                cut_ms,
1759                cut_bytes,
1760                drain_ms,
1761            )
1762            return
1763        self._start_position -= shortfall_ms / 1000
1764        if shortfall_ms > 0:
1765            self.player.logger.warning(
1766                "AirPlay content cut on %s fell %d ms short: the corrected anchor asked "
1767                "for %d ms and the cut took %d ms (%d bytes in %d ms). Reported position "
1768                "re-based by -%d ms; playback is that much ahead of the group timeline.",
1769                self.player.display_name,
1770                shortfall_ms,
1771                applied_ms,
1772                cut_ms,
1773                cut_bytes,
1774                drain_ms,
1775                shortfall_ms,
1776            )
1777            return
1778        overcut_ms = -shortfall_ms
1779        self.player.logger.warning(
1780            "AirPlay content cut on %s overran by %d ms: the corrected anchor asked "
1781            "for %d ms and the cut took %d ms (%d bytes in %d ms). Reported position "
1782            "was under-advanced by that much and is re-based by +%d ms.",
1783            self.player.display_name,
1784            overcut_ms,
1785            applied_ms,
1786            cut_ms,
1787            cut_bytes,
1788            drain_ms,
1789            overcut_ms,
1790        )
1791
1792    def _parse_clock_ready(self, line: str) -> None:
1793        """
1794        Parse a [STATUS] clock_ready line into the receiver's readiness projection.
1795
1796        A ``stalled`` state means the receiver never answered our clock and will
1797        render silence, which is warned about once per stream session.
1798
1799        :param line: The status line, e.g. ``[STATUS] clock_ready mode=ptp
1800            state=probing streak_ms=0 exchanges=1 ready_in_ms=2300
1801            ready_at_unix_ms=1750000002300``.
1802        """
1803        try:
1804            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1805            mode = fields.get("mode", "")
1806            state = fields.get("state", "")
1807            ready_at_unix_ms = int(fields.get("ready_at_unix_ms", 0))
1808        except ValueError, IndexError:
1809            # Malformed line: drop it rather than react to bogus numbers.
1810            return
1811        if state == "cold" and mode != "ntp":
1812            # No probe seen yet, so the line carries no projection; the binary
1813            # keeps reporting until one exists.
1814            return
1815        stalled = state == "stalled" and mode != "ntp"
1816        if stalled and not self._clock_stall_warned:
1817            # The receiver is not slaving to our clock at all, so it renders
1818            # silence while everything else about the session looks healthy.
1819            self._clock_stall_warned = True
1820            self.player.logger.warning(
1821                "%s has not answered the server's PTP clock (%s clock exchange(s), "
1822                "probe streak %s ms), so it will not play any audio. Check that UDP "
1823                "319/320 traffic can flow between the speaker and the server.",
1824                self.player.display_name,
1825                fields.get("exchanges", "?"),
1826                fields.get("streak_ms", "?"),
1827            )
1828        # NTP timing has no receiver clock to wait for, and a state without a
1829        # projection resolves the wait with nothing so a caller falls back
1830        # instead of blocking on evidence that will not arrive. A stalled clock
1831        # is one of those states however the line is numbered — but the caller
1832        # has to be able to tell the three apart, so each carries its own
1833        # readiness rather than a bare missing instant.
1834        if mode == "ntp":
1835            self._clock_readiness = ClockReadiness.NOT_APPLICABLE
1836        elif stalled:
1837            self._clock_readiness = ClockReadiness.STALLED
1838        else:
1839            self._clock_readiness = ClockReadiness.PROJECTED
1840        self._clock_ready_at_unix_ms = (
1841            ready_at_unix_ms if self._clock_readiness is ClockReadiness.PROJECTED else 0
1842        )
1843        self._clock_ready.set()
1844        self.player.logger.debug(
1845            "cliairplay reports the clock for %s as %s (mode=%s, readiness=%s, usable at %d)",
1846            self.player.display_name,
1847            state,
1848            mode,
1849            self._clock_readiness,
1850            self._clock_ready_at_unix_ms,
1851        )
1852
1853    def _parse_clock_verified(self, line: str) -> None:
1854        """Debug-log a [STATUS] clock_verified line; no correction means no server action."""
1855        try:
1856            margin_ms = int(line.split("margin_ms=")[1])
1857        except ValueError, IndexError:
1858            return
1859        self.player.logger.debug(
1860            "cliairplay clock verified for %s (margin %d ms)",
1861            self.player.display_name,
1862            margin_ms,
1863        )
1864
1865
1866def _status_int(fields: Mapping[str, str], key: str) -> int:
1867    """
1868    Return one integer field of a [STATUS] line, or 0 when it is unusable.
1869
1870    Status lines are read field by field so a value the binary could not format
1871    - or one an older build does not emit at all - does not take the rest of the
1872    line down with it. Every field read this way means "unreported" at 0.
1873
1874    :param fields: The parsed key=value pairs of the status line.
1875    :param key: The field to read.
1876    """
1877    try:
1878        return int(fields[key])
1879    except KeyError, ValueError:
1880        return 0
1881