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