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