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