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