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