/
/
/
1"""
2Sonos Player provider for Music Assistant for speakers running the S2 firmware.
3
4Based on the aiosonos library, which leverages the new websockets API of the Sonos S2 firmware.
5https://github.com/music-assistant/aiosonos
6"""
7
8from __future__ import annotations
9
10from typing import TYPE_CHECKING, Any
11
12from aiohttp import web
13from aiohttp.client_exceptions import ClientError
14from aiosonos.api.models import SonosCapability
15from aiosonos.utils import get_discovery_info
16from music_assistant_models.enums import PlaybackState
17from zeroconf import ServiceStateChange
18
19from music_assistant.constants import (
20 CONF_ENTRY_MANUAL_DISCOVERY_IPS,
21 MASS_LOGO_ONLINE,
22 VERBOSE_LOG_LEVEL,
23)
24from music_assistant.models.player_provider import PlayerProvider
25
26from .helpers import get_primary_ip_address
27from .player import SonosPlayer
28
29if TYPE_CHECKING:
30 from music_assistant_models.config_entries import PlayerConfig
31 from music_assistant_models.player import PlayerMedia
32 from zeroconf.asyncio import AsyncServiceInfo
33
34
35class SonosPlayerProvider(PlayerProvider):
36 """Sonos Player provider."""
37
38 async def handle_async_init(self) -> None:
39 """Handle async initialization of the provider."""
40 self.mass.streams.register_dynamic_route(
41 "/sonos_queue/*", self._handle_sonos_cloud_queue_request
42 )
43
44 async def loaded_in_mass(self) -> None:
45 """Call after the provider has been loaded."""
46 await super().loaded_in_mass()
47 # Handle config option for manual IP's
48 manual_ip_config: list[str] = self.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key)
49 for ip_address in manual_ip_config:
50 try:
51 # get discovery info from SONOS speaker so we can provide an ID & other info
52 discovery_info = await get_discovery_info(self.mass.http_session_no_ssl, ip_address)
53 except ClientError as err:
54 self.logger.debug(
55 "Ignoring %s (manual IP) as it is not reachable: %s", ip_address, str(err)
56 )
57 continue
58 player_id = discovery_info["device"]["id"]
59 sonos_player = SonosPlayer(self, player_id, discovery_info=discovery_info)
60 sonos_player.device_info.ip_address = ip_address
61 await sonos_player.setup()
62
63 async def unload(self, is_removed: bool = False) -> None:
64 """Handle close/cleanup of the provider."""
65 self.mass.streams.unregister_dynamic_route("/sonos_queue/*")
66
67 async def on_mdns_service_state_change(
68 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
69 ) -> None:
70 """Handle MDNS service state callback."""
71 if state_change == ServiceStateChange.Removed:
72 # we don't listen for removed players here.
73 # instead we just wait for the player connection to fail
74 return
75 if "uuid" not in info.decoded_properties:
76 # not a S2 player
77 return
78 name = name.split("@", 1)[1] if "@" in name else name
79 player_id = info.decoded_properties["uuid"]
80 # handle update for existing device
81 if sonos_player := self.mass.players.get(player_id):
82 assert isinstance(sonos_player, SonosPlayer), (
83 "Player ID already exists but is not a SonosPlayer"
84 )
85 # if mass_player := sonos_player.mass_player:
86 cur_address = get_primary_ip_address(info)
87 if cur_address and cur_address != sonos_player.device_info.ip_address:
88 sonos_player.logger.debug(
89 "Address updated from %s to %s",
90 sonos_player.device_info.ip_address,
91 cur_address,
92 )
93 sonos_player.device_info.ip_address = cur_address
94 if not sonos_player.connected:
95 self.logger.debug("Player back online: %s", sonos_player.display_name)
96 sonos_player.client.player_ip = cur_address
97 # schedule reconnect
98 sonos_player.reconnect()
99 self.mass.players.trigger_player_update(player_id)
100 return
101 # handle new player setup in a delayed task because mdns announcements
102 # can arrive in (duplicated) bursts
103 task_id = f"setup_sonos_{player_id}"
104 self.mass.call_later(5, self._setup_player, player_id, name, info, task_id=task_id)
105
106 async def on_player_config_change(self, config: PlayerConfig, changed_keys: set[str]) -> None:
107 """Call (by config manager) when the configuration of a player changes."""
108 await super().on_player_config_change(config, changed_keys)
109 if "values/airplay_mode" in changed_keys and (
110 (sonos_player := self.mass.players.get(config.player_id))
111 and (airplay_player := sonos_player.get_linked_airplay_player(False))
112 and airplay_player.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
113 ):
114 # edge case: we switched from airplay mode to sonos mode (or vice versa)
115 # we need to make sure that playback gets stopped on the airplay player
116 await airplay_player.stop()
117 # We also need to run setup again on the Sonos player to ensure the supported
118 # features are updated.
119 await sonos_player.setup()
120
121 async def _setup_player(self, player_id: str, name: str, info: AsyncServiceInfo) -> None:
122 """Handle setup of a new player that is discovered using mdns."""
123 assert not self.mass.players.get(player_id)
124 address = get_primary_ip_address(info)
125 if address is None:
126 return
127 if not self.mass.config.get_raw_player_config_value(player_id, "enabled", True):
128 self.logger.debug("Ignoring %s in discovery as it is disabled.", name)
129 return
130 try:
131 discovery_info = await get_discovery_info(self.mass.http_session_no_ssl, address)
132 except ClientError as err:
133 self.logger.debug("Ignoring %s in discovery as it is not reachable: %s", name, str(err))
134 return
135 display_name = discovery_info["device"].get("name") or name
136 if SonosCapability.PLAYBACK not in discovery_info["device"]["capabilities"]:
137 # this will happen for satellite speakers in a surround/stereo setup
138 self.logger.debug(
139 "Ignoring %s in discovery as it is a passive satellite.", display_name
140 )
141 return
142 self.logger.debug("Discovered Sonos device %s on %s", name, address)
143 sonos_player = SonosPlayer(self, player_id, discovery_info=discovery_info)
144 sonos_player.device_info.ip_address = address
145 await sonos_player.setup()
146
147 async def _handle_sonos_cloud_queue_request(self, request: web.Request) -> web.Response:
148 """
149 Handle the Sonos CloudQueue request.
150
151 https://docs.sonos.com/reference/itemwindow
152 """
153 self.logger.log(
154 VERBOSE_LOG_LEVEL,
155 "Cloud Queue request\n - path: %s\n - query: %s\n",
156 request.path,
157 request.query,
158 )
159 path_parts = request.path.strip("/").split("/")
160 if len(path_parts) != 4 or path_parts[0] != "sonos_queue":
161 return web.Response(status=404)
162 player_id = path_parts[1]
163 if not (sonos_player := self.mass.players.get_player(player_id)):
164 return web.Response(status=501)
165 if TYPE_CHECKING:
166 assert isinstance(sonos_player, SonosPlayer)
167 endpoint = path_parts[3]
168 if endpoint == "itemWindow":
169 return await self._handle_sonos_queue_itemwindow(sonos_player, request)
170 if endpoint == "version":
171 return await self._handle_sonos_queue_version(sonos_player, request)
172 if endpoint == "context":
173 return await self._handle_sonos_queue_context(sonos_player, request)
174 if endpoint == "timePlayed":
175 return await self._handle_sonos_queue_time_played(sonos_player, request)
176 return web.Response(status=404)
177
178 async def _handle_sonos_queue_itemwindow(
179 self, player: SonosPlayer, request: web.Request
180 ) -> web.Response:
181 """
182 Handle the Sonos CloudQueue ItemWindow endpoint.
183
184 https://docs.sonos.com/reference/itemwindow
185 """
186 context_version = request.query.get("contextVersion", "1")
187 queue_version = request.query.get("queueVersion", str(int(player.sonos_queue.last_updated)))
188 # because Sonos does not show our queue in the app anyways,
189 # we just return the previous, current and next item in the queue
190 items = list(player.sonos_queue.items)
191 result = {
192 "includesBeginningOfQueue": False,
193 "includesEndOfQueue": False,
194 "contextVersion": context_version,
195 "queueVersion": queue_version,
196 "items": [self._parse_sonos_queue_item(x) for x in items],
197 }
198 return web.json_response(result)
199
200 async def _handle_sonos_queue_version(
201 self, player: SonosPlayer, request: web.Request
202 ) -> web.Response:
203 """
204 Handle the Sonos CloudQueue Version endpoint.
205
206 https://docs.sonos.com/reference/version
207 """
208 context_version = request.query.get("contextVersion") or "1"
209 result = {
210 "contextVersion": context_version,
211 "queueVersion": str(int(player.sonos_queue.last_updated)),
212 }
213 return web.json_response(result)
214
215 async def _handle_sonos_queue_context(
216 self, player: SonosPlayer, request: web.Request
217 ) -> web.Response:
218 """
219 Handle the Sonos CloudQueue Context endpoint.
220
221 https://docs.sonos.com/reference/context
222 """
223 result = {
224 "contextVersion": "1",
225 "queueVersion": str(int(player.sonos_queue.last_updated)),
226 "container": {
227 "type": "trackList",
228 "name": "Music Assistant",
229 "imageUrl": MASS_LOGO_ONLINE,
230 "service": {"name": "Music Assistant", "id": "mass"},
231 "id": {
232 "serviceId": "mass",
233 "objectId": f"mass:{player.sonos_queue.items[-1].source_id}"
234 if player.sonos_queue.items
235 else "mass:unknown",
236 "accountId": "",
237 },
238 },
239 "reports": {
240 "sendUpdateAfterMillis": 1000,
241 "periodicIntervalMillis": 30000,
242 "sendPlaybackActions": True,
243 },
244 "playbackPolicies": {
245 "canSkip": True,
246 "limitedSkips": True,
247 "canSkipToItem": True, # unsure
248 "canSkipBack": True,
249 # seek needs to be disabled because we dont properly support range requests
250 "canSeek": False,
251 "canRepeat": False, # handled by MA queue controller
252 "canRepeatOne": False, # synced from MA queue controller
253 "canCrossfade": False, # handled by MA queue controller
254 "canShuffle": False, # handled by MA queue controller
255 },
256 }
257 return web.json_response(result)
258
259 async def _handle_sonos_queue_time_played(
260 self, player: SonosPlayer, request: web.Request
261 ) -> web.Response:
262 """
263 Handle the Sonos CloudQueue TimePlayed endpoint.
264
265 https://docs.sonos.com/reference/timeplayed
266 """
267 json_body = await request.json()
268 for item in json_body["items"]:
269 if item["type"] != "update":
270 continue
271 if "positionMillis" not in item:
272 continue
273 if player.current_media and player.current_media.queue_item_id == item["id"]:
274 player.update_elapsed_time(item["positionMillis"] / 1000)
275 break
276 return web.Response(status=204)
277
278 def _parse_sonos_queue_item(self, media: PlayerMedia) -> dict[str, Any]:
279 """Parse MusicAssistant PlayerMedia to a Sonos Media (queue) object."""
280 return {
281 "id": media.queue_item_id or media.uri,
282 "track": {
283 "type": "track",
284 "mediaUrl": media.uri,
285 "contentType": f"audio/{media.uri.split('.')[-1]}",
286 "service": {"name": "Music Assistant", "id": "mass"},
287 "name": media.title,
288 "imageUrl": media.image_url,
289 "durationMillis": int(media.duration * 1000) if media.duration else 0,
290 "artist": {
291 "name": media.artist,
292 }
293 if media.artist
294 else None,
295 "album": {
296 "name": media.album,
297 }
298 if media.album
299 else None,
300 },
301 }
302