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