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 then enforces receiver clock readiness and
607            holds its ack until that resolves, so the returned instant is the one
608            the caller must map the joiner's content onto. Group/solo origin
609            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 until the receiver clock
674        # verification resolves and therefore gets a much wider window than a
675        # plain start, which acks within the command round-trip. A reported
676        # failure 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/buffer (2000 ms default, clamped to
1003        # the device-reported window); there is no user override for it.
1004
1005        # The endpoint must follow the same capability decision as the binary:
1006        # legacy RAOP uses _raop, while native and RAOP-compatible AP2 use _airplay.
1007        if target_protocol == StreamingProtocol.AIRPLAY2 and airplay_info:
1008            args += ["--port", str(airplay_info.port)]
1009            args += ["--name", self.player.display_name]
1010            args += ["--hostname", str(airplay_info.server)]
1011        elif raop_info:
1012            args += ["--port", str(raop_info.port)]
1013
1014        # mDNS properties from the RAOP service (needed by the RAOP-based flows)
1015        if raop_info:
1016            args += ["--udn", raop_info.name]
1017            for prop in ("et", "md", "am", "pk", "pw", "cn"):
1018                if prop_value := raop_info.decoded_properties.get(prop):
1019                    args += [f"--{prop}", prop_value]
1020        if target_protocol == StreamingProtocol.RAOP and self.player.config.get_value(
1021            CONF_ENCRYPTION, True
1022        ):
1023            args += ["--encrypt"]
1024
1025        # Full _airplay._tcp TXT for the binary's automatic route selection.
1026        # Some receivers advertise their AP2 feature bits only on _raop.ft.
1027        txt_records = serialize_txt_records(airplay_info) if airplay_info else ""
1028        if (
1029            airplay_info
1030            and not (
1031                airplay_info.decoded_properties.get("features")
1032                or airplay_info.decoded_properties.get("ft")
1033            )
1034            and raop_info
1035            and (raop_features := raop_info.decoded_properties.get("ft"))
1036        ):
1037            txt_records = f"{txt_records} ft={raop_features}".strip()
1038        if txt_records:
1039            args += ["--txt", txt_records]
1040
1041        # HAP credentials (triggers native AP2 flow when present)
1042        if creds := self.player.get_setup_value(CONF_AIRPLAY_CREDENTIALS):
1043            creds_str = str(creds)
1044            if len(creds_str) == 192:
1045                args += ["--auth", creds_str]
1046            else:
1047                self.player.logger.warning(
1048                    "Invalid credentials length: %d (expected 192)", len(creds_str)
1049                )
1050
1051        # Legacy Apple TV RAOP pairing secret
1052        if raop_creds := self.player.get_setup_value(CONF_RAOP_CREDENTIALS):
1053            # Credentials format is "client_id:auth_secret", the binary expects the secret
1054            creds_str = str(raop_creds)
1055            auth_secret = creds_str.split(":", 1)[1] if ":" in creds_str else creds_str
1056            args += ["--secret", auth_secret]
1057
1058        # Device password
1059        if password := self.player.config.get_value(CONF_PASSWORD):
1060            args += ["--password", str(password)]
1061
1062        # Shared PTP daemon clock (multi-room sync for native AP2 streams). The
1063        # decision is made once per session and passed in, so every native AP2
1064        # member of a sync group uses the same timing source and cannot drift.
1065        # Without a caller-supplied decision (use_shared_ptp is None) the stream
1066        # gates on the daemon serving rather than merely running: between spawn and
1067        # the daemon publishing its clock there is nothing to attach to, and a
1068        # stream that asks anyway silently takes its own timing instead.
1069        if target_protocol == StreamingProtocol.AIRPLAY2:
1070            # Deeper receiver queue for devices whose pipeline starves at the
1071            # stock depth. The stored value wins; Automatic (0) resolves
1072            # through the same device-family table that seeds the config
1073            # entry default, so selecting it never downgrades a device.
1074            depth_ms = cast("int", self.player.config.get_value(CONF_BUFFER_DEPTH) or 0)
1075            if not depth_ms:
1076                depth_ms = default_buffer_depth(
1077                    self.player.device_info.manufacturer or "",
1078                    self.player.device_info.model or "",
1079                    get_decoded_property(airplay_info, "fv") if airplay_info else None,
1080                )
1081            if depth_ms:
1082                args += ["--latency", str(depth_ms)]
1083            shared_ptp = prov.ptp_daemon_ready if use_shared_ptp is None else use_shared_ptp
1084            if shared_ptp:
1085                args += ["--ptp-shared"]
1086
1087        # Local interface binding
1088        target_ip = str(self.player.device_info.ip_address)
1089        if_arg = await self.mass.streams.get_source_ip(target_ip)
1090        if if_arg:
1091            args += ["--if", if_arg]
1092
1093        # Address advertised inside the protocol (timing peers) for hosts where the
1094        # reachable address differs from the bind address (e.g. containers). The binary
1095        # treats this as authoritative for the receiver's clock-source filter and it
1096        # outranks its own connection-derived fallback, so only pass an address the user
1097        # actually configured: an auto-detected one would silence a receiver whenever it
1098        # names an interface the timing packets do not leave from.
1099        publish_arg = self.mass.streams.get_publish_ip(target_ip)
1100        if publish_arg and publish_arg != if_arg:
1101            args += ["--publish-ip", publish_arg]
1102
1103        # The addressing the stream ends up with is the first thing needed when triaging
1104        # a connection or timing issue from a user's log, and it is invisible otherwise:
1105        # both flags are dropped silently when no value applies here.
1106        self.player.logger.debug(
1107            "cliairplay network binding for player %s: if=%s publish_ip=%s",
1108            self.player.player_id,
1109            if_arg or "<all interfaces>",
1110            publish_arg or "<not configured>",
1111        )
1112
1113        # Debug level
1114        if self.prov.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1115            args += ["--debug", "10"]
1116        elif self.prov.logger.isEnabledFor(logging.DEBUG):
1117            args += ["--debug", "5"]
1118
1119        # Audio is fed continuously on the process stdin; the binary reads it into
1120        # a single ring buffer for the whole session lifetime (flushed and
1121        # refilled in place on a seek, never reconnected).
1122        args.append(self.player.address)
1123        return args
1124
1125    async def _stdout_reader(self) -> None:
1126        """
1127        Monitor stdout for the running cliairplay process.
1128
1129        The binary reports its resolved route at startup, the effective lead
1130        plus receiver-reported buffering window after connect, the audio formats
1131        the receiver advertises, and the result of MediaRemote now-playing
1132        pushes (Apple devices):
1133          [STATUS] route protocol=<raop|airplay2> flow=<...> timing=<ntp|ptp> buffered=<0|1>
1134          [STATUS] latency lead_ms=<int> device_min_frames=<int> device_max_frames=<int>
1135          [STATUS] capabilities requested=<hex> realtime_formats=<hex> realtime_known=<0|1>
1136            buffered_formats=<hex> buffered_known=<0|1>
1137          [STATUS] mrp path=<command> status=<http status>
1138          [EVENT] remote command=<play|pause|play_pause|next|previous>
1139        """
1140        if not self._cli_proc:
1141            return
1142        buffer = b""
1143        while chunk := await self._cli_proc.read(1024):
1144            buffer += chunk
1145            while b"\n" in buffer:
1146                raw_line, buffer = buffer.split(b"\n", 1)
1147                line = raw_line.decode("utf-8", errors="ignore").strip()
1148                if not line:
1149                    continue
1150                if "[STATUS] route" in line:
1151                    self._parse_route_status(line)
1152                elif "[STATUS] mrp" in line:
1153                    self._parse_mrp_status(line)
1154                elif "[STATUS] latency" in line:
1155                    self._parse_latency_status(line)
1156                elif "[STATUS] capabilities" in line:
1157                    self._parse_capabilities_status(line)
1158                elif line.startswith("[EVENT] remote command="):
1159                    self._parse_remote_event(line)
1160                self.player.logger.log(
1161                    VERBOSE_LOG_LEVEL, "cliairplay for %s: %s", self.player.display_name, line
1162                )
1163
1164    def _parse_remote_event(self, line: str) -> None:
1165        """Dispatch a normalized remote command reported by cliairplay."""
1166        command_value = line.removeprefix("[EVENT] remote command=").strip()
1167        try:
1168            command = AirPlayRemoteCommand(command_value)
1169        except ValueError:
1170            self.player.logger.warning(
1171                "Ignoring unknown cliairplay remote command: %s", command_value
1172            )
1173            return
1174        # Receiver storm guard: MediaRemote re-sends an unfulfilled transport
1175        # command at ~10 Hz (measured on tvOS: a next-track storm at end of
1176        # item drowned the whole server in seeks until playback collapsed).
1177        # No human intends repeats that fast, so only one command per window
1178        # is honored; the suppressed repeats are counted and reported loudly
1179        # as the support signal.
1180        window = (
1181            2.0 if command in (AirPlayRemoteCommand.NEXT, AirPlayRemoteCommand.PREVIOUS) else 0.5
1182        )
1183        now = time.monotonic()
1184        last = self._remote_command_last.get(command_value, 0.0)
1185        if now - last < window:
1186            self._remote_commands_suppressed += 1
1187            suppressed = self._remote_commands_suppressed
1188            if suppressed in (1, 10) or suppressed % 100 == 0:
1189                self.player.logger.warning(
1190                    "Storm guard: suppressed %d repeated remote '%s' "
1191                    "command(s) from %s within %.1fs",
1192                    suppressed,
1193                    command_value,
1194                    self.player.display_name,
1195                    window,
1196                )
1197            return
1198        self._remote_command_last[command_value] = now
1199        self._remote_commands_suppressed = 0
1200        prov = cast("AirPlayProvider", self.prov)
1201        prov.handle_remote_command(self.player, command)
1202
1203    def _parse_mrp_status(self, line: str) -> None:
1204        """
1205        Parse a [STATUS] mrp line and report how the now-playing push landed.
1206
1207        A push the device accepted is routine bookkeeping and stays at debug.
1208        A rejection is not: it is why a now-playing screen stays blank or keeps
1209        the previous track's art, with nothing else about the session looking
1210        wrong, so it is reported.
1211
1212        :param line: The status line, in one of the shapes
1213            ``[STATUS] mrp path=<command|channel> status=<int>`` or
1214            ``[STATUS] mrp artwork=<posted|rejected> ...``.
1215        """
1216        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1217        display_name = self.player.display_name
1218        if artwork := fields.get("artwork"):
1219            # The artwork variants carry no path=, and a rejection reports its
1220            # reason plus a clear_status rather than a status of its own.
1221            if artwork == "rejected":
1222                self.player.logger.warning(
1223                    "%s rejected the now-playing artwork (%s, %s bytes); its screen keeps "
1224                    "whatever art it had",
1225                    display_name,
1226                    fields.get("reason", "no reason given"),
1227                    fields.get("bytes", "?"),
1228                )
1229            else:
1230                self.player.logger.debug(
1231                    "MRP now-playing artwork accepted by %s (%s bytes, HTTP %s)",
1232                    display_name,
1233                    fields.get("bytes", "?"),
1234                    fields.get("status", "?"),
1235                )
1236            return
1237        if fields.get("path") == "channel":
1238            # Not an HTTP status: 0 = the opt-in data channel was attempted and
1239            # did not come up, 1 = established.
1240            self.player.logger.debug(
1241                "MRP data channel for %s: %s",
1242                display_name,
1243                "established" if fields.get("status") == "1" else "not established",
1244            )
1245            return
1246        # Only a 2xx means the device took the push. Nothing else on this
1247        # control channel does - a redirect is as much "not accepted" as a 4xx.
1248        # A status of 0 is the "unreported" reading (the field is missing or
1249        # unusable), which says nothing about how the push landed, so only a
1250        # reported status is judged - warning about a field the binary never
1251        # sent would report a rejection that nothing observed.
1252        status = _status_int(fields, "status")
1253        rejected = bool(status) and not (HTTPStatus.OK <= status < HTTPStatus.MULTIPLE_CHOICES)
1254        self.player.logger.log(
1255            logging.WARNING if rejected else logging.DEBUG,
1256            "MRP now-playing push (%s path) for %s: HTTP %s",
1257            fields.get("path", "?"),
1258            display_name,
1259            fields.get("status", "?"),
1260        )
1261
1262    def _parse_capabilities_status(self, line: str) -> None:
1263        """Parse the [STATUS] capabilities line and refresh the player's audio formats."""
1264        # The binary reports the format tables it read from the receiver's /info,
1265        # which corrects a device that was unreachable when it was discovered.
1266        # Only the native AirPlay 2 flow reads them; the other routes report
1267        # zeroes with the known flags unset. A change applies to the next stream.
1268        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1269        formats = 0
1270        for mask_field, known_field in (
1271            ("realtime_formats", "realtime_known"),
1272            ("buffered_formats", "buffered_known"),
1273        ):
1274            if fields.get(known_field) != "1":
1275                continue
1276            try:
1277                formats |= int(fields[mask_field], 16)
1278            except KeyError, ValueError:
1279                continue
1280        if formats and formats != self.player.advertised_audio_formats:
1281            self.player.logger.debug(
1282                "Audio formats advertised by %s changed to 0x%x",
1283                self.player.display_name,
1284                formats,
1285            )
1286            self.player.advertised_audio_formats = formats
1287
1288    def _parse_route_status(self, line: str) -> None:
1289        """Parse the [STATUS] route line and log which route this stream took."""
1290        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1291        protocol = fields.get("protocol", "")
1292        if protocol == "airplay2":
1293            flow = fields.get("flow", "")
1294            timing = fields.get("timing", "")
1295            details = "buffered" if fields.get("buffered") == "1" else flow
1296            self.active_route = f"AirPlay 2 ({details}, {timing.upper()})"
1297        else:
1298            self.active_route = "RAOP"
1299        self.player.logger.info(
1300            "Streaming to %s via %s", self.player.display_name, self.active_route
1301        )
1302
1303    def _parse_latency_status(self, line: str) -> None:
1304        """Parse and store the [STATUS] latency line reported by the binary."""
1305        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1306        # Read field by field: one unusable value used to abandon the rest of
1307        # the line, leaving whatever came after it stale from the previous
1308        # report while the values before it had already moved. Every field here
1309        # means "unreported" at 0, so a missing or malformed one lands there.
1310        self.latency_lead_ms = _status_int(fields, "lead_ms")
1311        self.device_min_frames = _status_int(fields, "device_min_frames")
1312        self.device_max_frames = _status_int(fields, "device_max_frames")
1313        self.warm_lead_ms = _status_int(fields, "warm_lead_ms")
1314        self.player.logger.debug(
1315            "Device latency for %s: lead=%dms, warm lead=%dms, "
1316            "buffer window=%d-%d frames (0=unreported)",
1317            self.player.display_name,
1318            self.latency_lead_ms,
1319            self.warm_lead_ms,
1320            self.device_min_frames,
1321            self.device_max_frames,
1322        )
1323
1324    async def _stderr_reader(self) -> None:
1325        """
1326        Monitor stderr for the running cliairplay process.
1327
1328        The binary emits normalized [STATUS] messages:
1329          [STATUS] connected
1330          [STATUS] playing elapsed_ms=<ms>
1331          [STATUS] paused
1332          [STATUS] eof
1333          [STATUS] announce_started at_unix_ms=<ms> duration_ms=<ms>
1334          [STATUS] announce_done [cancelled=1]
1335          [STATUS] error code=<slug> http=<int> detail="<short text>"
1336            (auth_required/auth_failed/connect_failed are terminal; the
1337             start_failed/flush_failed/announce_failed command slugs answer
1338             a pending ack)
1339          [ERROR] <message>
1340        """
1341        player = self.player
1342        logger = player.logger
1343        if not self._cli_proc:
1344            return
1345        async for line in self._cli_proc.iter_stderr():
1346            if self._stopped:
1347                break
1348            if self._handle_status_line(line):
1349                self.ended_cleanly = True
1350                break
1351            # Routine binary output is verbose-only so it never floods a user's log, but
1352            # its own diagnostics (a failed socket bind, a missing receiver clock) are the
1353            # first thing needed when triaging silent playback, so those stay visible.
1354            # Every line names its speaker: the provider logger is shared, so the output
1355            # of several concurrent streams is otherwise impossible to tell apart.
1356            level = (
1357                logging.WARNING
1358                if any(marker in line.lower() for marker in CLI_PROBLEM_MARKERS)
1359                else VERBOSE_LOG_LEVEL
1360            )
1361            logger.log(level, "cliairplay for %s: %s", player.display_name, line)
1362            await asyncio.sleep(0)
1363
1364        logger.debug("cliairplay stderr reader ended for %s", player.display_name)
1365        self._process_ended.set()
1366        if not self._stopped and not self._stopping:
1367            self._stopped = True
1368            try:
1369                if not self.ended_cleanly:
1370                    logger.warning(
1371                        "cliairplay process stopped unexpectedly for %s", player.display_name
1372                    )
1373                    # Candidates for the automatic re-join: the leader this member
1374                    # was synced to (plus its other members, in case leadership
1375                    # transfers while the backoff runs), or - when this was the
1376                    # leader itself - the members that survive it. Captured before
1377                    # the ungroup below mutates the group state (create_task
1378                    # starts eagerly).
1379                    was_leader = bool(player.group_members)
1380                    if player.synced_to:
1381                        rejoin_candidates = [player.synced_to]
1382                        if leader := self.mass.players.get_player(player.synced_to):
1383                            rejoin_candidates += [
1384                                member_id
1385                                for member_id in leader.group_members
1386                                if member_id not in (player.player_id, player.synced_to)
1387                            ]
1388                    else:
1389                        rejoin_candidates = [
1390                            m for m in player.group_members if m != player.player_id
1391                        ]
1392                    # Hand off to the player controller so it drops just this member, or
1393                    # transfers leadership to a healthy member, instead of dissolving the
1394                    # whole group over a single dead transport. A sync leader is left in
1395                    # its current state here on purpose: the controller only transfers
1396                    # leadership while the queue still looks active, and transfer_queue or
1397                    # dissolve sets the final state.
1398                    # One exception: a member that is a STATIC member of an active group
1399                    # player must not go through cmd_ungroup - the controller interprets
1400                    # unjoining a static member as releasing the whole group (HA unjoin
1401                    # semantics), which would silence every room over one dead transport.
1402                    # Its membership is configuration; drop only this member from the
1403                    # leader's live session instead. The set_members call cannot bounce
1404                    # back to the group player: the controller only redirects it when
1405                    # the group advertises SET_MEMBERS, which a static group never does.
1406                    static_member_of = self._static_group_membership(player)
1407                    if static_member_of and player.synced_to:
1408                        self.mass.create_task(
1409                            self.mass.players.cmd_set_members(
1410                                player.synced_to, player_ids_to_remove=[player.player_id]
1411                            )
1412                        )
1413                    else:
1414                        self.mass.create_task(self.mass.players.cmd_ungroup(player.player_id))
1415                    if rejoin_candidates:
1416                        # the group (or its successor) may still be playing:
1417                        # schedule bounded attempts to re-join it
1418                        player.schedule_group_rejoin(rejoin_candidates)
1419                    if was_leader:
1420                        return
1421                player.set_state_from_stream(state=PlaybackState.IDLE, elapsed_time=0, stream=self)
1422            finally:
1423                await self.commands_pipe.remove()
1424
1425    def _static_group_membership(self, player: AirPlayPlayer) -> str | None:
1426        """Return the active group player id the player is a static member of, if any."""
1427        active_group_id = player.state.active_group
1428        if not active_group_id:
1429            return None
1430        group_player = self.mass.players.get_player(active_group_id)
1431        if group_player and player.player_id in group_player.static_group_members:
1432            return active_group_id
1433        return None
1434
1435    def _handle_status_line(self, line: str) -> bool:  # noqa: PLR0915
1436        """Dispatch one cliairplay status line; True ends the stderr loop."""
1437        player = self.player
1438        if "[STATUS] connected" in line:
1439            self._connected.set()
1440            # whatever the device accepted just now is a working password
1441            player.set_password_invalid(False)
1442        elif "[STATUS] playing elapsed_ms=" in line:
1443            try:
1444                millis = int(line.split("elapsed_ms=")[1])
1445            except ValueError, IndexError:
1446                pass
1447            else:
1448                self._update_elapsed(millis / 1000)
1449        elif "[STATUS] paused" in line:
1450            player.set_state_from_stream(state=PlaybackState.PAUSED, stream=self)
1451        elif "[STATUS] started " in line:
1452            try:
1453                fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1454                requested = int(fields.get("requested_unix_ms", 0))
1455                actual = int(fields.get("at_unix_ms", 0))
1456            except ValueError, IndexError:
1457                # Malformed ack: leave _start_ack unset so the caller falls
1458                # back to trusting the commanded instant, without waiting out
1459                # the ack timeout.
1460                pass
1461            else:
1462                # A line missing at_unix_ms parses as 0, which is never a real
1463                # instant: treat it like the malformed ack above rather than
1464                # handing the caller 0 as the scheduled instant.
1465                if actual:
1466                    self._start_ack = (requested, actual)
1467                if requested and actual and abs(actual - requested) > 2:
1468                    # A correction on its own is self-healed: a join lands on it
1469                    # by design, and a solo start simply adopts it as the anchor.
1470                    # Only the session knows when one costs something - a group
1471                    # that has to be re-anchored to converge - and it raises that
1472                    # to a warning itself.
1473                    player.logger.info(
1474                        "AirPlay start corrected by %+d ms on %s (requested %d, scheduled %d)",
1475                        actual - requested,
1476                        player.display_name,
1477                        requested,
1478                        actual,
1479                    )
1480            # An ack and a failure are the two mutually exclusive answers to one
1481            # START, so each clears the other: the caller then reads whichever
1482            # answer released its wait, with nothing left from the previous one.
1483            self._start_error = None
1484            self._started.set()
1485        elif "[STATUS] anchor_corrected " in line:
1486            self._parse_anchor_corrected(line)
1487        elif "[STATUS] content_cut " in line:
1488            self._parse_content_cut(line)
1489        elif "[STATUS] clock_ready " in line:
1490            self._parse_clock_ready(line)
1491        elif "[STATUS] clock_verified" in line:
1492            self._parse_clock_verified(line)
1493        elif "[STATUS] flushed" in line:
1494            # A splice-timeline member reports the audible instant of its
1495            # frozen delivery head; the warm START must anchor beyond it (a
1496            # commanded instant at or behind the head splices at the head,
1497            # silently breaking the shared instant).
1498            if "head_unix_ms=" in line:
1499                try:
1500                    self.flushed_head_unix_ms = int(
1501                        line.split("head_unix_ms=")[1].split(maxsplit=1)[0]
1502                    )
1503                except ValueError, IndexError:
1504                    self.flushed_head_unix_ms = 0
1505            else:
1506                self.flushed_head_unix_ms = 0
1507            self._flush_error = None
1508            self._flushed.set()
1509        elif "[STATUS] announce_started" in line:
1510            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1511            self._announce_ack = (
1512                _status_int(fields, "at_unix_ms"),
1513                _status_int(fields, "duration_ms"),
1514            )
1515            # The started report and a reported announce failure are the two
1516            # mutually exclusive answers to one arm, so each clears the other.
1517            self._announce_error = None
1518            self._announce_started.set()
1519        elif "[STATUS] announce_done" in line:
1520            self._announce_done_cancelled = "cancelled=1" in line
1521            self._announce_done.set()
1522        elif "[STATUS] audio " in line:
1523            if "buffered_ms=" in line:
1524                try:
1525                    self.audio_pending_ms = int(line.split("buffered_ms=")[1].split(maxsplit=1)[0])
1526                except ValueError, IndexError:
1527                    self.audio_pending_ms = 0
1528            else:
1529                self.audio_pending_ms = 0
1530            self._audio_present.set()
1531        elif "[STATUS] mrp" in line:
1532            # The artwork reports arrive on stderr; the now-playing push
1533            # status arrives on stdout. One parser serves both shapes, so
1534            # each reader dispatches the mrp lines its own pipe carries.
1535            self._parse_mrp_status(line)
1536        elif "[STATUS] idle_timeout" in line:
1537            # a parked (paused) session outlived the binary's idle cap;
1538            # treat it as a normal end of stream
1539            player.logger.debug("cliairplay idle timeout reached")
1540            return True
1541        elif "[STATUS] eof" in line:
1542            player.logger.debug("End of stream reached")
1543            return True
1544        elif "[STATUS] REANCHOR" in line:
1545            self._parse_reanchor_status(line)
1546        elif "[STATUS] error " in line:
1547            self._parse_error_status(line)
1548        elif "[ERROR]" in line:
1549            player.logger.error("cliairplay: %s", line.strip())
1550        return False
1551
1552    def _update_elapsed(self, elapsed_time: float) -> None:
1553        """Update elapsed time against the current start anchor's media position."""
1554        # the binary's elapsed restarts at each START; report against its base
1555        elapsed_time += self._start_position
1556        # The binary only emits this status while actually playing, so it is
1557        # also the signal that drives the player into the PLAYING state.
1558        self.player.set_state_from_stream(
1559            state=PlaybackState.PLAYING, elapsed_time=elapsed_time, stream=self
1560        )
1561
1562    def _parse_reanchor_status(self, line: str) -> None:
1563        """
1564        Parse the machine-readable [STATUS] REANCHOR line.
1565
1566        The binary reports the shift cumulative since the last start/resume
1567        directly, so this SETS the tracked shift, using the sample rate carried
1568        on the line when present.
1569
1570        :param line: The status line, e.g. ``[STATUS] REANCHOR shifted_frames=67870
1571            total_shifted_frames=135740 sample_rate=44100``.
1572        """
1573        fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1574        try:
1575            total_frames = int(fields["total_shifted_frames"])
1576        except KeyError, ValueError:
1577            return
1578        self.cumulative_shift_seconds = total_frames / self._reanchor_sample_rate(
1579            fields.get("sample_rate")
1580        )
1581        self.player.logger.debug(
1582            "cliairplay re-anchored %s after PCM starvation: cumulative shift %.3fs",
1583            self.player.display_name,
1584            self.cumulative_shift_seconds,
1585        )
1586
1587    def _reanchor_sample_rate(self, reported: str | None) -> int:
1588        """Return the frame->seconds rate, preferring a valid rate reported on the line."""
1589        if reported is not None:
1590            try:
1591                rate = int(reported)
1592            except ValueError:
1593                rate = 0
1594            if rate > 0:
1595                return rate
1596        return self.pcm_format.sample_rate or 44100
1597
1598    async def _prepare_artwork(self, image_url: str, _generation: int) -> str | None:
1599        """
1600        Return a cached JPEG path for the binary to embed.
1601
1602        The binary consumes artwork as a local file only; it does not fetch
1603        URLs. The image is flattened to JPEG and stored in the shared thumbnail
1604        cache.
1605
1606        :param image_url: The (imageproxy or remote) cover-art URL.
1607        :param _generation: Metadata generation associated with the render request.
1608        """
1609        try:
1610            return await get_image_thumb_path(
1611                self.mass,
1612                image_url,
1613                AIRPLAY_ARTWORK_SIZE,
1614                "",
1615                image_format="JPEG",
1616                flatten_transparency=True,
1617            )
1618        except Exception as err:
1619            self.player.logger.debug("Could not prepare artwork: %s", err)
1620            return None
1621
1622    async def _send_current_metadata(self, send_artwork: bool = True) -> None:
1623        """
1624        Send metadata for the media owned by the active stream.
1625
1626        :param send_artwork: Whether artwork should be rendered and sent.
1627        """
1628        metadata = self.session.media if self.session else self.player.current_media
1629        if not metadata:
1630            return
1631        progress = int(metadata.corrected_elapsed_time or 0)
1632        await self.send_metadata(progress, metadata, send_artwork=send_artwork)
1633
1634    async def _send_current_metadata_without_progress(self) -> None:
1635        """
1636        Send only the identity metadata for the active stream's media.
1637
1638        Used right after a commanded START: the anchor may still be settling,
1639        so the position correction is left to the post-anchor media-updated
1640        nudge — one settled now-playing refresh on the device instead of two.
1641        """
1642        metadata = self.session.media if self.session else self.player.current_media
1643        if not metadata:
1644            return
1645        await self.send_metadata(None, metadata)
1646
1647    async def _send_current_volume(self) -> None:
1648        """Send the player's current volume level to the device, muted as zero."""
1649        volume = 0 if self.player.volume_muted else self.player.volume_level
1650        await self.send_cli_command(f"VOLUME={volume}")
1651
1652    async def _cleanup_failed_start(self) -> None:
1653        """Release all resources owned by a cliairplay process that failed to start."""
1654        self._stopping = True
1655        self._stopped = True
1656        stdout_reader_task = self._stdout_reader_task
1657        if stdout_reader_task and not stdout_reader_task.done():
1658            stdout_reader_task.cancel()
1659            try:
1660                await stdout_reader_task
1661            except asyncio.CancelledError:
1662                pass
1663            except Exception as err:
1664                self.player.logger.debug("cliairplay stdout reader cleanup failed: %s", err)
1665        try:
1666            if self._cli_proc and not self._cli_proc.closed:
1667                await self._cli_proc.kill()
1668        finally:
1669            await self.commands_pipe.remove()
1670            self._cleanup_complete = True
1671            self._cli_proc = None
1672
1673    def _arm_start_answer(self) -> None:
1674        """Clear the slots the binary answers a START in, so only this one's answer is read."""
1675        # The ack and the failure are filled by the stderr reader while start()
1676        # waits, so both slots have to be emptied at the command itself: a
1677        # rejection left over from the previous START would otherwise be read
1678        # as this one's answer the moment an ack releases the wait.
1679        self._started.clear()
1680        self._start_ack = None
1681        self._start_error = None
1682
1683    def _arm_flush_answer(self) -> None:
1684        """Clear the slots the binary answers a FLUSH in, so only this one's answer is read."""
1685        self._flushed.clear()
1686        self._flush_error = None
1687
1688    def _arm_announce_answer(self) -> None:
1689        """Clear the slots the binary answers an ANNOUNCE in, so only this one's answer is read."""
1690        self._announce_started.clear()
1691        self._announce_done.clear()
1692        self._announce_ack = None
1693        self._announce_error = None
1694        self._announce_done_cancelled = False
1695
1696    async def _write_cli_command(self, command: str) -> bool:
1697        """Write an interactive command regardless of stream teardown state."""
1698        if not self._cli_proc or self._cli_proc.closed:
1699            return False
1700        if not command.endswith("\n"):
1701            command += "\n"
1702        command_delivered = await self.commands_pipe.write(command.encode("utf-8"))
1703        if command_delivered:
1704            self.player.last_command_sent = time.time()
1705        return command_delivered
1706
1707    def _check_password_preflight(self) -> None:
1708        """
1709        Refuse a native AirPlay 2 connect that has nothing to authenticate with.
1710
1711        A password-protected receiver answers the RTSP setup with a 401 unless the
1712        binary can present the device password or stored pairing credentials, so
1713        without either there is no point in spawning the process at all. A player
1714        in this state is already blocked from playback by ``needs_setup``, leaving
1715        this as the backstop for a device that only announced its password
1716        protection after the player was resolved as a playback target.
1717
1718        :raises PlayerCommandFailed: If the device password is missing.
1719        """
1720        target_protocol = self.player.protocol_override or self.player.protocol
1721        if target_protocol != StreamingProtocol.AIRPLAY2 or not self.player.password_required:
1722            return
1723        if self.player.config.get_value(CONF_PASSWORD):
1724            return
1725        # stored credentials keep the binary's pair-verify leg viable, and its own
1726        # failure report guides the user when that leg is rejected after all
1727        if self.player.get_setup_value(CONF_AIRPLAY_CREDENTIALS) or self.player.get_setup_value(
1728            CONF_RAOP_CREDENTIALS
1729        ):
1730            return
1731        raise self._password_required_error()
1732
1733    async def _await_connected(self, timeout: float = 10) -> None:
1734        """
1735        Wait for the binary to confirm the device connection.
1736
1737        :param timeout: Seconds to wait for the confirmation.
1738        """
1739        waiters = [
1740            asyncio.ensure_future(self._connected.wait()),
1741            asyncio.ensure_future(self._process_ended.wait()),
1742        ]
1743        try:
1744            await asyncio.wait(waiters, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
1745        finally:
1746            for waiter in waiters:
1747                waiter.cancel()
1748        if self._connected.is_set():
1749            return
1750        raise self._connect_failed_error()
1751
1752    def _connect_failed_error(self) -> Exception:
1753        """
1754        Return the error for a connection that was never established.
1755
1756        A binary that reported why it gave up produces a specific, actionable
1757        error. Everything else - including an unreported reason - keeps the plain
1758        timeout the callers already handle.
1759        """
1760        error = self._connect_error
1761        if error and error.code == CLI_ERROR_AUTH_REQUIRED:
1762            return self._password_required_error()
1763        if error and error.code == CLI_ERROR_AUTH_FAILED:
1764            return PlayerCommandFailed(
1765                f"{self.player.display_name} rejected the saved password. "
1766                "Run the setup for this player to enter it again.",
1767                translation_key="authentication_failed",
1768                translation_owner=self.player.translation_owner,
1769            )
1770        reason = f": {error.detail}" if error and error.detail else ""
1771        return TimeoutError(f"cliairplay did not connect to {self.player.display_name}{reason}")
1772
1773    def _password_required_error(self) -> PlayerCommandFailed:
1774        """Return the error that points the user at the player's setup flow."""
1775        return PlayerCommandFailed(
1776            f"{self.player.display_name} requires a password. "
1777            "Run the setup for this player to enter it.",
1778            translation_key="password_required",
1779        )
1780
1781    def _parse_error_status(self, line: str) -> None:
1782        """Parse the structured failure the binary reports and route it to its waiter."""
1783        payload = line.split("[STATUS] error ", 1)[-1]
1784        code_match = _CLI_ERROR_CODE_RE.search(payload)
1785        http_match = _CLI_ERROR_HTTP_RE.search(payload)
1786        detail_match = _CLI_ERROR_DETAIL_RE.search(payload)
1787        error = CliError(
1788            code=code_match.group(1) if code_match else "",
1789            http_status=int(http_match.group(1)) if http_match else 0,
1790            detail=detail_match.group(1) if detail_match else "",
1791        )
1792        self.player.logger.debug(
1793            "cliairplay reported an error for %s: code=%s http=%s detail=%s",
1794            self.player.display_name,
1795            error.code,
1796            error.http_status,
1797            error.detail,
1798        )
1799        # A rejected transport command leaves the connection alive, so it
1800        # answers only the ack it failed - never the connect error, which
1801        # decides how a NEW connection is reported to the user.
1802        if error.code == CLI_ERROR_START_FAILED:
1803            self._start_error = error
1804            self._started.set()
1805            return
1806        if error.code == CLI_ERROR_FLUSH_FAILED:
1807            self._flush_error = error
1808            self._flushed.set()
1809            return
1810        if error.code == CLI_ERROR_ANNOUNCE_FAILED:
1811            # A rejected arm plays nothing, so both announce waits are answered
1812            # at once - nothing else will ever answer them.
1813            self._announce_error = error
1814            self._announce_started.set()
1815            self._announce_done.set()
1816            return
1817        self._connect_error = error
1818        if error.code in (CLI_ERROR_AUTH_FAILED, CLI_ERROR_AUTH_REQUIRED):
1819            # The stored password is wrong, or the device demanded one we could
1820            # not supply (devices can enforce a password without announcing it -
1821            # e.g. an Apple TV with stale TXT records after the password was
1822            # enabled). Persist that so the player keeps offering its setup
1823            # action (across restarts) until a working password is entered,
1824            # instead of only failing at the next play attempt.
1825            self.player.set_password_invalid(True)
1826
1827    def _parse_anchor_corrected(self, line: str) -> None:
1828        """
1829        Parse a post-commit [STATUS] anchor_corrected line and re-base the position.
1830
1831        The binary emits this at most once per START, when a receiver clock
1832        exchange that only resumed after the START ack finds the committed
1833        instant infeasible: it moves the anchor forward and advances the queued
1834        content by the same amount (``content_cut_ms``), so the member still
1835        lands on the group timeline — only the reported media position shifts.
1836        That amount is what the correction ASKED for; :meth:`_parse_content_cut`
1837        reconciles it with what the cut managed to take.
1838
1839        :param line: The status line, e.g. ``[STATUS] anchor_corrected
1840            requested_unix_ms=1750000000000 from_unix_ms=1750000000400
1841            at_unix_ms=1750000000900 content_cut_ms=500``.
1842        """
1843        try:
1844            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1845            requested_unix_ms = int(fields.get("requested_unix_ms", 0))
1846            from_unix_ms = int(fields.get("from_unix_ms", 0))
1847            at_unix_ms = int(fields.get("at_unix_ms", 0))
1848            content_cut_ms = int(fields.get("content_cut_ms", 0))
1849        except ValueError, IndexError:
1850            # Malformed line: drop it rather than react to bogus numbers.
1851            return
1852        # The binary's elapsed counts only the retained content, so the
1853        # position base moves by the cut to keep reported progress exact.
1854        self._start_position += content_cut_ms / 1000
1855        # Track the cut owed the same way the base above tracks it: both
1856        # accumulate over an anchor and both are zeroed at every anchor
1857        # boundary. Overwriting here would settle only the newest correction
1858        # against a base that carries all of them.
1859        self._pending_content_cut_ms += content_cut_ms
1860        # Routine for a join start: the low join headroom defers to this
1861        # correction, which lands the anchor at exact receiver readiness. A
1862        # post-commit correction on any other START stays loud.
1863        self.player.logger.log(
1864            logging.INFO if self._start_was_join else logging.WARNING,
1865            "AirPlay anchor for %s corrected %+d ms after commit "
1866            "(requested %d, at %d, content advanced %d ms to stay in sync)",
1867            self.player.display_name,
1868            at_unix_ms - from_unix_ms,
1869            requested_unix_ms,
1870            at_unix_ms,
1871            content_cut_ms,
1872        )
1873
1874    def _parse_content_cut(self, line: str) -> None:
1875        """
1876        Settle a corrected anchor's content cut against the cut it asked for.
1877
1878        ``anchor_corrected`` reports the cut arithmetic on two instants demands,
1879        which :meth:`_parse_anchor_corrected` folds into the position base right
1880        away. This line reports what the cut actually took once the last byte is
1881        discarded, and the two disagree when the cut ended short — the input ran
1882        out inside it, or a teardown settled it. Every ms it fell short is a ms
1883        the reported position stays over-advanced by for the rest of the anchor,
1884        so the base is corrected back and the shortfall reported.
1885
1886        :param line: The status line, e.g. ``[STATUS] content_cut
1887            requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210``.
1888        """
1889        try:
1890            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1891            cut_ms = int(fields["cut_ms"])
1892            requested_ms = int(fields.get("requested_ms", 0))
1893            cut_bytes = int(fields.get("cut_bytes", 0))
1894            drain_ms = int(fields.get("drain_ms", 0))
1895        except KeyError, ValueError, IndexError:
1896            # Malformed line: drop it rather than react to bogus numbers.
1897            return
1898        applied_ms = self._pending_content_cut_ms
1899        self._pending_content_cut_ms = 0
1900        if not applied_ms:
1901            # The cut settled against an anchor whose base this stream no longer
1902            # reports on (a START or a join re-base landed in between), so there
1903            # is nothing of it left to correct.
1904            self.player.logger.debug(
1905                "AirPlay content cut on %s settled after the anchor it belonged to "
1906                "(requested %d ms, cut %d ms)",
1907                self.player.display_name,
1908                requested_ms,
1909                cut_ms,
1910            )
1911            return
1912        # A cut can miss the amount it was asked for in either direction, and
1913        # both leave the base wrong by the difference, so the magnitude decides
1914        # whether it settled cleanly. Re-basing by the signed difference then
1915        # corrects either one; only the report differs.
1916        shortfall_ms = applied_ms - cut_ms
1917        if abs(shortfall_ms) < AIRPLAY_CONTENT_CUT_TOLERANCE_MS:
1918            self.player.logger.debug(
1919                "AirPlay content cut on %s took the full %d ms (%d bytes in %d ms)",
1920                self.player.display_name,
1921                cut_ms,
1922                cut_bytes,
1923                drain_ms,
1924            )
1925            return
1926        self._start_position -= shortfall_ms / 1000
1927        if shortfall_ms > 0:
1928            self.player.logger.warning(
1929                "AirPlay content cut on %s fell %d ms short: the corrected anchor asked "
1930                "for %d ms and the cut took %d ms (%d bytes in %d ms). Reported position "
1931                "re-based by -%d ms; playback is that much ahead of the group timeline.",
1932                self.player.display_name,
1933                shortfall_ms,
1934                applied_ms,
1935                cut_ms,
1936                cut_bytes,
1937                drain_ms,
1938                shortfall_ms,
1939            )
1940            return
1941        overcut_ms = -shortfall_ms
1942        self.player.logger.warning(
1943            "AirPlay content cut on %s overran by %d ms: the corrected anchor asked "
1944            "for %d ms and the cut took %d ms (%d bytes in %d ms). Reported position "
1945            "was under-advanced by that much and is re-based by +%d ms.",
1946            self.player.display_name,
1947            overcut_ms,
1948            applied_ms,
1949            cut_ms,
1950            cut_bytes,
1951            drain_ms,
1952            overcut_ms,
1953        )
1954
1955    def _parse_clock_ready(self, line: str) -> None:
1956        """
1957        Parse a [STATUS] clock_ready line into the receiver's readiness projection.
1958
1959        A ``stalled`` state means the receiver never answered our clock and will
1960        render silence, which is warned about once per stream session.
1961
1962        :param line: The status line, e.g. ``[STATUS] clock_ready mode=ptp
1963            state=probing streak_ms=0 exchanges=1 ready_in_ms=2300
1964            ready_at_unix_ms=1750000002300``.
1965        """
1966        try:
1967            fields = dict(part.split("=", 1) for part in line.split() if "=" in part)
1968            mode = fields.get("mode", "")
1969            state = fields.get("state", "")
1970            ready_at_unix_ms = int(fields.get("ready_at_unix_ms", 0))
1971        except ValueError, IndexError:
1972            # Malformed line: drop it rather than react to bogus numbers.
1973            return
1974        if state == "cold" and mode != "ntp":
1975            # No probe seen yet, so the line carries no projection; the binary
1976            # keeps reporting until one exists.
1977            return
1978        stalled = state == "stalled" and mode != "ntp"
1979        if stalled and not self._clock_stall_warned:
1980            # The receiver is not slaving to our clock at all, so it renders
1981            # silence while everything else about the session looks healthy.
1982            self._clock_stall_warned = True
1983            self.player.logger.warning(
1984                "%s has not answered the server's PTP clock (%s clock exchange(s), "
1985                "probe streak %s ms), so it will not play any audio. Check that UDP "
1986                "319/320 traffic can flow between the speaker and the server.",
1987                self.player.display_name,
1988                fields.get("exchanges", "?"),
1989                fields.get("streak_ms", "?"),
1990            )
1991        # NTP timing has no receiver clock to wait for, and a state without a
1992        # projection resolves the wait with nothing so a caller falls back
1993        # instead of blocking on evidence that will not arrive. A stalled clock
1994        # is one of those states however the line is numbered — but the caller
1995        # has to be able to tell the three apart, so each carries its own
1996        # readiness rather than a bare missing instant.
1997        if mode == "ntp":
1998            self._clock_readiness = ClockReadiness.NOT_APPLICABLE
1999        elif stalled:
2000            self._clock_readiness = ClockReadiness.STALLED
2001        else:
2002            self._clock_readiness = ClockReadiness.PROJECTED
2003        self._clock_ready_at_unix_ms = (
2004            ready_at_unix_ms if self._clock_readiness is ClockReadiness.PROJECTED else 0
2005        )
2006        self._clock_ready.set()
2007        self.player.logger.debug(
2008            "cliairplay reports the clock for %s as %s (mode=%s, readiness=%s, usable at %d)",
2009            self.player.display_name,
2010            state,
2011            mode,
2012            self._clock_readiness,
2013            self._clock_ready_at_unix_ms,
2014        )
2015
2016    def _parse_clock_verified(self, line: str) -> None:
2017        """Debug-log a [STATUS] clock_verified line; no correction means no server action."""
2018        try:
2019            margin_ms = int(line.split("margin_ms=")[1])
2020        except ValueError, IndexError:
2021            return
2022        self.player.logger.debug(
2023            "cliairplay clock verified for %s (margin %d ms)",
2024            self.player.display_name,
2025            margin_ms,
2026        )
2027
2028
2029def _artwork_identity(image_url: str) -> str:
2030    """
2031    Return a URL-form independent identity for a cover-art URL.
2032
2033    :param image_url: The cover-art URL as carried on the player media.
2034    """
2035    # An imageproxy URL embeds a server base URL (webserver or stream server,
2036    # depending on who built the PlayerMedia) plus size/format parameters, but
2037    # the opaque image id in its path alone identifies the underlying image.
2038    # Any other URL is its own identity.
2039    return _extract_imageproxy_id(image_url) or image_url
2040
2041
2042def _status_int(fields: Mapping[str, str], key: str) -> int:
2043    """
2044    Return one integer field of a [STATUS] line, or 0 when it is unusable.
2045
2046    Status lines are read field by field so a value the binary could not format
2047    - or one an older build does not emit at all - does not take the rest of the
2048    line down with it. Every field read this way means "unreported" at 0.
2049
2050    :param fields: The parsed key=value pairs of the status line.
2051    :param key: The field to read.
2052    """
2053    try:
2054        return int(fields[key])
2055    except KeyError, ValueError:
2056        return 0
2057