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