music-assistant-server

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