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