music-assistant-server

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