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