music-assistant-server

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