music-assistant-server

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