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