music-assistant-server

1"""AirPlay Player provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import json
8import logging
9import os
10import socket
11import time
12from contextlib import suppress
13from ipaddress import IPv4Address, IPv6Address, ip_address
14from typing import TYPE_CHECKING, Final, cast
15
16from music_assistant_models.config_entries import ConfigEntry
17from music_assistant_models.enums import ConfigEntryType, PlaybackState
18from music_assistant_models.errors import MediaNotFoundError
19from zeroconf import NonUniqueNameException, ServiceStateChange
20from zeroconf.asyncio import AsyncServiceInfo
21
22from music_assistant.constants import (
23    CONF_LOG_LEVEL,
24    CONF_PLAYERS,
25    CONF_PROVIDERS,
26    VERBOSE_LOG_LEVEL,
27)
28from music_assistant.helpers.config_entries import (
29    CONF_CONNECTED_PLAYERS,
30    CONF_PUBLISH_NAME_TEMPLATE,
31    resolve_publish_name,
32)
33from music_assistant.helpers.datetime import utc
34from music_assistant.helpers.json import SerializableType
35from music_assistant.helpers.process import AsyncProcess
36from music_assistant.helpers.util import (
37    get_ip_addresses,
38    get_ip_pton,
39    get_primary_ip_address_from_zeroconf,
40    select_free_port,
41)
42from music_assistant.models.player_provider import PlayerProvider
43
44from .constants import (
45    AIRPLAY_DISCOVERY_TYPE,
46    AIRPLAY_VOLUME_MUTE,
47    CLI_PROBLEM_MARKERS,
48    COMPANION_DISCOVERY_TYPE,
49    CONF_PASSWORD_INVALID,
50    CONF_PASSWORD_MARKERS_REVIEWED,
51    CONF_STORED_VOLUME,
52    CONF_VERBOSE_PTP_LOGGING,
53    DACP_DISCOVERY_TYPE,
54    EXTERNAL_ARTWORK_PATH_PREFIX,
55    FALLBACK_VOLUME,
56    MRP_DISCOVERY_TYPE,
57    PTP_DAEMON_WARN_BURST,
58    PTP_DAEMON_WARN_WINDOW,
59    RAOP_DISCOVERY_TYPE,
60    AirPlayRemoteCommand,
61    StreamingProtocol,
62)
63from .control_player import AirPlayControlPlayer
64from .dashboard import AirPlayDashboards
65from .helpers import (
66    convert_airplay_volume,
67    get_cli_binary,
68    get_model_info,
69    is_apple_device,
70    probe_audio_formats,
71)
72from .player import AirPlayPlayer, GenericAirPlayPlayer
73from .sendspin_bridge import SendspinBridgeManager
74
75if TYPE_CHECKING:
76    from music_assistant_models.config_entries import ProviderConfig
77
78# Marker the `cliairplay --ptp-daemon` process prints once it has bound the
79# privileged PTP ports (UDP 319/320) and opened its control channel. Until this
80# line is seen the daemon is spawned but not yet able to serve shared-clock
81# streams, so a group start before it would race the daemon's readiness.
82PTP_DAEMON_READY_MARKER: Final[str] = "[PTP] daemon up"
83# Bounded wait for the daemon to report readiness before a stream session
84# decides its timing source. Once ready the check returns immediately; only a
85# session that starts while the daemon is still coming up (or never binds) pays
86# any of this, and only up to the moment readiness is signalled.
87PTP_DAEMON_READY_TIMEOUT: Final[float] = 3.0
88# Grace period after broadcasting a goodbye for a stale DACP registration before
89# re-registering the (name-stable) service, letting the cache flush the old record.
90DACP_RECLAIM_DELAY: Final[float] = 1.0
91# Opt-in for pyatv's own debug logging. Set to any non-empty value to trace the
92# pyatv protocol traffic itself.
93ENV_PYATV_DEBUG: Final[str] = "MASS_PYATV_DEBUG"
94
95
96class AirPlayProvider(PlayerProvider):
97    """Player provider for AirPlay based players."""
98
99    reload_on_streams_network_change = True
100    _dacp_server: asyncio.Server
101    _dacp_info: AsyncServiceInfo
102    _bridge_manager: SendspinBridgeManager
103    dashboards: AirPlayDashboards
104    _ptp_daemon: AsyncProcess | None = None
105    _ptp_daemon_stdout_task: asyncio.Task[None] | None = None
106    _ptp_daemon_started: float = 0.0
107    _ptp_daemon_restarted: bool = False
108    _ptp_daemon_stop_requested: bool = False
109    # Set once the running daemon reports it has bound 319/320 and opened its
110    # control channel; created/cleared per daemon start so a crash+restart
111    # re-gates readiness. None until the daemon is first started.
112    _ptp_daemon_ready: asyncio.Event | None = None
113    # Rate-limit state for daemon lines promoted to WARNING (see
114    # PTP_DAEMON_WARN_BURST).
115    _ptp_daemon_warn_window_start: float | None = None
116    _ptp_daemon_warns_in_window: int = 0
117    _ptp_daemon_warns_suppressed: int = 0
118
119    @property
120    def bridge_manager(self) -> SendspinBridgeManager:
121        """Return the Sendspin bridge manager."""
122        return self._bridge_manager
123
124    @property
125    def ptp_daemon_running(self) -> bool:
126        """
127        Return if the shared PTP clock daemon process is alive.
128
129        This reflects process liveness only (spawned, not closed, still
130        running). It says nothing about whether streams can use it: a daemon
131        that never bound its ports stays alive and reports True here while every
132        group silently degrades to NTP. Use :attr:`ptp_daemon_ready` for the
133        real state, or :meth:`wait_ptp_daemon_ready` to wait for it.
134        """
135        return (
136            self._ptp_daemon is not None
137            and not self._ptp_daemon.closed
138            and self._ptp_daemon.returncode is None
139        )
140
141    @property
142    def ptp_daemon_ready(self) -> bool:
143        """
144        Return whether the shared PTP clock daemon is serving streams right now.
145
146        This is the condition a session actually gates on: the daemon has bound
147        the privileged PTP ports (UDP 319/320), opened its control channel, and
148        is still alive.
149        """
150        return (
151            self.ptp_daemon_running
152            and self._ptp_daemon_ready is not None
153            and self._ptp_daemon_ready.is_set()
154        )
155
156    async def wait_ptp_daemon_ready(self, timeout: float = PTP_DAEMON_READY_TIMEOUT) -> bool:
157        """
158        Wait until the shared PTP clock daemon is ready to serve streams.
159
160        Readiness means the daemon has bound the privileged PTP ports (UDP
161        319/320) and opened its control channel - not merely that the process is
162        alive. A stream session gates its group-wide timing decision on this so
163        it never attaches members to a clock that is not yet serving.
164
165        :param timeout: Maximum seconds to wait for the readiness signal.
166        :return: True if the daemon has signalled readiness, False otherwise
167            (never started, failed to bind, or not ready within the timeout).
168        """
169        event = self._ptp_daemon_ready
170        daemon = self._ptp_daemon
171        if event is None or daemon is None or daemon.closed or daemon.returncode is not None:
172            return False
173        if event.is_set():
174            return True
175
176        ready_task = asyncio.create_task(event.wait())
177        exit_task = asyncio.create_task(daemon.wait())
178        try:
179            done, _ = await asyncio.wait(
180                (ready_task, exit_task),
181                timeout=timeout,
182                return_when=asyncio.FIRST_COMPLETED,
183            )
184            return (
185                ready_task in done
186                and event.is_set()
187                and self._ptp_daemon is daemon
188                and not daemon.closed
189                and daemon.returncode is None
190            )
191        finally:
192            ready_task.cancel()
193            exit_task.cancel()
194            await asyncio.gather(ready_task, exit_task, return_exceptions=True)
195
196    def handle_remote_command(self, player: AirPlayPlayer, command: AirPlayRemoteCommand) -> None:
197        """Dispatch a transport command received from an AirPlay receiver."""
198        player_id = (
199            self.bridge_manager.get_transport_command_target(player.player_id) or player.player_id
200        )
201        match command:
202            case AirPlayRemoteCommand.PLAY:
203                # Some receivers echo play as confirmation of a command from MA.
204                if player.playback_state != PlaybackState.PLAYING:
205                    self.mass.create_task(self.mass.players.cmd_play(player_id))
206            case AirPlayRemoteCommand.PAUSE:
207                if player.playback_state == PlaybackState.PLAYING:
208                    self.mass.create_task(self.mass.players.cmd_pause(player_id))
209            case AirPlayRemoteCommand.PLAY_PAUSE:
210                self.mass.create_task(self.mass.players.cmd_play_pause(player_id))
211            case AirPlayRemoteCommand.NEXT:
212                self.mass.create_task(self.mass.players.cmd_next_track(player_id))
213            case AirPlayRemoteCommand.PREVIOUS:
214                self.mass.create_task(self.mass.players.cmd_previous_track(player_id))
215
216    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
217        """Return Config entries to configure this provider."""
218        return (
219            ConfigEntry(
220                key=CONF_VERBOSE_PTP_LOGGING,
221                type=ConfigEntryType.BOOLEAN,
222                default_value=False,
223                required=False,
224                advanced=True,
225            ),
226        )
227
228    async def handle_async_init(self) -> None:
229        """Handle async initialization of the provider."""
230        self._set_pyatv_log_level()
231        self._drop_unverified_password_markers()
232        self._companion_info_by_address: dict[str, AsyncServiceInfo] = {}
233        self._mrp_info_by_address: dict[str, AsyncServiceInfo] = {}
234        # Shared audible instants for in-flight announcements, keyed by
235        # (group-or-leader player id, render key) -> unix ms. The controller
236        # forwards a group-entity announcement to every member concurrently;
237        # each call arms its own member and reuses the instant the first call
238        # planned, so all rooms render the clip in sync (managed by
239        # announce.py, pruned as plans pass).
240        self._announce_plans: dict[tuple[str, str], int] = {}
241
242        # Initialize Sendspin bridge manager for protocol linking
243        self._bridge_manager = SendspinBridgeManager(self)
244
245        # Registers eligible Apple TVs as dashboard endpoints for the tvOS app
246        self.dashboards = AirPlayDashboards(self)
247
248        # register DACP zeroconf service
249        dacp_port = await select_free_port(39831, 49831)
250        # Use first 16 hex chars of server_id as a persistent DACP ID
251        # This ensures the DACP ID remains the same across restarts, which is required
252        # for AirPlay 2 (HAP) pair-verify to work with previously paired devices
253        self.dacp_id = dacp_id = self.mass.server_id[:16].upper()
254        self.logger.debug("Starting DACP ActiveRemote %s on port %s", dacp_id, dacp_port)
255        self._dacp_server = await asyncio.start_server(self._handle_dacp_request, port=dacp_port)
256        server_id = f"iTunes_Ctrl_{dacp_id}.{DACP_DISCOVERY_TYPE}"
257        self._dacp_info = AsyncServiceInfo(
258            DACP_DISCOVERY_TYPE,
259            name=server_id,
260            addresses=[await get_ip_pton(self.mass.streams.publish_ip)],
261            port=dacp_port,
262            properties={
263                "txtvers": "1",
264                "Ver": "63B5E5C0C201542E",
265                "DbId": "63B5E5C0C201542E",
266                "OSsi": "0x1F5",
267            },
268            server=f"{socket.gethostname()}.local",
269        )
270        await self._register_dacp_service()
271
272        # Run one shared PTP clock daemon for the provider lifetime: all native
273        # AirPlay 2 streams attach to it (--ptp-shared) so multi-room sync groups
274        # lock to a single grandmaster while UDP 319/320 is bound only once.
275        await self._start_ptp_daemon()
276
277    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
278        """Handle logic when the config is updated."""
279        await super().update_config(config, changed_keys)
280        # a log level(-only) change does not reload the provider,
281        # so realign pyatv's logger here
282        if f"values/{CONF_LOG_LEVEL}" in changed_keys:
283            self._set_pyatv_log_level()
284
285    async def on_mdns_service_state_change(
286        self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
287    ) -> None:
288        """Handle MDNS service state callback."""
289        if (info and info.type == COMPANION_DISCOVERY_TYPE) or name.endswith(
290            COMPANION_DISCOVERY_TYPE
291        ):
292            await self._handle_companion_service_state_change(name, state_change, info)
293            return
294        if (info and info.type == MRP_DISCOVERY_TYPE) or name.endswith(MRP_DISCOVERY_TYPE):
295            await self._handle_mrp_service_state_change(name, state_change, info)
296            return
297        if not info:
298            if state_change == ServiceStateChange.Removed and "@" in name:
299                # Service name is enough to mark the player as unavailable on 'Removed' notification
300                raw_id, display_name = name.split(".", maxsplit=1)[0].split("@", 1)
301            else:
302                # If we are not in a 'Removed' state, we need info to be filled to update the player
303                return
304        elif "@" in info.name:
305            raw_id, display_name = info.name.split(".")[0].split("@", 1)
306        elif deviceid := info.decoded_properties.get("deviceid"):
307            raw_id = deviceid.replace(":", "")
308            display_name = info.name.split(".")[0]
309        else:
310            return
311        player_id = f"ap{raw_id.lower()}"
312        # handle removed player
313        if state_change == ServiceStateChange.Removed:
314            if _player := self.mass.players.get_player(player_id):
315                # the player has become unavailable
316                self.logger.debug("Player offline: %s", _player.display_name)
317                # Remove the Sendspin bridge first
318                await self._bridge_manager.remove_bridge(player_id)
319                await self.mass.players.unregister(player_id)
320            self.dashboards.unregister(player_id)
321            return
322        # handle update for existing device
323        assert info is not None  # type guard
324        player: AirPlayPlayer | None
325        if player := cast("AirPlayPlayer | None", self.mass.players.get_player(player_id)):
326            # update the latest discovery info for existing player
327            player.set_discovery_info(info, display_name)
328            # only control players can ever be dashboard endpoints
329            if isinstance(player, AirPlayControlPlayer):
330                self.dashboards.reconcile(player_id)
331            return
332        await self._setup_player(player_id, display_name, info)
333
334    async def unload(self, is_removed: bool = False) -> None:
335        """Handle unload/close of the provider."""
336        # Unregister all dashboard endpoints
337        dashboards = getattr(self, "dashboards", None)
338        if dashboards:
339            await dashboards.unload()
340        # Stop all Sendspin bridges
341        bridge_manager = getattr(self, "_bridge_manager", None)
342        if bridge_manager:
343            await bridge_manager.close()
344        # terminate the shared PTP clock daemon
345        self._ptp_daemon_stop_requested = True
346        if self._ptp_daemon_ready is not None:
347            self._ptp_daemon_ready.clear()
348        ptp_stdout_task = self._ptp_daemon_stdout_task
349        if ptp_stdout_task and not ptp_stdout_task.done():
350            ptp_stdout_task.cancel()
351            with suppress(asyncio.CancelledError):
352                await ptp_stdout_task
353        if self._ptp_daemon and not self._ptp_daemon.closed:
354            await self._ptp_daemon.close()
355        self._ptp_daemon = None
356        # shutdown DACP server
357        if self._dacp_server:
358            self._dacp_server.close()
359        # shutdown DACP zeroconf service
360        if self._dacp_info:
361            await self.mass.discovery.aiozc.async_unregister_service(self._dacp_info)
362
363    async def get_diagnostics(self) -> dict[str, SerializableType]:
364        """Return diagnostics info for this provider to include in diagnostics reports."""
365        streams_by_type: dict[str, int] = {}
366        streams_by_route: dict[str, int] = {}
367        for player in self.get_players():
368            if not (player.stream and player.stream.running):
369                continue
370            stream_type = "airplay2" if player.protocol == StreamingProtocol.AIRPLAY2 else "raop"
371            streams_by_type[stream_type] = streams_by_type.get(stream_type, 0) + 1
372            # The route the binary resolved names the timing source each stream
373            # really got (PTP or NTP), which the daemon flags cannot: a daemon
374            # can be alive and every stream still be running on NTP. A process
375            # that has not reported its route yet is counted apart rather than
376            # folded into either.
377            route = player.stream.active_route or "unreported"
378            streams_by_route[route] = streams_by_route.get(route, 0) + 1
379        return {
380            "dacp_server_running": self._dacp_server.is_serving(),
381            "ptp_daemon_running": self.ptp_daemon_running,
382            "ptp_daemon_ready": self.ptp_daemon_ready,
383            "active_streams": sum(streams_by_type.values()),
384            "streams_by_type": streams_by_type,
385            "streams_by_route": streams_by_route,
386        }
387
388    def get_players(self) -> list[AirPlayPlayer]:
389        """Return all airplay players belonging to this instance."""
390        return cast("list[AirPlayPlayer]", self.players)
391
392    def get_player(self, player_id: str) -> AirPlayPlayer | None:
393        """Return AirplayPlayer by id."""
394        return cast("AirPlayPlayer | None", self.mass.players.get_player(player_id))
395
396    async def resolve_image(self, path: str) -> bytes:
397        """
398        Resolve artwork for the current external media on an Apple device.
399
400        :param path: AirPlay artwork path produced for the image proxy.
401        :return: Raw artwork bytes.
402        :raises MediaNotFoundError: If the artwork is invalid, stale, or unavailable.
403        """
404        try:
405            prefix, player_id, artwork_id = path.split("/", 2)
406        except ValueError as err:
407            raise MediaNotFoundError("Invalid AirPlay artwork path") from err
408        player = self.get_player(player_id)
409        if (
410            prefix != EXTERNAL_ARTWORK_PATH_PREFIX
411            or not artwork_id
412            or not isinstance(player, AirPlayControlPlayer)
413        ):
414            raise MediaNotFoundError("AirPlay artwork is unavailable")
415        return await player.async_get_external_artwork(artwork_id)
416
417    def _set_pyatv_log_level(self) -> None:
418        """Keep pyatv's (very chatty) logging quiet unless it is explicitly asked for."""
419        # pyatv logs every protocol message, HTTP exchange and encrypted payload of
420        # each control connection at debug level, which buries our own logging and
421        # rotates the log file within minutes. Its debug output is therefore held
422        # back even on verbose sessions, which are meant to surface our own deep
423        # diagnostics (the cliairplay [STATUS] and PTP traces) rather than pyatv's.
424        if os.environ.get(ENV_PYATV_DEBUG):
425            logging.getLogger("pyatv").setLevel(logging.DEBUG)
426        else:
427            logging.getLogger("pyatv").setLevel(max(self.logger.level + 10, logging.INFO))
428
429    async def _setup_player(
430        self, player_id: str, display_name: str, discovery_info: AsyncServiceInfo
431    ) -> None:
432        """Handle setup of a new player that is discovered using mdns."""
433        # return early if player is disabled in config
434        if not self.mass.config.get_raw_player_config_value(player_id, "enabled", True):
435            self.logger.debug("Ignoring %s in discovery as it is disabled.", display_name)
436            return
437        # Filter out this server's own AirPlay Receiver (shairport-sync) instances
438        # before anything else: they must never register as AirPlay players.
439        if await self._is_own_airplay_receiver(display_name, discovery_info):
440            self.logger.debug(
441                "Ignoring %s in discovery: it is an AirPlay Receiver instance of this server",
442                display_name,
443            )
444            return
445        raop_discovery_info: AsyncServiceInfo | None = None
446        airplay_discovery_info: AsyncServiceInfo | None = None
447        if discovery_info.type == RAOP_DISCOVERY_TYPE:
448            # RAOP service discovered - try to also find the AirPlay service
449            raop_discovery_info = discovery_info
450            self.logger.debug("Discovered RAOP service for %s", display_name)
451            airplay_discovery_info = await self.mass.discovery.async_find_mdns_service(
452                AIRPLAY_DISCOVERY_TYPE, display_name, timeout=10.0
453            )
454        else:
455            # AirPlay service discovered - try to also find the RAOP service
456            self.logger.debug("Discovered AirPlay service for %s", display_name)
457            airplay_discovery_info = discovery_info
458            raop_discovery_info = await self.mass.discovery.async_find_mdns_service(
459                RAOP_DISCOVERY_TYPE, display_name, timeout=10.0
460            )
461
462        if airplay_discovery_info:
463            model_discovery_info = airplay_discovery_info
464        elif raop_discovery_info:
465            model_discovery_info = raop_discovery_info
466        else:
467            return  # should not happen, but guard just in case
468        manufacturer, model = get_model_info(model_discovery_info)
469
470        prefer_ipv6 = ":" in str(self.mass.streams.publish_ip)
471        address = get_primary_ip_address_from_zeroconf(discovery_info, prefer_ipv6=prefer_ipv6)
472        if not address:
473            return  # should not happen, but guard just in case
474
475        # if we reach this point, all preflights are ok and we can create the player
476        self.logger.debug("Discovered AirPlay device %s on %s", display_name, address)
477
478        # Get stored volume from playerconfig
479        volume = int(
480            self.mass.config.get_raw_player_config_value(
481                player_id, CONF_STORED_VOLUME, FALLBACK_VOLUME
482            )
483        )
484
485        # Final check before registration to handle race conditions
486        # (multiple MDNS events processed in parallel for same device)
487        if self.mass.players.get_player(player_id):
488            self.logger.debug(
489                "Player %s already registered during setup, skipping registration", player_id
490            )
491            return
492
493        self.logger.debug(
494            "Setting up player %s: manufacturer=%s, model=%s",
495            display_name,
496            manufacturer,
497            model,
498        )
499
500        player_addresses = self._get_discovery_addresses(
501            airplay_discovery_info,
502            raop_discovery_info,
503        )
504        player_addresses.add(address)
505        # Apple TVs and HomePods are standalone players with a native control
506        # plane (Companion/MRP); all other receivers are protocol endpoints.
507        # The model is decided from the device's own identity only: it defines
508        # the player id exposed to API consumers (e.g. Home Assistant), so it
509        # must not vary with the discovery timing of the separate Companion/MRP
510        # mDNS records. Which control features are offered on a control player
511        # is decided from the advertised capabilities instead.
512        enhanced_control = is_apple_device(manufacturer, model)
513        companion_info: AsyncServiceInfo | None = None
514        mrp_info: AsyncServiceInfo | None = None
515        if enhanced_control:
516            companion_info, mrp_info = await asyncio.gather(
517                self._get_related_discovery_info(
518                    COMPANION_DISCOVERY_TYPE,
519                    self._companion_info_by_address,
520                    player_addresses,
521                    display_name,
522                ),
523                self._get_related_discovery_info(
524                    MRP_DISCOVERY_TYPE,
525                    self._mrp_info_by_address,
526                    player_addresses,
527                    display_name,
528                ),
529            )
530
531        player: AirPlayPlayer
532        if enhanced_control:
533            player = AirPlayControlPlayer(
534                provider=self,
535                player_id=player_id,
536                raop_discovery_info=raop_discovery_info,
537                airplay_discovery_info=airplay_discovery_info,
538                companion_discovery_info=companion_info,
539                mrp_discovery_info=mrp_info,
540                address=address,
541                display_name=display_name,
542                manufacturer=manufacturer,
543                model=model,
544                initial_volume=volume,
545            )
546        else:
547            player = GenericAirPlayPlayer(
548                provider=self,
549                player_id=player_id,
550                raop_discovery_info=raop_discovery_info,
551                airplay_discovery_info=airplay_discovery_info,
552                address=address,
553                display_name=display_name,
554                manufacturer=manufacturer,
555                model=model,
556                initial_volume=volume,
557            )
558        await self.mass.players.register(player)
559
560        # A receiver only publishes its audio formats (and so whether it can do
561        # 24-bit) in its /info response, never in its mDNS records, so ask it
562        # directly. Off the discovery path: mdns callbacks are serialized per
563        # provider, so an unreachable device must not hold up the next player.
564        if airplay_discovery_info and airplay_discovery_info.port:
565            self.mass.create_task(
566                self._learn_audio_formats(player, address, airplay_discovery_info.port)
567            )
568
569        # Set up Sendspin bridge for protocol linking (if Sendspin provider is available)
570        await self._bridge_manager.evaluate_bridge(player)
571
572        # Track control players (Apple TVs) for dashboard eligibility
573        if isinstance(player, AirPlayControlPlayer):
574            self.dashboards.setup_player(player)
575
576    async def _learn_audio_formats(self, player: AirPlayPlayer, host: str, port: int) -> None:
577        """Read the audio formats a receiver advertises, so 24-bit can be auto-enabled."""
578        if formats := await probe_audio_formats(self.mass, host, port):
579            player.advertised_audio_formats = formats
580
581    async def _is_own_airplay_receiver(
582        self, display_name: str, discovery_info: AsyncServiceInfo
583    ) -> bool:
584        """
585        Return whether a discovered service is an AirPlay Receiver instance of this server.
586
587        :param display_name: The advertised device name (mdns name without any id prefix).
588        :param discovery_info: The mdns service info that triggered the discovery.
589        """
590        from music_assistant.providers.airplay_receiver import (  # noqa: PLC0415
591            airplay_receiver_ports,
592        )
593
594        # Collect the advertised names and ports of all configured AirPlay Receiver
595        # instances from raw config storage: this works regardless of provider load
596        # order at boot, when the receiver instances may not be running yet.
597        receiver_names: set[str] = set()
598        receiver_ports: set[int] = set()
599        for instance_id, raw_conf in self.mass.config.get(CONF_PROVIDERS, {}).items():
600            if not isinstance(raw_conf, dict) or raw_conf.get("domain") != "airplay_receiver":
601                continue
602            if not raw_conf.get("enabled", True):
603                continue
604            values = raw_conf.get("values")
605            values = values if isinstance(values, dict) else {}
606            player_ids = [str(player_id) for player_id in values.get(CONF_CONNECTED_PLAYERS) or []]
607            receiver_ports.update(airplay_receiver_ports(str(instance_id), player_ids).values())
608            template = values.get(CONF_PUBLISH_NAME_TEMPLATE)
609            for player_id in player_ids:
610                # best effort: a registered player's live name, else its stored config
611                # name; when neither resolves the port match stays the strong signal
612                if player := self.mass.players.get_player(player_id):
613                    player_name: str | None = player.display_name
614                else:
615                    stored_name = self.mass.config.get_raw_player_config_value(
616                        player_id, "name"
617                    ) or self.mass.config.get_raw_player_config_value(player_id, "default_name")
618                    player_name = str(stored_name) if stored_name else None
619                if player_name:
620                    receiver_names.add(resolve_publish_name(template, player_name))
621        # running instances are authoritative for the actual daemon ports
622        for prov in self.mass.get_provider_instances("airplay_receiver"):
623            if ports := getattr(prov, "airplay_ports", None):
624                receiver_ports.update(ports)
625        if not receiver_names and not receiver_ports:
626            return False
627
628        # The advertisement must originate from this host. shairport-sync's embedded
629        # mDNS responder announces address records for every host interface and which
630        # one resolves (first) is undefined, so match the full advertised address set
631        # against all of this host's addresses instead of a single picked address.
632        advertised_ips: set[IPv4Address | IPv6Address] = set()
633        for value in discovery_info.parsed_addresses():
634            with suppress(ValueError):
635                advertised_ips.add(ip_address(value))
636        if not advertised_ips:
637            return False
638        if not any(advertised_ip.is_loopback for advertised_ip in advertised_ips):
639            host_ips: set[IPv4Address | IPv6Address] = set()
640            for value in await get_ip_addresses(include_ipv6=True):
641                with suppress(ValueError):
642                    host_ips.add(ip_address(value))
643            if advertised_ips.isdisjoint(host_ips):
644                return False
645
646        # Match by advertised name (robust even when the TXT record did not resolve).
647        if display_name in receiver_names:
648            return True
649        # Match by port, gated on the shairport-sync model so user-run AirPlay
650        # receivers on this machine (e.g. the macOS built-in receiver, which also
651        # listens on port 7000) remain usable as players when their name differs.
652        _, model = get_model_info(discovery_info)
653        return model == "ShairportSync" and discovery_info.port in receiver_ports
654
655    async def _handle_companion_service_state_change(
656        self,
657        name: str,
658        state_change: ServiceStateChange,
659        info: AsyncServiceInfo | None,
660    ) -> None:
661        """Associate a Companion service with its controlled AirPlay player."""
662        if state_change == ServiceStateChange.Removed:
663            # Keep the last endpoint while the device sleeps. Some devices can
664            # withdraw services from mDNS before Companion reports the asleep
665            # state, but the existing endpoint is still needed to wake them.
666            return
667        if info is None:
668            return
669
670        addresses = self._cache_control_service(self._companion_info_by_address, info)
671        player = self._find_airplay_player(addresses)
672        if isinstance(player, AirPlayControlPlayer):
673            await player.set_companion_discovery_info(info)
674
675    async def _handle_mrp_service_state_change(
676        self,
677        name: str,
678        state_change: ServiceStateChange,
679        info: AsyncServiceInfo | None,
680    ) -> None:
681        """Associate an MRP service with its controlled AirPlay player."""
682        if state_change == ServiceStateChange.Removed or info is None:
683            return
684        addresses = self._cache_control_service(self._mrp_info_by_address, info)
685        player = self._find_airplay_player(addresses)
686        if isinstance(player, AirPlayControlPlayer):
687            await player.set_mrp_discovery_info(info)
688
689    async def _get_related_discovery_info(
690        self,
691        service_type: str,
692        info_by_address: dict[str, AsyncServiceInfo],
693        player_addresses: set[str],
694        display_name: str,
695    ) -> AsyncServiceInfo | None:
696        """Return a related control service from cache or mDNS discovery."""
697        for address in player_addresses:
698            if discovery_info := info_by_address.get(address):
699                return discovery_info
700        if discovery_info := await self._find_cached_discovery_info(
701            service_type,
702            player_addresses,
703        ):
704            self._cache_control_service(info_by_address, discovery_info)
705            return discovery_info
706        discovery_info = await self.mass.discovery.async_find_mdns_service(
707            service_type,
708            display_name,
709        )
710        if discovery_info is None:
711            discovery_info = await self._find_cached_discovery_info(
712                service_type,
713                player_addresses,
714            )
715            if discovery_info is None:
716                return None
717        if player_addresses.isdisjoint(discovery_info.parsed_addresses()):
718            return None
719        self._cache_control_service(info_by_address, discovery_info)
720        return discovery_info
721
722    async def _find_cached_discovery_info(
723        self,
724        service_type: str,
725        player_addresses: set[str],
726    ) -> AsyncServiceInfo | None:
727        """Find a cached mDNS service sharing an address with the AirPlay endpoint."""
728        service_type_lower = service_type.lower()
729        for mdns_name in set(self.mass.discovery.aiozc.zeroconf.cache.cache):
730            if service_type_lower not in mdns_name or mdns_name == service_type_lower:
731                continue
732            discovery_info = AsyncServiceInfo(service_type, mdns_name)
733            if not await discovery_info.async_request(
734                self.mass.discovery.aiozc.zeroconf,
735                3000,
736            ):
737                continue
738            if not player_addresses.isdisjoint(discovery_info.parsed_addresses()):
739                return discovery_info
740        return None
741
742    def _find_airplay_player(self, addresses: set[str]) -> AirPlayPlayer | None:
743        """Return the AirPlay player matching any advertised address."""
744        return next(
745            (
746                candidate
747                for candidate in self.get_players()
748                if not addresses.isdisjoint(
749                    {
750                        candidate.address,
751                        *self._get_discovery_addresses(
752                            candidate.airplay_discovery_info,
753                            candidate.raop_discovery_info,
754                            candidate.mrp_discovery_info
755                            if isinstance(candidate, AirPlayControlPlayer)
756                            else None,
757                        ),
758                    }
759                )
760            ),
761            None,
762        )
763
764    @staticmethod
765    def _cache_control_service(
766        info_by_address: dict[str, AsyncServiceInfo],
767        info: AsyncServiceInfo,
768    ) -> set[str]:
769        """Cache a control service by address and return its current addresses."""
770        addresses = set(info.parsed_addresses())
771        # Drop addresses this service no longer advertises (e.g. after a DHCP
772        # change), so a stale entry can never classify a future device that
773        # gets the old address assigned.
774        for stale_address in [
775            address
776            for address, cached in info_by_address.items()
777            if cached.name == info.name and address not in addresses
778        ]:
779            del info_by_address[stale_address]
780        for address in addresses:
781            info_by_address[address] = info
782        return addresses
783
784    @staticmethod
785    def _get_discovery_addresses(
786        *discovery_infos: AsyncServiceInfo | None,
787    ) -> set[str]:
788        """Return all IP addresses advertised by the given mDNS services."""
789        return {
790            address
791            for discovery_info in discovery_infos
792            if discovery_info is not None
793            for address in discovery_info.parsed_addresses()
794        }
795
796    async def _register_dacp_service(self) -> None:
797        """
798        Register the DACP ActiveRemote mDNS service, reclaiming a stale name if needed.
799
800        The service name is derived from the (persistent) server id, so it is identical
801        across every reload and restart. A previous registration that was not cleanly
802        torn down - a reload racing a slow initial load, or a leftover from a prior crash -
803        keeps the same name in the zeroconf cache and makes a plain register raise
804        NonUniqueNameException. In that case we broadcast a goodbye for the name to flush
805        the stale record, then register again.
806        """
807        aiozc = self.mass.discovery.aiozc
808        try:
809            await aiozc.async_register_service(self._dacp_info)
810            return
811        except NonUniqueNameException:
812            self.logger.debug(
813                "DACP service %s already advertised - reclaiming stale registration",
814                self._dacp_info.name,
815            )
816        await aiozc.async_unregister_service(self._dacp_info)
817        await asyncio.sleep(DACP_RECLAIM_DELAY)
818        await aiozc.async_register_service(self._dacp_info)
819
820    async def _start_ptp_daemon(self) -> None:
821        """Spawn the shared PTP clock daemon (cliairplay --ptp-daemon)."""
822        try:
823            cli_binary = await get_cli_binary()
824        except RuntimeError as err:
825            self.logger.warning(
826                "cliairplay binary unavailable (%s) - "
827                "PTP timing is degraded, AirPlay 2 streams fall back to NTP",
828                err,
829            )
830            return
831        # Advertise the same grandmaster identity as the per-stream sessions
832        # (which derive their PTP clock id from --dacp), so receivers see one
833        # consistent clock whether the daemon or an in-process engine serves it.
834        args = [cli_binary, "--ptp-daemon", "--dacp", self.dacp_id]
835        if_arg = await self.mass.streams.get_source_ip()
836        if if_arg:
837            args += ["--if", if_arg]
838        # The daemon runs quiet by default: its per-packet PTP tracing
839        # (Announce/Sync/Delay_Req, ~10 lines/s) needs BOTH verbose logging and
840        # the dedicated opt-in, so ordinary verbose sessions are not flooded
841        # with timing chatter that only matters for clock-sync debugging.
842        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) and self.config.get_value(
843            CONF_VERBOSE_PTP_LOGGING
844        ):
845            args += ["--debug", "10"]
846        # The binding the daemon ends up with is the first thing needed when triaging
847        # timing issues from a user's log, and it is invisible otherwise: --if is left
848        # out entirely for a default bind ip.
849        self.logger.debug("Starting shared PTP clock daemon: if=%s", if_arg or "<all interfaces>")
850        daemon = AsyncProcess(args, stdout=True, stderr=True, name="cliairplay-ptp-daemon")
851        # (Re)gate readiness for this daemon instance: not ready until a reader
852        # sees the "daemon up" line (a restart clears any previous readiness).
853        if self._ptp_daemon_ready is None:
854            self._ptp_daemon_ready = asyncio.Event()
855        else:
856            self._ptp_daemon_ready.clear()
857        self._reset_ptp_daemon_warn_budget()
858        await daemon.start()
859        self._ptp_daemon = daemon
860        self._ptp_daemon_started = time.monotonic()
861        daemon.attach_stderr_reader(self.mass.create_task(self._ptp_daemon_stderr_reader(daemon)))
862        self._ptp_daemon_stdout_task = self.mass.create_task(self._ptp_daemon_stdout_reader(daemon))
863        self.mass.create_task(self._ptp_daemon_monitor(daemon))
864
865    async def _ptp_daemon_stderr_reader(self, daemon: AsyncProcess) -> None:
866        """Forward PTP daemon stderr output to the debug log."""
867        async for line in daemon.iter_stderr():
868            self._handle_ptp_daemon_line(line)
869
870    async def _ptp_daemon_stdout_reader(self, daemon: AsyncProcess) -> None:
871        """Drain (and debug-log) the PTP daemon stdout pipe."""
872        buffer = b""
873        while chunk := await daemon.read(1024):
874            buffer += chunk
875            while b"\n" in buffer:
876                raw_line, buffer = buffer.split(b"\n", 1)
877                if line := raw_line.decode("utf-8", errors="ignore").strip():
878                    self._handle_ptp_daemon_line(line)
879
880    def _handle_ptp_daemon_line(self, line: str) -> None:
881        """Log a PTP daemon output line and detect its readiness signal."""
882        # Routine daemon output is verbose-only so it never floods a user's log
883        # (the per-packet timing trace only runs at verbose in the first place).
884        lowered = line.lower()
885        if any(marker in lowered for marker in CLI_PROBLEM_MARKERS):
886            self._warn_ptp_daemon_line(line)
887        else:
888            self.logger.log(VERBOSE_LOG_LEVEL, "PTP daemon: %s", line)
889        # The readiness marker is matched on either pipe: the daemon's diagnostic
890        # output is not contractually stdout-vs-stderr, so both readers feed this
891        # handler and setting the event is idempotent.
892        if (
893            (event := self._ptp_daemon_ready) is not None
894            and not event.is_set()
895            and PTP_DAEMON_READY_MARKER in line
896        ):
897            self.logger.debug("Shared PTP clock daemon reported ready")
898            event.set()
899
900    def _reset_ptp_daemon_warn_budget(self) -> None:
901        """
902        Give a newly spawned daemon a full warning budget of its own.
903
904        The budget is per daemon, not per provider: a daemon that spent it
905        before crashing would otherwise have its replacement's startup failure -
906        the one line worth reading - suppressed for the rest of the window.
907        Whatever the old daemon had held back is reported on the way out rather
908        than dropped.
909        """
910        if self._ptp_daemon_warns_suppressed:
911            self.logger.warning(
912                "PTP daemon: %d further problem line(s) from the previous daemon were "
913                "suppressed; enable verbose logging to see them all",
914                self._ptp_daemon_warns_suppressed,
915            )
916        self._ptp_daemon_warn_window_start = None
917        self._ptp_daemon_warns_in_window = 0
918        self._ptp_daemon_warns_suppressed = 0
919
920    def _warn_ptp_daemon_line(self, line: str) -> None:
921        """
922        Warn about a daemon line that reads like a problem, at a bounded rate.
923
924        The markers are broad by design, and "error" is ordinary vocabulary in
925        clock telemetry (offset error, path delay error), so one matching line
926        in the daemon's per-packet trace would fill a user's log with warnings.
927        A burst still gets through - which is what a real one-shot daemon
928        failure looks like - and the rest of the window is counted and reported
929        once instead. Suppressed lines still reach the verbose log.
930
931        :param line: The daemon output line that matched a problem marker.
932        """
933        now = time.monotonic()
934        window_start = self._ptp_daemon_warn_window_start
935        # None means no window is open yet, which is not the same as one that
936        # started at monotonic zero: time.monotonic() counts from boot on Linux,
937        # so a server started in the first minute of uptime would otherwise get
938        # a first window shorter than the rest.
939        if window_start is None or now - window_start >= PTP_DAEMON_WARN_WINDOW:
940            if self._ptp_daemon_warns_suppressed:
941                self.logger.warning(
942                    "PTP daemon: %d further problem line(s) were suppressed over the last "
943                    "%.0fs; enable verbose logging to see them all",
944                    self._ptp_daemon_warns_suppressed,
945                    PTP_DAEMON_WARN_WINDOW,
946                )
947            self._ptp_daemon_warn_window_start = now
948            self._ptp_daemon_warns_in_window = 0
949            self._ptp_daemon_warns_suppressed = 0
950        if self._ptp_daemon_warns_in_window < PTP_DAEMON_WARN_BURST:
951            self._ptp_daemon_warns_in_window += 1
952            self.logger.warning("PTP daemon: %s", line)
953            return
954        self._ptp_daemon_warns_suppressed += 1
955        self.logger.log(VERBOSE_LOG_LEVEL, "PTP daemon: %s", line)
956
957    async def _ptp_daemon_monitor(self, daemon: AsyncProcess) -> None:
958        """Watch the PTP daemon process and restart it once if it crashes."""
959        returncode = await daemon.wait()
960        if self._ptp_daemon_stop_requested or self._ptp_daemon is not daemon:
961            return
962        self._ptp_daemon = None
963        # Crash detected: drop readiness immediately so a session starting during
964        # the restart window degrades consistently instead of attaching to a dead
965        # clock (a restart re-sets it once the new daemon reports ready).
966        if self._ptp_daemon_ready is not None:
967            self._ptp_daemon_ready.clear()
968        runtime = time.monotonic() - self._ptp_daemon_started
969        if runtime < 5:
970            # immediate exit: UDP 319/320 already taken or missing privileges
971            # (root or CAP_NET_BIND_SERVICE) - streams fall back to their
972            # in-process timing engine, so playback keeps working.
973            self.logger.warning(
974                "PTP clock daemon could not start (exit code %s) - PTP timing is degraded. "
975                "Multi-room sync of native AirPlay 2 players may drift; ensure UDP ports "
976                "319/320 are free and run the server as root or grant cliairplay "
977                "CAP_NET_BIND_SERVICE.",
978                returncode,
979            )
980            return
981        if not self._ptp_daemon_restarted:
982            self._ptp_daemon_restarted = True
983            self.logger.warning(
984                "PTP clock daemon stopped unexpectedly (exit code %s) - restarting", returncode
985            )
986            await self._start_ptp_daemon()
987            return
988        self.logger.warning(
989            "PTP clock daemon stopped again (exit code %s) - giving up. "
990            "PTP timing is degraded until the provider is reloaded.",
991            returncode,
992        )
993
994    async def _handle_dacp_request(  # noqa: PLR0915
995        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
996    ) -> None:
997        """Handle new connection on the socket."""
998        try:
999            raw_request = b""
1000            while recv := await reader.read(1024):
1001                raw_request += recv
1002                if len(recv) < 1024:
1003                    break
1004            if not raw_request:
1005                # Some device (Phorus PS10) seems to send empty request
1006                # Maybe as a ack message? we have nothing to do here with empty request
1007                # so we return early.
1008                return
1009
1010            request = raw_request.decode("UTF-8")
1011            if "\r\n\r\n" in request:
1012                headers_raw, body = request.split("\r\n\r\n", 1)
1013            else:
1014                headers_raw = request
1015                body = ""
1016            headers_split = headers_raw.split("\r\n")
1017            headers = {}
1018            for line in headers_split[1:]:
1019                if ":" not in line:
1020                    continue
1021                x, y = line.split(":", 1)
1022                headers[x.strip()] = y.strip()
1023            active_remote = headers.get("Active-Remote")
1024            _, path, _ = headers_split[0].split(" ")
1025            # lookup airplay player by active remote id
1026            player: AirPlayPlayer | None = next(
1027                (
1028                    x
1029                    for x in self.get_players()
1030                    if x.stream and x.stream.active_remote_id == active_remote
1031                ),
1032                None,
1033            )
1034            self.logger.debug(
1035                "DACP request for %s (%s): %s -- %s",
1036                player.name if player else "UNKNOWN PLAYER",
1037                active_remote,
1038                path,
1039                body,
1040            )
1041            # machine-parseable capture of raw DACP traffic for building replay test fixtures
1042            if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1043                capture = {
1044                    "player": player.name if player else None,
1045                    "active_remote": active_remote,
1046                    "method": headers_split[0].split(" ", 1)[0],
1047                    "path": path,
1048                    "headers": headers,
1049                    "body": body,
1050                    "raw_b64": base64.b64encode(raw_request).decode("ascii"),
1051                }
1052                self.logger.log(VERBOSE_LOG_LEVEL, "AIRPLAY_DACP_CAPTURE %s", json.dumps(capture))
1053            if not player:
1054                return
1055            if player.protocol_parent_id and (
1056                parent := self.mass.players.get_player(player.protocol_parent_id)
1057            ):
1058                parent_player = parent
1059            else:
1060                parent_player = player
1061
1062            player_id = player.player_id
1063            if path == "/ctrl-int/1/nextitem":
1064                self.handle_remote_command(player, AirPlayRemoteCommand.NEXT)
1065            elif path == "/ctrl-int/1/previtem":
1066                self.handle_remote_command(player, AirPlayRemoteCommand.PREVIOUS)
1067            elif path == "/ctrl-int/1/play":
1068                self.handle_remote_command(player, AirPlayRemoteCommand.PLAY)
1069            elif path == "/ctrl-int/1/playpause":
1070                self.handle_remote_command(player, AirPlayRemoteCommand.PLAY_PAUSE)
1071            elif path == "/ctrl-int/1/stop":
1072                self.mass.create_task(self.mass.players.cmd_stop(player_id))
1073            elif path == "/ctrl-int/1/volumeup":
1074                self.mass.create_task(self.mass.players.cmd_volume_up(player_id))
1075            elif path == "/ctrl-int/1/volumedown":
1076                self.mass.create_task(self.mass.players.cmd_volume_down(player_id))
1077            elif path == "/ctrl-int/1/shuffle_songs":
1078                active_queue = self.mass.players.get_active_queue(player)
1079                if not active_queue:
1080                    return
1081                await self.mass.player_queues.set_shuffle(
1082                    active_queue.queue_id, not active_queue.shuffle_enabled
1083                )
1084            elif path == "/ctrl-int/1/pause":
1085                self.handle_remote_command(player, AirPlayRemoteCommand.PAUSE)
1086            elif path == "/ctrl-int/1/discrete-pause":
1087                # Some devices send discrete-pause right before device-prevent-playback=1
1088                # when switching to another source. We debounce the pause to avoid
1089                # unnecessary pause commands that would interfere with source switching
1090                # so we only process the pause command if we don't receive a
1091                # prevent-playback=1 within a short time window.
1092                if player.state.playback_state == PlaybackState.PLAYING:
1093                    self.mass.call_later(
1094                        1.0,
1095                        self.mass.players.cmd_pause,
1096                        player_id,
1097                        task_id=f"debounced_pause_{player_id}",
1098                    )
1099            elif "dmcp.device-volume=" in path and not player.ignore_volume_reports:
1100                # This is a bit annoying as this can be either the device confirming a new volume
1101                # we've sent or the device requesting a new volume itself.
1102                # In case of a small rounding difference, we ignore this,
1103                # to prevent an endless pingpong of volume changes
1104                airplay_volume = float(path.split("dmcp.device-volume=", 1)[-1])
1105                if airplay_volume <= AIRPLAY_VOLUME_MUTE:
1106                    player._attr_volume_muted = True
1107                    if player.stream and player.stream.running:
1108                        self.mass.create_task(player.stream.send_cli_command("VOLUME=0"))
1109                    player.update_state()
1110                else:
1111                    if player.volume_muted:
1112                        player._attr_volume_muted = False
1113                        if player.stream and player.stream.running:
1114                            self.mass.create_task(
1115                                player.stream.send_cli_command(f"VOLUME={player.volume_level or 0}")
1116                            )
1117                    volume = convert_airplay_volume(airplay_volume)
1118                    player.update_volume_from_device(volume)
1119            elif "dmcp.volume=" in path:
1120                # volume change request from device (e.g. volume buttons)
1121                volume = int(path.split("dmcp.volume=", 1)[-1])
1122                player.update_volume_from_device(volume)
1123            elif "device-prevent-playback=1" in path:
1124                # device switched to another source (or is powered off)
1125                # Cancel any pending debounced pause since prevent-playback takes precedence
1126                self.mass.cancel_timer(f"debounced_pause_{player_id}")
1127                # Ignore during stream transition (stale message from old CLI process)
1128                if player._transitioning or not player.stream:
1129                    self.logger.debug("Ignoring prevent-playback during stream transition")
1130                elif player.stream.prevent_playback:
1131                    # Already handling a prevent-playback for this stream
1132                    # (duplicate message while ungroup/stop is still in progress)
1133                    self.logger.debug("Ignoring duplicate prevent-playback for %s", player.name)
1134                elif not player.stream.connected:
1135                    # Some devices (e.g. Denon AVR-X2700H) emit a transient
1136                    # prevent-playback=1/=0 pair during RAOP session setup.
1137                    # A real "device switched off / source switched" event only happens
1138                    # once the stream is actually established, so ignore these.
1139                    self.logger.debug(
1140                        "Ignoring prevent-playback for %s - stream not yet established",
1141                        player.name,
1142                    )
1143                else:
1144                    player.stream.prevent_playback = True
1145                    if player.stream.session:
1146                        self.logger.debug(
1147                            "Prevent playback command detected for player %s",
1148                            player.name,
1149                        )
1150                        if player.synced_to or parent_player.state.active_group:
1151                            self.mass.create_task(
1152                                self.mass.players.cmd_ungroup(parent_player.player_id)
1153                            )
1154                        else:
1155                            self.mass.create_task(player.stream.stop())
1156            elif "device-prevent-playback=0" in path:
1157                # device reports that its ready for playback again
1158                # use a debounced reset to avoid race conditions where a quick
1159                # prevent-playback=0 between duplicate prevent-playback=1 messages
1160                # would reset the flag and allow the second message to act
1161                if (stream := player.stream) and stream.prevent_playback:
1162                    self.mass.call_later(
1163                        5,
1164                        setattr,
1165                        stream,
1166                        "prevent_playback",
1167                        False,
1168                        task_id=f"reset_prevent_playback_{player_id}",
1169                    )
1170
1171            # send response
1172            date_str = utc().strftime("%a, %-d %b %Y %H:%M:%S")
1173            response = (
1174                f"HTTP/1.0 204 No Content\r\nDate: {date_str} "
1175                "GMT\r\nDAAP-Server: iTunes/7.6.2 (Windows; N;)\r\nContent-Type: "
1176                "application/x-dmap-tagged\r\nContent-Length: 0\r\n"
1177                "Connection: close\r\n\r\n"
1178            )
1179            writer.write(response.encode())
1180            await writer.drain()
1181        finally:
1182            writer.close()
1183            with suppress(Exception):
1184                await writer.wait_closed()
1185
1186    def _drop_unverified_password_markers(self) -> None:
1187        """
1188        Clear the stored "password rejected" verdicts left by earlier releases, once.
1189
1190        Those releases marked a player whenever the binary reported an auth-shaped
1191        rejection, without separating a password challenge from a device that
1192        refused the handshake outright. The refusals put players that have no
1193        password at all into a setup flow only a password could leave, so the
1194        verdicts are dropped and left to be earned again on the next connect.
1195        """
1196        if self.mass.config.get_raw_provider_config_value(
1197            self.instance_id, CONF_PASSWORD_MARKERS_REVIEWED, False
1198        ):
1199            return
1200        # walks the stored configs rather than get_player_configs(), which drops
1201        # every protocol player - the type each non-Apple receiver is registered
1202        # as - and only lists the ones discovered so far
1203        for player_id, raw_conf in self.mass.config.get(CONF_PLAYERS, {}).items():
1204            if not isinstance(raw_conf, dict) or raw_conf.get("provider") != self.instance_id:
1205                continue
1206            if not self.mass.config.get_raw_player_config_value(
1207                player_id, CONF_PASSWORD_INVALID, False
1208            ):
1209                continue
1210            self.logger.info("Clearing the unverified password marker on %s", player_id)
1211            self.mass.config.set_raw_player_config_value(player_id, CONF_PASSWORD_INVALID, False)
1212        # last, so a failure part-way through leaves the review to be retried on
1213        # the next load instead of stranding the players it never reached
1214        self.mass.config.set_raw_provider_config_value(
1215            self.instance_id, CONF_PASSWORD_MARKERS_REVIEWED, True
1216        )
1217