music-assistant-server

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