music-assistant-server
15.1 KB•PY
provider.py
15.1 KB • 342 lines • python
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 active_queue = self.mass.players.get_active_queue(player)
260 if not active_queue:
261 self.logger.warning(
262 "DACP request for %s (%s) but no active queue found, ignoring request",
263 player.display_name,
264 player_id,
265 )
266 return
267 if path == "/ctrl-int/1/nextitem":
268 self.mass.create_task(self.mass.player_queues.next(active_queue.queue_id))
269 elif path == "/ctrl-int/1/previtem":
270 self.mass.create_task(self.mass.player_queues.previous(active_queue.queue_id))
271 elif path == "/ctrl-int/1/play":
272 # sometimes this request is sent by a device as confirmation of a play command
273 # we ignore this if the player is already playing
274 if player.playback_state != PlaybackState.PLAYING:
275 self.mass.create_task(self.mass.player_queues.play(active_queue.queue_id))
276 elif path == "/ctrl-int/1/playpause":
277 self.mass.create_task(self.mass.player_queues.play_pause(active_queue.queue_id))
278 elif path == "/ctrl-int/1/stop":
279 self.mass.create_task(self.mass.player_queues.stop(active_queue.queue_id))
280 elif path == "/ctrl-int/1/volumeup":
281 self.mass.create_task(self.mass.players.cmd_volume_up(player_id))
282 elif path == "/ctrl-int/1/volumedown":
283 self.mass.create_task(self.mass.players.cmd_volume_down(player_id))
284 elif path == "/ctrl-int/1/shuffle_songs":
285 queue = self.mass.player_queues.get(player_id)
286 if not queue:
287 return
288 await self.mass.player_queues.set_shuffle(
289 active_queue.queue_id, not queue.shuffle_enabled
290 )
291 elif path in ("/ctrl-int/1/pause", "/ctrl-int/1/discrete-pause"):
292 # sometimes this request is sent by a device as confirmation of a play command
293 # we ignore this if the player is already playing
294 if player.playback_state == PlaybackState.PLAYING:
295 self.mass.create_task(self.mass.player_queues.pause(active_queue.queue_id))
296 elif "dmcp.device-volume=" in path and not ignore_volume_report:
297 # This is a bit annoying as this can be either the device confirming a new volume
298 # we've sent or the device requesting a new volume itself.
299 # In case of a small rounding difference, we ignore this,
300 # to prevent an endless pingpong of volume changes
301 airplay_volume = float(path.split("dmcp.device-volume=", 1)[-1])
302 volume = convert_airplay_volume(airplay_volume)
303 player.update_volume_from_device(volume)
304 elif "dmcp.volume=" in path:
305 # volume change request from device (e.g. volume buttons)
306 volume = int(path.split("dmcp.volume=", 1)[-1])
307 player.update_volume_from_device(volume)
308 elif "device-prevent-playback=1" in path:
309 # device switched to another source (or is powered off)
310 # Ignore during stream transition (stale message from old CLI process)
311 if player._transitioning:
312 self.logger.debug("Ignoring prevent-playback during stream transition")
313 elif stream := player.stream:
314 stream.prevent_playback = True
315 if stream.session:
316 self.mass.create_task(stream.session.remove_client(player))
317 elif "device-prevent-playback=0" in path:
318 # device reports that its ready for playback again
319 if stream := player.stream:
320 stream.prevent_playback = False
321
322 # send response
323 date_str = utc().strftime("%a, %-d %b %Y %H:%M:%S")
324 response = (
325 f"HTTP/1.0 204 No Content\r\nDate: {date_str} "
326 "GMT\r\nDAAP-Server: iTunes/7.6.2 (Windows; N;)\r\nContent-Type: "
327 "application/x-dmap-tagged\r\nContent-Length: 0\r\n"
328 "Connection: close\r\n\r\n"
329 )
330 writer.write(response.encode())
331 await writer.drain()
332 finally:
333 writer.close()
334
335 def get_players(self) -> list[AirPlayPlayer]:
336 """Return all airplay players belonging to this instance."""
337 return cast("list[AirPlayPlayer]", self.players)
338
339 def get_player(self, player_id: str) -> AirPlayPlayer | None:
340 """Return AirplayPlayer by id."""
341 return cast("AirPlayPlayer | None", self.mass.players.get_player(player_id))
342