music-assistant-server

1"""AirPlay Player provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import socket
7from typing import cast
8
9from music_assistant_models.enums import PlaybackState
10from zeroconf import ServiceStateChange
11from zeroconf.asyncio import AsyncServiceInfo
12
13from music_assistant.helpers.datetime import utc
14from music_assistant.helpers.util import (
15    get_ip_pton,
16    get_primary_ip_address_from_zeroconf,
17    select_free_port,
18)
19from music_assistant.models.player_provider import PlayerProvider
20
21from .constants import (
22    AIRPLAY_DISCOVERY_TYPE,
23    CACHE_CATEGORY_PREV_VOLUME,
24    CONF_IGNORE_VOLUME,
25    DACP_DISCOVERY_TYPE,
26    FALLBACK_VOLUME,
27    RAOP_DISCOVERY_TYPE,
28)
29from .helpers import convert_airplay_volume, get_model_info
30from .player import AirPlayPlayer
31
32# TODO: AirPlay provider
33# Implement Companion protocol for communicating with original Apple (TV) devices
34# This allows for getting state/metadata changes from the device,
35# even if we are not actively streaming to it.
36
37
38class AirPlayProvider(PlayerProvider):
39    """Player provider for AirPlay based players."""
40
41    _dacp_server: asyncio.Server
42    _dacp_info: AsyncServiceInfo
43
44    async def handle_async_init(self) -> None:
45        """Handle async initialization of the provider."""
46        # register DACP zeroconf service
47        dacp_port = await select_free_port(39831, 49831)
48        # Use first 16 hex chars of server_id as a persistent DACP ID
49        # This ensures the DACP ID remains the same across restarts, which is required
50        # for AirPlay 2 (HAP) pair-verify to work with previously paired devices
51        self.dacp_id = dacp_id = self.mass.server_id[:16].upper()
52        self.logger.debug("Starting DACP ActiveRemote %s on port %s", dacp_id, dacp_port)
53        self._dacp_server = await asyncio.start_server(self._handle_dacp_request, port=dacp_port)
54        server_id = f"iTunes_Ctrl_{dacp_id}.{DACP_DISCOVERY_TYPE}"
55        self._dacp_info = AsyncServiceInfo(
56            DACP_DISCOVERY_TYPE,
57            name=server_id,
58            addresses=[await get_ip_pton(str(self.mass.streams.publish_ip))],
59            port=dacp_port,
60            properties={
61                "txtvers": "1",
62                "Ver": "63B5E5C0C201542E",
63                "DbId": "63B5E5C0C201542E",
64                "OSsi": "0x1F5",
65            },
66            server=f"{socket.gethostname()}.local",
67        )
68        await self.mass.aiozc.async_register_service(self._dacp_info)
69
70    async def on_mdns_service_state_change(
71        self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
72    ) -> None:
73        """Handle MDNS service state callback."""
74        if not info:
75            if state_change == ServiceStateChange.Removed and "@" in name:
76                # Service name is enough to mark the player as unavailable on 'Removed' notification
77                raw_id, display_name = name.split(".")[0].split("@", 1)
78            else:
79                # If we are not in a 'Removed' state, we need info to be filled to update the player
80                return
81        elif "@" in info.name:
82            raw_id, display_name = info.name.split(".")[0].split("@", 1)
83        elif deviceid := info.decoded_properties.get("deviceid"):
84            raw_id = deviceid.replace(":", "")
85            display_name = info.name.split(".")[0]
86        else:
87            return
88        player_id = f"ap{raw_id.lower()}"
89        # handle removed player
90        if state_change == ServiceStateChange.Removed:
91            if _player := self.mass.players.get_player(player_id):
92                # the player has become unavailable
93                self.logger.debug("Player offline: %s", _player.display_name)
94                await self.mass.players.unregister(player_id)
95            return
96        # handle update for existing device
97        assert info is not None  # type guard
98        player: AirPlayPlayer | None
99        if player := cast("AirPlayPlayer | None", self.mass.players.get_player(player_id)):
100            # update the latest discovery info for existing player
101            player.set_discovery_info(info, display_name)
102            return
103        await self._setup_player(player_id, display_name, info)
104
105    async def unload(self, is_removed: bool = False) -> None:
106        """Handle unload/close of the provider."""
107        # shutdown DACP server
108        if self._dacp_server:
109            self._dacp_server.close()
110        # shutdown DACP zeroconf service
111        if self._dacp_info:
112            await self.mass.aiozc.async_unregister_service(self._dacp_info)
113
114    async def _setup_player(
115        self, player_id: str, display_name: str, discovery_info: AsyncServiceInfo
116    ) -> None:
117        """Handle setup of a new player that is discovered using mdns."""
118        raop_discovery_info: AsyncServiceInfo | None = None
119        airplay_discovery_info: AsyncServiceInfo | None = None
120        if discovery_info.type == RAOP_DISCOVERY_TYPE:
121            # RAOP service discovered
122            raop_discovery_info = discovery_info
123            self.logger.debug("Discovered RAOP service for %s", display_name)
124            # always prefer airplay mdns info as it has more details
125            # fallback to raop info if airplay info is not available,
126            # (old device only announcing raop)
127            airplay_discovery_info = AsyncServiceInfo(
128                AIRPLAY_DISCOVERY_TYPE,
129                discovery_info.name.split("@")[-1].replace("_raop", "_airplay"),
130            )
131            await airplay_discovery_info.async_request(self.mass.aiozc.zeroconf, 3000)
132        else:
133            # AirPlay service discovered
134            self.logger.debug("Discovered AirPlay service for %s", display_name)
135            airplay_discovery_info = discovery_info
136
137        if airplay_discovery_info:
138            manufacturer, model = get_model_info(airplay_discovery_info)
139        elif raop_discovery_info:
140            manufacturer, model = get_model_info(raop_discovery_info)
141        else:
142            manufacturer, model = "Unknown", "Unknown"
143
144        address = get_primary_ip_address_from_zeroconf(discovery_info)
145        if not address:
146            return  # should not happen, but guard just in case
147
148        # Filter out shairport-sync instances running on THIS Music Assistant server
149        # These are managed by the AirPlay Receiver provider, not the AirPlay provider
150        # We check both model name AND that it's a local address to avoid filtering
151        # shairport-sync instances running on other machines
152        if model == "ShairportSync":
153            # Check if this is a local address (127.x.x.x or matches our server's IP)
154            if address.startswith("127.") or address == self.mass.streams.publish_ip:
155                return
156
157        if not self.mass.config.get_raw_player_config_value(player_id, "enabled", True):
158            self.logger.debug("Ignoring %s in discovery as it is disabled.", display_name)
159            return
160        if not discovery_info:
161            return  # should not happen, but guard just in case
162
163        # if we reach this point, all preflights are ok and we can create the player
164        self.logger.debug("Discovered AirPlay device %s on %s", display_name, address)
165
166        # Get volume from cache
167        if not (
168            volume := await self.mass.cache.get(
169                key=player_id, provider=self.instance_id, category=CACHE_CATEGORY_PREV_VOLUME
170            )
171        ):
172            volume = FALLBACK_VOLUME
173
174        # Final check before registration to handle race conditions
175        # (multiple MDNS events processed in parallel for same device)
176        if self.mass.players.get_player(player_id):
177            self.logger.debug(
178                "Player %s already registered during setup, skipping registration", player_id
179            )
180            return
181
182        self.logger.debug(
183            "Setting up player %s: manufacturer=%s, model=%s",
184            display_name,
185            manufacturer,
186            model,
187        )
188
189        # Create single AirPlayPlayer for all devices
190        # Pairing config entries will be shown conditionally based on device type
191        player = AirPlayPlayer(
192            provider=self,
193            player_id=player_id,
194            raop_discovery_info=raop_discovery_info,
195            airplay_discovery_info=airplay_discovery_info,
196            address=address,
197            display_name=display_name,
198            manufacturer=manufacturer,
199            model=model,
200            initial_volume=volume,
201        )
202        await self.mass.players.register(player)
203
204    async def _handle_dacp_request(  # noqa: PLR0915
205        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
206    ) -> None:
207        """Handle new connection on the socket."""
208        try:
209            raw_request = b""
210            while recv := await reader.read(1024):
211                raw_request += recv
212                if len(recv) < 1024:
213                    break
214            if not raw_request:
215                # Some device (Phorus PS10) seems to send empty request
216                # Maybe as a ack message? we have nothing to do here with empty request
217                # so we return early.
218                return
219
220            request = raw_request.decode("UTF-8")
221            if "\r\n\r\n" in request:
222                headers_raw, body = request.split("\r\n\r\n", 1)
223            else:
224                headers_raw = request
225                body = ""
226            headers_split = headers_raw.split("\r\n")
227            headers = {}
228            for line in headers_split[1:]:
229                if ":" not in line:
230                    continue
231                x, y = line.split(":", 1)
232                headers[x.strip()] = y.strip()
233            active_remote = headers.get("Active-Remote")
234            _, path, _ = headers_split[0].split(" ")
235            # lookup airplay player by active remote id
236            player: AirPlayPlayer | None = next(
237                (
238                    x
239                    for x in self.get_players()
240                    if x.stream and x.stream.active_remote_id == active_remote
241                ),
242                None,
243            )
244            self.logger.debug(
245                "DACP request for %s (%s): %s -- %s",
246                player.name if player else "UNKNOWN PLAYER",
247                active_remote,
248                path,
249                body,
250            )
251            if not player:
252                return
253
254            player_id = player.player_id
255            ignore_volume_report = (
256                self.mass.config.get_raw_player_config_value(player_id, CONF_IGNORE_VOLUME, False)
257                or player.device_info.manufacturer.lower() == "apple"
258            )
259            if path == "/ctrl-int/1/nextitem":
260                self.mass.create_task(self.mass.players.cmd_next_track(player_id))
261            elif path == "/ctrl-int/1/previtem":
262                self.mass.create_task(self.mass.players.cmd_previous_track(player_id))
263            elif path == "/ctrl-int/1/play":
264                # sometimes this request is sent by a device as confirmation of a play command
265                # we ignore this if the player is already playing
266                if player.playback_state != PlaybackState.PLAYING:
267                    self.mass.create_task(self.mass.players.cmd_play(player_id))
268            elif path == "/ctrl-int/1/playpause":
269                self.mass.create_task(self.mass.players.cmd_play_pause(player_id))
270            elif path == "/ctrl-int/1/stop":
271                self.mass.create_task(self.mass.players.cmd_stop(player_id))
272            elif path == "/ctrl-int/1/volumeup":
273                self.mass.create_task(self.mass.players.cmd_volume_up(player_id))
274            elif path == "/ctrl-int/1/volumedown":
275                self.mass.create_task(self.mass.players.cmd_volume_down(player_id))
276            elif path == "/ctrl-int/1/shuffle_songs":
277                active_queue = self.mass.players.get_active_queue(player)
278                if not active_queue:
279                    return
280                await self.mass.player_queues.set_shuffle(
281                    active_queue.queue_id, not active_queue.shuffle_enabled
282                )
283            elif path == "/ctrl-int/1/pause":
284                if player.state.playback_state == PlaybackState.PLAYING:
285                    self.mass.create_task(self.mass.players.cmd_pause(player_id))
286            elif path == "/ctrl-int/1/discrete-pause":
287                # Some devices send discrete-pause right before device-prevent-playback=1
288                # when switching to another source. We debounce the pause to avoid
289                # unnecessary pause commands that would interfere with source switching
290                # so we only process the pause command if we don't receive a
291                # prevent-playback=1 within a short time window.
292                if player.state.playback_state == PlaybackState.PLAYING:
293                    self.mass.call_later(
294                        1.0,
295                        self.mass.players.cmd_pause,
296                        player_id,
297                        task_id=f"debounced_pause_{player_id}",
298                    )
299            elif "dmcp.device-volume=" in path and not ignore_volume_report:
300                # This is a bit annoying as this can be either the device confirming a new volume
301                # we've sent or the device requesting a new volume itself.
302                # In case of a small rounding difference, we ignore this,
303                # to prevent an endless pingpong of volume changes
304                airplay_volume = float(path.split("dmcp.device-volume=", 1)[-1])
305                volume = convert_airplay_volume(airplay_volume)
306                player.update_volume_from_device(volume)
307            elif "dmcp.volume=" in path:
308                # volume change request from device (e.g. volume buttons)
309                volume = int(path.split("dmcp.volume=", 1)[-1])
310                player.update_volume_from_device(volume)
311            elif "device-prevent-playback=1" in path:
312                # device switched to another source (or is powered off)
313                # Cancel any pending debounced pause since prevent-playback takes precedence
314                self.mass.cancel_timer(f"debounced_pause_{player_id}")
315                # Ignore during stream transition (stale message from old CLI process)
316                if player._transitioning or not player.stream:
317                    self.logger.debug("Ignoring prevent-playback during stream transition")
318                else:
319                    player.stream.prevent_playback = True
320                    if player.stream.session:
321                        self.logger.debug(
322                            "Prevent playback command detected for player %s",
323                            player.name,
324                        )
325                        if player.state.synced_to or player.state.group_members:
326                            self.mass.create_task(self.mass.players.cmd_ungroup(player_id))
327                        else:
328                            self.mass.create_task(player.stream.session.stop())
329            elif "device-prevent-playback=0" in path:
330                # device reports that its ready for playback again
331                if stream := player.stream:
332                    stream.prevent_playback = False
333
334            # send response
335            date_str = utc().strftime("%a, %-d %b %Y %H:%M:%S")
336            response = (
337                f"HTTP/1.0 204 No Content\r\nDate: {date_str} "
338                "GMT\r\nDAAP-Server: iTunes/7.6.2 (Windows; N;)\r\nContent-Type: "
339                "application/x-dmap-tagged\r\nContent-Length: 0\r\n"
340                "Connection: close\r\n\r\n"
341            )
342            writer.write(response.encode())
343            await writer.drain()
344        finally:
345            writer.close()
346
347    def get_players(self) -> list[AirPlayPlayer]:
348        """Return all airplay players belonging to this instance."""
349        return cast("list[AirPlayPlayer]", self.players)
350
351    def get_player(self, player_id: str) -> AirPlayPlayer | None:
352        """Return AirplayPlayer by id."""
353        return cast("AirPlayPlayer | None", self.mass.players.get_player(player_id))
354