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