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