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