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