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