/
/
/
1"""
2Sonos Player provider for Music Assistant: SonosPlayer object/model.
3
4Note that large parts of this code are copied over from the Home Assistant
5integration for Sonos.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import contextlib
12import logging
13import time
14from collections.abc import Callable, Coroutine
15from typing import TYPE_CHECKING, Any, cast
16
17from music_assistant_models.enums import (
18 IdentifierType,
19 MediaType,
20 PlaybackState,
21 PlayerFeature,
22 PlayerState,
23)
24from music_assistant_models.errors import PlayerCommandFailed
25from soco import SoCoException
26from soco.core import MUSIC_SRC_RADIO, SoCo
27from soco.data_structures import DidlAudioBroadcast
28
29from music_assistant.constants import (
30 CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED,
31 VERBOSE_LOG_LEVEL,
32)
33from music_assistant.helpers.upnp import create_didl_metadata
34from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
35
36from .constants import (
37 AVAILABILITY_TIMEOUT,
38 COMMAND_POLL_DELAY,
39 DURATION_SECONDS,
40 LINEIN_SOURCE_IDS,
41 LINEIN_SOURCE_MAPPING,
42 NEVER_TIME,
43 PLAYER_FEATURES,
44 PLAYER_SOURCE_MAP,
45 POLL_INTERVAL,
46 POSITION_SECONDS,
47 RESUB_COOLDOWN_SECONDS,
48 SONOS_STATE_TRANSITIONING,
49 SOURCE_LINEIN,
50 SOURCE_TV,
51 SUBSCRIPTION_SERVICES,
52 SUBSCRIPTION_TIMEOUT,
53 TRANSITION_POLL_INTERVAL,
54)
55from .helpers import SonosUpdateError, soco_error
56
57if TYPE_CHECKING:
58 from music_assistant_models.config_entries import ConfigEntry
59 from soco.events_base import Event as SonosEvent
60 from soco.events_base import SubscriptionBase
61
62 from .provider import SonosPlayerProvider
63
64CALLBACK_TYPE = Callable[[], None]
65LOGGER = logging.getLogger(__name__)
66
67
68class SonosSubscriptionsFailed(PlayerCommandFailed):
69 """Subscription creation failed."""
70
71
72class SonosPlayer(Player):
73 """Sonos Player implementation for S1 speakers."""
74
75 def __init__(
76 self,
77 provider: SonosPlayerProvider,
78 soco: SoCo,
79 fixed_volume: bool,
80 ) -> None:
81 """
82 Initialize SonosPlayer instance.
83
84 :param fixed_volume: Whether the speaker is set to fixed volume output.
85 """
86 super().__init__(provider, soco.uid)
87 self.soco = soco
88 self.household_id: str = soco.household_id
89
90 # Set player attributes
91 self._attr_supported_features = set(PLAYER_FEATURES)
92 # a speaker playing out at a fixed level (a Connect or Port wired into an amplifier)
93 # rejects volume commands, so it is left without volume and mute control at all
94 if not fixed_volume:
95 self._attr_supported_features |= {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
96 # S1 hardware is fixed to 16-bit at 44.1/48 kHz
97 self._attr_supported_sample_rates = [(44100, 16), (48000, 16)]
98 self._attr_name = soco.player_name
99 self._attr_device_info = DeviceInfo(
100 model=soco.speaker_info["model_name"],
101 manufacturer="Sonos",
102 )
103 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, soco.ip_address)
104 self._attr_device_info.add_identifier(IdentifierType.UUID, soco.uid)
105 mac_address = self._extract_mac_from_player_id()
106 if mac_address:
107 self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
108 self._attr_needs_poll = True
109 self._attr_poll_interval = POLL_INTERVAL
110 self._attr_available = True
111 self._attr_can_group_with = {provider.instance_id}
112
113 # Subscriptions and events
114 self._subscriptions: list[SubscriptionBase] = []
115 self._subscription_lock: asyncio.Lock = asyncio.Lock()
116 self._last_activity: float = NEVER_TIME
117 self._resub_cooldown_expires_at: float | None = None
118 self._poll_task_id: str = f"sonos_poll_{self.player_id}"
119 self._unloaded: bool = False
120
121 @property
122 def missing_subscriptions(self) -> set[str]:
123 """Return a list of missing service subscriptions."""
124 subscribed_services = {sub.service.service_type for sub in self._subscriptions}
125 return SUBSCRIPTION_SERVICES - subscribed_services
126
127 async def setup(self) -> None:
128 """Set up the player."""
129
130 def _read_speaker_state() -> None:
131 """Read the initial state from the speaker (NOT async friendly)."""
132 self._attr_volume_level = self.soco.volume
133 self._attr_volume_muted = self.soco.mute
134 self.update_groups()
135 if not self.synced_to:
136 self.poll_media()
137
138 await asyncio.to_thread(_read_speaker_state)
139 await self.subscribe()
140 await self.mass.players.register_or_update(self)
141
142 async def get_config_entries(self) -> list[ConfigEntry]:
143 """Return all provider-specific configuration entries for the player."""
144 return [CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED]
145
146 async def offline(self) -> None:
147 """Handle removal of speaker when unavailable."""
148 async with self._subscription_lock:
149 await self._offline()
150
151 async def on_unload(self) -> None:
152 """Handle logic when the player is unloaded from the Player controller."""
153 await super().on_unload()
154 # a poll already running in its worker thread cannot be interrupted, so the flag
155 # is what keeps its results from reaching a player the controller no longer has
156 self._unloaded = True
157 # cancel_timer only reaches a poll that is still pending: once the timer fired,
158 # the poll runs as a task under the same id and cancel_task is what stops it
159 self.mass.cancel_timer(self._poll_task_id)
160 self.mass.cancel_task(self._poll_task_id)
161 # unsubscribe directly: offline() skips a speaker that is already marked
162 # unavailable, which would leave its subscriptions behind. The lock keeps a
163 # subscribe() that is still in flight from re-populating them afterwards.
164 async with self._subscription_lock:
165 await self.unsubscribe()
166
167 async def stop(self) -> None:
168 """Send STOP command to the player."""
169 if self.synced_to:
170 self.logger.debug(
171 "Ignore STOP command for %s: Player is synced to another player.",
172 self.player_id,
173 )
174 return
175 if self._attr_active_source in LINEIN_SOURCE_IDS:
176 # Play an invalid URI to force stop line-in sources
177 with contextlib.suppress(SoCoException):
178 await asyncio.to_thread(self.soco.play_uri, "")
179 else:
180 await asyncio.to_thread(self.soco.stop)
181 self.schedule_poll()
182 self.update_state()
183
184 async def play(self) -> None:
185 """Send PLAY command to the player."""
186 if self.synced_to:
187 self.logger.debug(
188 "Ignore PLAY command for %s: Player is synced to another player.",
189 self.player_id,
190 )
191 return
192 await asyncio.to_thread(self.soco.play)
193 self.schedule_poll()
194
195 async def pause(self) -> None:
196 """Send PAUSE command to the player."""
197 if self.synced_to:
198 self.logger.debug(
199 "Ignore PAUSE command for %s: Player is synced to another player.",
200 self.player_id,
201 )
202 return
203
204 def _pause() -> bool:
205 """Pause the speaker, reporting whether it accepts the command."""
206 if "Pause" not in self.soco.available_actions:
207 return False
208 self.soco.pause()
209 return True
210
211 if not await asyncio.to_thread(_pause):
212 # pause not possible
213 await self.stop()
214 return
215 self.schedule_poll()
216
217 async def volume_set(self, volume_level: int) -> None:
218 """Send VOLUME_SET command to the player."""
219
220 def set_volume_level(volume_level: int) -> None:
221 self.soco.volume = volume_level
222
223 await asyncio.to_thread(set_volume_level, volume_level)
224 self.schedule_poll()
225
226 async def volume_mute(self, muted: bool) -> None:
227 """Send VOLUME MUTE command to the player."""
228
229 def set_volume_mute(muted: bool) -> None:
230 self.soco.mute = muted
231
232 await asyncio.to_thread(set_volume_mute, muted)
233 self.schedule_poll()
234
235 async def play_media(self, media: PlayerMedia) -> None:
236 """Handle PLAY MEDIA on the player."""
237 if self.synced_to:
238 # this should be already handled by the player manager, but just in case...
239 msg = (
240 f"Player {self.display_name} can not "
241 "accept play_media command, it is synced to another player."
242 )
243 raise PlayerCommandFailed(
244 msg,
245 translation_key="play_media_synced",
246 translation_owner=self.translation_owner,
247 translation_args=[self.display_name],
248 )
249
250 stream_url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
251 if not media.duration:
252 # Sonos really does not like FLAC streams without duration
253 stream_url = stream_url.replace(".flac", ".mp3")
254
255 didl_metadata = create_didl_metadata(media, stream_url)
256 is_announcement = media.media_type == MediaType.ANNOUNCEMENT
257 force_radio = False if is_announcement else not media.duration
258
259 await asyncio.to_thread(
260 self.soco.play_uri, stream_url, meta=didl_metadata, force_radio=force_radio
261 )
262 self.schedule_poll()
263
264 async def enqueue_next_media(self, media: PlayerMedia) -> None:
265 """Handle enqueuing next media item."""
266 if self.synced_to:
267 # this should be already handled by the player manager, but just in case...
268 msg = (
269 f"Player {self.display_name} can not "
270 "accept enqueue command, it is synced to another player."
271 )
272 raise PlayerCommandFailed(
273 msg,
274 translation_key="enqueue_synced",
275 translation_owner=self.translation_owner,
276 translation_args=[self.display_name],
277 )
278
279 stream_url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
280 didl_metadata = create_didl_metadata(media, stream_url)
281
282 def add_to_queue() -> None:
283 self.soco.avTransport.SetNextAVTransportURI(
284 [
285 ("InstanceID", 0),
286 ("NextURI", stream_url),
287 ("NextURIMetaData", didl_metadata),
288 ]
289 )
290
291 await asyncio.to_thread(add_to_queue)
292 self.schedule_poll()
293
294 async def select_source(self, source: str) -> None:
295 """Handle SELECT SOURCE command on the player."""
296 if source in LINEIN_SOURCE_IDS:
297
298 def _switch_to_linein() -> None:
299 if source == SOURCE_TV:
300 self.soco.switch_to_tv()
301 elif source == SOURCE_LINEIN:
302 self.soco.switch_to_line_in()
303
304 await asyncio.to_thread(_switch_to_linein)
305 self.schedule_poll()
306 else:
307 await self.stop()
308
309 async def set_members(
310 self,
311 player_ids_to_add: list[str] | None = None,
312 player_ids_to_remove: list[str] | None = None,
313 ) -> None:
314 """Handle SET_MEMBERS command on the player."""
315 if self.synced_to:
316 # this should not happen, but guard anyways
317 raise RuntimeError("Player is synced, cannot set members")
318 if not player_ids_to_add and not player_ids_to_remove:
319 return
320 player_ids_to_add = player_ids_to_add or []
321 player_ids_to_remove = player_ids_to_remove or []
322
323 if player_ids_to_remove:
324 for player_id in player_ids_to_remove:
325 if player_to_remove := cast("SonosPlayer", self.mass.players.get_player(player_id)):
326 await player_to_remove._unjoin()
327
328 if player_ids_to_add:
329 for player_id in player_ids_to_add:
330 if player_to_add := cast("SonosPlayer", self.mass.players.get_player(player_id)):
331 await player_to_add._join(self.soco)
332
333 def schedule_poll(self) -> None:
334 """Read the speaker state back shortly after a command was sent to it."""
335 self.mass.call_later(COMMAND_POLL_DELAY, self.poll, task_id=self._poll_task_id)
336
337 async def poll(self) -> None:
338 """Poll player for state updates."""
339
340 def _poll() -> None:
341 """Poll the speaker for updates (NOT async friendly)."""
342 self.update_groups()
343 self.poll_media()
344 self._attr_volume_level = self.soco.volume
345 self._attr_volume_muted = self.soco.mute
346
347 if not self._attr_available:
348 await self._check_availability()
349 if not self._attr_available:
350 return
351 try:
352 await asyncio.to_thread(_poll)
353 except OSError, SoCoException, SonosUpdateError:
354 # a single failed poll does not mean the speaker is gone; the availability
355 # check decides based on how long it has been silent
356 await self._check_availability()
357 else:
358 self._speaker_activity("poll")
359
360 @soco_error()
361 def poll_media(self) -> None:
362 """Poll information about currently playing media."""
363 transport_info = self.soco.get_current_transport_info()
364 new_status = transport_info["current_transport_state"]
365
366 if new_status == SONOS_STATE_TRANSITIONING:
367 self._attr_poll_interval = TRANSITION_POLL_INTERVAL
368 return
369 self._attr_poll_interval = POLL_INTERVAL
370
371 new_status = _convert_state(new_status)
372 update_position = new_status != self._attr_playback_state
373 self._attr_playback_state = new_status
374 self._set_basic_track_info(update_position=update_position)
375 self.update_player()
376
377 async def update_ip(self, soco: SoCo) -> None:
378 """
379 Handle a Sonos player that was rediscovered at another IP-address.
380
381 :param soco: The SoCo instance discovered at the new address.
382 """
383 if self._unloaded or self._attr_available:
384 return
385 self.logger.debug(
386 "Player IP-address changed from %s to %s", self.soco.ip_address, soco.ip_address
387 )
388 # the UPnP endpoints of a SoCo instance are resolved once, when it is constructed,
389 # so reaching the speaker at its new address takes the rediscovered instance
390 self.soco = soco
391 try:
392 await asyncio.to_thread(self.ping)
393 except SonosUpdateError:
394 # the regular poll retries the new address until the speaker answers again
395 return
396 # mark the speaker alive before reading it back, so its state survives the update
397 self._speaker_activity("IP change")
398 await self.setup()
399 self._attr_device_info = DeviceInfo(
400 model=self._attr_device_info.model,
401 manufacturer=self._attr_device_info.manufacturer,
402 )
403 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, soco.ip_address)
404 self._attr_device_info.add_identifier(IdentifierType.UUID, self.player_id)
405 mac_address = self._extract_mac_from_player_id()
406 if mac_address:
407 self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
408 self.update_player()
409
410 @soco_error()
411 def ping(self) -> None:
412 """Test device availability. Failure will raise SonosUpdateError."""
413 self.soco.renderingControl.GetVolume([("InstanceID", 0), ("Channel", "Master")], timeout=1)
414
415 def update_player(self, signal_update: bool = True) -> None:
416 """Update Sonos Player."""
417 if self._unloaded:
418 return
419 self._update_attributes()
420 if signal_update:
421 # send update to the player manager right away only if we are triggered from an event
422 # when we're just updating from a manual poll, the player manager
423 # will detect changes to the player object itself
424 self.mass.loop.call_soon_threadsafe(self.update_state)
425
426 def log_subscription_result(self, result: Any, event: str, level: int = logging.DEBUG) -> None:
427 """Log a message if a subscription action (create/renew/stop) results in an exception."""
428 if not isinstance(result, Exception):
429 return
430
431 if isinstance(result, asyncio.exceptions.TimeoutError):
432 message = "Request timed out"
433 exc_info = None
434 else:
435 message = str(result)
436 exc_info = result if not str(result) else None
437
438 self.logger.log(
439 level,
440 "%s failed for %s: %s",
441 event,
442 self.display_name,
443 message,
444 exc_info=exc_info if self.logger.isEnabledFor(10) else None,
445 )
446
447 async def subscribe(self) -> None:
448 """Initiate event subscriptions under an async lock."""
449 async with self._subscription_lock:
450 try:
451 # Create event subscriptions.
452 subscriptions = [
453 self._subscribe_target(getattr(self.soco, service), self._handle_event)
454 for service in self.missing_subscriptions
455 ]
456 if not subscriptions:
457 return
458 self.logger.log(
459 VERBOSE_LOG_LEVEL, "Creating subscriptions for %s", self.display_name
460 )
461 results = await asyncio.gather(*subscriptions, return_exceptions=True)
462 for result in results:
463 self.log_subscription_result(result, "Creating subscription", logging.WARNING)
464 if any(isinstance(result, Exception) for result in results):
465 raise SonosSubscriptionsFailed
466 except SonosSubscriptionsFailed:
467 self.logger.warning("Creating subscriptions failed for %s", self.display_name)
468 # the subscription lock is already held here and is not reentrant
469 await self._offline()
470
471 async def unsubscribe(self) -> None:
472 """Cancel all subscriptions."""
473 if not self._subscriptions:
474 return
475 self.logger.log(VERBOSE_LOG_LEVEL, "Unsubscribing from events for %s", self.display_name)
476 # drop the subscriptions before awaiting: if the unsubscribe is cancelled midway
477 # they would stay behind as stale entries, and subscribe() skips the services it
478 # believes are still subscribed, leaving the speaker without events entirely
479 subscriptions, self._subscriptions = self._subscriptions, []
480 results = await asyncio.gather(
481 *(subscription.unsubscribe() for subscription in subscriptions),
482 return_exceptions=True,
483 )
484 for result in results:
485 self.log_subscription_result(result, "Unsubscribe")
486
487 def update_groups(self) -> None:
488 """Update group topology when polling."""
489 asyncio.run_coroutine_threadsafe(self.create_update_groups_coro(), self.mass.loop)
490
491 def create_update_groups_coro(
492 self, event: SonosEvent | None = None
493 ) -> Coroutine[Any, Any, None]:
494 """Handle callback for topology change event."""
495
496 def _get_soco_group() -> list[str]:
497 """Ask SoCo cache for existing topology."""
498 coordinator_uid = self.soco.uid
499 joined_uids = []
500 with contextlib.suppress(OSError, SoCoException):
501 if self.soco.group and self.soco.group.coordinator:
502 coordinator_uid = self.soco.group.coordinator.uid
503 joined_uids = [
504 p.uid
505 for p in self.soco.group.members
506 if p.uid != coordinator_uid and p.is_visible
507 ]
508
509 return [coordinator_uid, *joined_uids]
510
511 async def _extract_group(event: SonosEvent | None) -> list[str]:
512 """Extract group layout from a topology event."""
513 group = event and event.zone_player_uui_ds_in_group
514 if group:
515 assert isinstance(group, str)
516 return group.split(",")
517 return await asyncio.to_thread(_get_soco_group)
518
519 def _regroup(group: list[str]) -> None:
520 """Rebuild internal group layout (async safe)."""
521 if group == [self.soco.uid] and not self._attr_group_members:
522 # Skip updating existing single speakers in polling mode
523 return
524
525 group_members_ids = []
526
527 for uid in group:
528 speaker = self.mass.players.get_player(uid)
529 if speaker:
530 group_members_ids.append(uid)
531 else:
532 self.logger.debug(
533 "%s group member unavailable (%s), will try again",
534 self.display_name,
535 uid,
536 )
537 return
538
539 if self._attr_group_members == group_members_ids:
540 # Useful in polling mode for speakers with stereo pairs or surrounds
541 # as those "invisible" speakers will bypass the single speaker check
542 return
543
544 self._attr_group_members = group_members_ids
545 self.mass.loop.call_soon_threadsafe(self.update_state)
546
547 self.logger.debug("Regrouped %s: %s", self.display_name, self._attr_group_members)
548 self.update_player()
549
550 async def _handle_group_event(event: SonosEvent | None) -> None:
551 """Get async lock and handle event."""
552 if self._unloaded:
553 return
554 _provider = cast("SonosPlayerProvider", self._provider)
555 async with _provider.topology_condition:
556 group = await _extract_group(event)
557 if self.soco.uid == group[0]:
558 _regroup(group)
559 _provider.topology_condition.notify_all()
560
561 return _handle_group_event(event)
562
563 async def wait_for_groups(self, groups: list[list[SonosPlayer]]) -> None:
564 """Wait until all groups are present, or timeout."""
565
566 def _test_groups(groups: list[list[SonosPlayer]]) -> bool:
567 """Return whether all groups exist now."""
568 for group in groups:
569 coordinator = group[0]
570
571 # Test that coordinator is coordinating
572 current_group = coordinator.group_members
573 if coordinator != current_group[0]:
574 return False
575
576 # Test that joined members match
577 if set(group[1:]) != set(current_group[1:]):
578 return False
579
580 return True
581
582 _provider = cast("SonosPlayerProvider", self._provider)
583 try:
584 async with asyncio.timeout(5):
585 while not _test_groups(groups):
586 await _provider.topology_condition.wait()
587 except TimeoutError:
588 self.logger.warning("Timeout waiting for target groups %s", groups)
589
590 if players := self.mass.players.all_players(provider_filter=_provider.instance_id):
591 any_speaker = cast("SonosPlayer", players[0])
592 any_speaker.soco.zone_group_state.clear_cache()
593
594 @soco_error()
595 async def _join(self, coordinator: SoCo) -> None:
596 """
597 Join this speaker to the group of the given coordinator.
598
599 :param coordinator: The SoCo instance of the speaker leading the group.
600 """
601 await asyncio.to_thread(self.soco.join, coordinator)
602 self.schedule_poll()
603
604 @soco_error()
605 async def _unjoin(self) -> None:
606 """Remove this speaker from the group it is currently in."""
607 await asyncio.to_thread(self.soco.unjoin)
608 self.schedule_poll()
609
610 def _extract_mac_from_player_id(self) -> str | None:
611 """
612 Extract MAC address from Sonos player_id.
613
614 Sonos player_ids follow the format RINCON_XXXXXXXXXXXX01400 where
615 the middle 12 hex characters represent the MAC address.
616
617 :return: MAC address string in XX:XX:XX:XX:XX:XX format, or None if not extractable.
618 """
619 # Remove RINCON_ prefix if present
620 player_id = self.player_id
621 player_id = player_id.removeprefix("RINCON_")
622
623 # Remove the 01400 suffix (or similar) - should be last 5 chars
624 if len(player_id) >= 17: # 12 hex chars for MAC + 5 chars suffix
625 mac_hex = player_id[:12]
626 else:
627 return None
628
629 # Validate it looks like a MAC (all hex characters)
630 try:
631 int(mac_hex, 16)
632 except ValueError:
633 return None
634
635 # Format as XX:XX:XX:XX:XX:XX
636 return ":".join(mac_hex[i : i + 2].upper() for i in range(0, 12, 2))
637
638 async def _check_availability(self) -> None:
639 """Check if the player is still available."""
640 # skip the ping while events or polls recently succeeded, so one slow or dropped
641 # request does not mark a healthy speaker unavailable. An unavailable speaker is
642 # always pinged so it recovers quickly, no matter why it went offline.
643 if self._attr_available and time.monotonic() - self._last_activity < AVAILABILITY_TIMEOUT:
644 return
645 try:
646 await asyncio.to_thread(self.ping)
647 self._speaker_activity("ping")
648 except SonosUpdateError:
649 if not self._attr_available:
650 return
651 self.logger.warning(
652 "No recent activity and cannot reach %s, marking unavailable",
653 self.display_name,
654 )
655 await self.offline()
656
657 @soco_error()
658 def _poll_track_info(self) -> dict[str, Any]:
659 """
660 Poll the speaker for current track info.
661
662 Add converted position values (NOT async fiendly).
663 """
664 track_info: dict[str, Any] = self.soco.get_current_track_info()
665 track_info[DURATION_SECONDS] = _timespan_secs(track_info.get("duration"))
666 track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position"))
667 return track_info
668
669 async def _offline(self) -> None:
670 """Handle removal of speaker when unavailable; caller must hold the subscription lock."""
671 if not self._attr_available:
672 return
673
674 if self._resub_cooldown_expires_at is None and not self.mass.closing:
675 self._resub_cooldown_expires_at = time.monotonic() + RESUB_COOLDOWN_SECONDS
676 self.logger.debug("Starting resubscription cooldown for %s", self.display_name)
677
678 self._attr_available = False
679
680 self.update_state()
681 await self.unsubscribe()
682
683 async def _subscribe_target(
684 self, target: SubscriptionBase, sub_callback: Callable[[SonosEvent], None]
685 ) -> None:
686 """Create a Sonos subscription for given target."""
687
688 def on_renew_failed(exception: Exception) -> None:
689 """Handle a failed subscription renewal callback."""
690 self.mass.create_task(self._renew_failed(exception))
691
692 # Use events_asyncio which makes subscribe() async-awaitable
693 subscription = await target.subscribe(
694 auto_renew=True, requested_timeout=SUBSCRIPTION_TIMEOUT
695 )
696 subscription.callback = sub_callback
697 subscription.auto_renew_fail = on_renew_failed
698 self._subscriptions.append(subscription)
699
700 async def _renew_failed(self, exception: Exception) -> None:
701 """
702 Mark the speaker as offline after a subscription renewal failure.
703
704 This is to reset the state to allow a future clean subscription attempt.
705 """
706 if not self._attr_available:
707 return
708
709 self.log_subscription_result(exception, "Subscription renewal", logging.WARNING)
710 await self.offline()
711
712 def _handle_event(self, event: SonosEvent) -> None:
713 """Handle SonosEvent callback."""
714 service_type: str = event.service.service_type
715 self._speaker_activity(f"{service_type} subscription")
716 if service_type == "DeviceProperties":
717 self.update_player()
718 return
719 if service_type == "AVTransport":
720 self._handle_avtransport_event(event)
721 return
722 if service_type == "RenderingControl":
723 self._handle_rendering_control_event(event)
724 return
725 if service_type == "ZoneGroupTopology":
726 self._handle_zone_group_topology_event(event)
727 return
728
729 def _handle_avtransport_event(self, event: SonosEvent) -> None:
730 """Update information about currently playing media from an event."""
731 # NOTE: The new coordinator can be provided in a media update event but
732 # before the ZoneGroupState updates. If this happens the playback
733 # state will be incorrect and should be ignored. Switching to the
734 # new coordinator will use its media. The regrouping process will
735 # be completed during the next ZoneGroupState update.
736
737 # Missing transport_state indicates a transient error
738 if (new_status := event.variables.get("transport_state")) is None:
739 return
740
741 # Ignore transitions, we should get the target state soon
742 if new_status == SONOS_STATE_TRANSITIONING:
743 self._attr_poll_interval = TRANSITION_POLL_INTERVAL
744 return
745 self._attr_poll_interval = POLL_INTERVAL
746
747 evars = event.variables
748 new_status = _convert_state(evars["transport_state"])
749 state_changed = new_status != self._attr_playback_state
750
751 self._attr_playback_state = new_status
752
753 track_uri = evars["enqueued_transport_uri"] or evars["current_track_uri"]
754 audio_source = self.soco.music_source_from_uri(track_uri)
755
756 self._set_basic_track_info(update_position=state_changed)
757 ct_md = evars["current_track_meta_data"]
758
759 et_uri_md = evars["enqueued_transport_uri_meta_data"]
760
761 channel = ""
762 if audio_source == MUSIC_SRC_RADIO:
763 if et_uri_md:
764 channel = et_uri_md.title
765
766 # Extra guards for S1 compatibility
767 if ct_md and hasattr(ct_md, "radio_show") and ct_md.radio_show:
768 radio_show = ct_md.radio_show.split(",")[0]
769 channel = " ⢠".join(filter(None, [channel, radio_show]))
770
771 if isinstance(et_uri_md, DidlAudioBroadcast) and self._attr_current_media:
772 self._attr_current_media.title = self._attr_current_media.title or channel
773
774 self.update_player()
775
776 def _handle_rendering_control_event(self, event: SonosEvent) -> None:
777 """Update information about currently volume settings."""
778 variables = event.variables
779
780 if "volume" in variables:
781 volume = variables["volume"]
782 self._attr_volume_level = int(volume["Master"])
783
784 if mute := variables.get("mute"):
785 self._attr_volume_muted = mute["Master"] == "1"
786
787 self.update_player()
788
789 def _handle_zone_group_topology_event(self, event: SonosEvent) -> None:
790 """Handle callback for topology change event."""
791 if "zone_player_uui_ds_in_group" not in event.variables:
792 return
793 asyncio.run_coroutine_threadsafe(self.create_update_groups_coro(event), self.mass.loop)
794
795 def _update_attributes(self) -> None:
796 """Update attributes of the MA Player from SoCo state."""
797 if not self._attr_available:
798 self._attr_playback_state = PlayerState.IDLE
799 self._attr_group_members.clear()
800 return
801
802 def _set_basic_track_info(self, update_position: bool = False) -> None:
803 """Query the speaker to update media metadata and position info."""
804 try:
805 track_info = self._poll_track_info()
806 except SonosUpdateError as err:
807 self.logger.warning("Fetching track info failed: %s", err)
808 return
809 uri = track_info["uri"]
810 if not uri:
811 # no current track means nothing is loaded, so no source is active either.
812 # Stopping a line-in source empties the transport, so this is a normal path.
813 self._attr_elapsed_time = None
814 self._attr_elapsed_time_last_updated = None
815 self._attr_active_source = None
816 self._attr_current_media = None
817 return
818
819 audio_source = self.soco.music_source_from_uri(uri)
820 if source_id := LINEIN_SOURCE_MAPPING.get(audio_source):
821 self._attr_elapsed_time = None
822 self._attr_elapsed_time_last_updated = None
823 self._attr_active_source = source_id
824 self._attr_current_media = None
825 if source_id not in [x.id for x in self._attr_source_list]:
826 self._attr_source_list.append(PLAYER_SOURCE_MAP[source_id])
827 return
828
829 current_media = PlayerMedia(
830 uri=uri,
831 artist=track_info.get("artist"),
832 album=track_info.get("album"),
833 title=track_info.get("title"),
834 image_url=track_info.get("album_art"),
835 )
836 self._attr_current_media = current_media
837 self._attr_active_source = None
838 self._update_media_position(track_info, force_update=update_position)
839
840 def _update_media_position(
841 self, position_info: dict[str, int], force_update: bool = False
842 ) -> None:
843 """Update state when playing music tracks."""
844 duration = position_info.get(DURATION_SECONDS)
845 current_position = position_info.get(POSITION_SECONDS)
846
847 if not (duration or current_position):
848 self._attr_elapsed_time = None
849 self._attr_elapsed_time_last_updated = None
850 return
851
852 should_update = force_update
853 if self._attr_current_media:
854 self._attr_current_media.duration = duration
855
856 # player started reporting position?
857 if current_position is not None and self._attr_elapsed_time is None:
858 should_update = True
859
860 # position jumped?
861 if current_position is not None and self._attr_elapsed_time is not None:
862 if self._attr_playback_state == PlaybackState.PLAYING:
863 assert self._attr_elapsed_time_last_updated is not None
864 time_diff = time.time() - self._attr_elapsed_time_last_updated
865 else:
866 time_diff = 0
867
868 calculated_position = self._attr_elapsed_time + time_diff
869
870 if abs(calculated_position - current_position) > 1.5:
871 should_update = True
872
873 if current_position is None:
874 self._attr_elapsed_time = None
875 self._attr_elapsed_time_last_updated = None
876 elif should_update:
877 self._attr_elapsed_time = current_position
878 self._attr_elapsed_time_last_updated = time.time()
879
880 def _speaker_activity(self, source: str) -> None:
881 """Track the last activity on this speaker, set availability and resubscribe."""
882 if self._unloaded:
883 return
884 if self._resub_cooldown_expires_at:
885 if time.monotonic() < self._resub_cooldown_expires_at:
886 self.logger.debug(
887 "Activity on %s from %s while in cooldown, ignoring",
888 self.display_name,
889 source,
890 )
891 return
892 self._resub_cooldown_expires_at = None
893
894 self.logger.log(VERBOSE_LOG_LEVEL, "Activity on %s from %s", self.display_name, source)
895 self._last_activity = time.monotonic()
896 was_available = self._attr_available
897 self._attr_available = True
898 if not was_available:
899 self.update_player()
900 self.mass.loop.call_soon_threadsafe(self.mass.create_task, self.subscribe())
901
902
903def _convert_state(sonos_state: str | None) -> PlayerState:
904 """Convert Sonos state to PlayerState."""
905 if sonos_state == "PLAYING":
906 return PlayerState.PLAYING
907 if sonos_state == "TRANSITIONING":
908 return PlayerState.PLAYING
909 if sonos_state == "PAUSED_PLAYBACK":
910 return PlayerState.PAUSED
911 return PlayerState.IDLE
912
913
914def _timespan_secs(timespan: str | None) -> int | None:
915 """Parse a time-span into number of seconds."""
916 if timespan in ("", "NOT_IMPLEMENTED"):
917 return None
918 if timespan is None:
919 return None
920 return int(sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":")))))
921