music-assistant-server

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