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    async def set_members(
287        self,
288        player_ids_to_add: list[str] | None = None,
289        player_ids_to_remove: list[str] | None = None,
290    ) -> None:
291        """Handle SET_MEMBERS command on the player."""
292        if self.synced_to:
293            # this should not happen, but guard anyways
294            raise RuntimeError("Player is synced, cannot set members")
295        if not player_ids_to_add and not player_ids_to_remove:
296            return
297        player_ids_to_add = player_ids_to_add or []
298        player_ids_to_remove = player_ids_to_remove or []
299
300        if player_ids_to_remove:
301            for player_id in player_ids_to_remove:
302                if player_to_remove := cast("SonosPlayer", self.mass.players.get_player(player_id)):
303                    await player_to_remove._unjoin()
304
305        if player_ids_to_add:
306            for player_id in player_ids_to_add:
307                if player_to_add := cast("SonosPlayer", self.mass.players.get_player(player_id)):
308                    await player_to_add._join(self.soco)
309
310    def schedule_poll(self) -> None:
311        """Read the speaker state back shortly after a command was sent to it."""
312        self.mass.call_later(COMMAND_POLL_DELAY, self.poll, task_id=self._poll_task_id)
313
314    async def poll(self) -> None:
315        """Poll player for state updates."""
316
317        def _poll() -> None:
318            """Poll the speaker for updates (NOT async friendly)."""
319            self.update_groups()
320            self.poll_media()
321            self._attr_volume_level = self.soco.volume
322            self._attr_volume_muted = self.soco.mute
323
324        await self._check_availability()
325        if self._attr_available:
326            await asyncio.to_thread(_poll)
327
328    @soco_error()
329    def poll_media(self) -> None:
330        """Poll information about currently playing media."""
331        transport_info = self.soco.get_current_transport_info()
332        new_status = transport_info["current_transport_state"]
333
334        if new_status == SONOS_STATE_TRANSITIONING:
335            self._attr_poll_interval = TRANSITION_POLL_INTERVAL
336            return
337        self._attr_poll_interval = POLL_INTERVAL
338
339        new_status = _convert_state(new_status)
340        update_position = new_status != self._attr_playback_state
341        self._attr_playback_state = new_status
342        self._set_basic_track_info(update_position=update_position)
343        self.update_player()
344
345    async def update_ip(self, soco: SoCo) -> None:
346        """
347        Handle a Sonos player that was rediscovered at another IP-address.
348
349        :param soco: The SoCo instance discovered at the new address.
350        """
351        if self._unloaded or self._attr_available:
352            return
353        self.logger.debug(
354            "Player IP-address changed from %s to %s", self.soco.ip_address, soco.ip_address
355        )
356        # the UPnP endpoints of a SoCo instance are resolved once, when it is constructed,
357        # so reaching the speaker at its new address takes the rediscovered instance
358        self.soco = soco
359        try:
360            await asyncio.to_thread(self.ping)
361        except SonosUpdateError:
362            # the regular poll retries the new address until the speaker answers again
363            return
364        # mark the speaker alive before reading it back, so its state survives the update
365        self._speaker_activity("IP change")
366        await self.setup()
367        self._attr_device_info = DeviceInfo(
368            model=self._attr_device_info.model,
369            manufacturer=self._attr_device_info.manufacturer,
370        )
371        self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, soco.ip_address)
372        self._attr_device_info.add_identifier(IdentifierType.UUID, self.player_id)
373        mac_address = self._extract_mac_from_player_id()
374        if mac_address:
375            self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
376        self.update_player()
377
378    @soco_error()
379    def ping(self) -> None:
380        """Test device availability. Failure will raise SonosUpdateError."""
381        self.soco.renderingControl.GetVolume([("InstanceID", 0), ("Channel", "Master")], timeout=1)
382
383    def update_player(self, signal_update: bool = True) -> None:
384        """Update Sonos Player."""
385        if self._unloaded:
386            return
387        self._update_attributes()
388        if signal_update:
389            # send update to the player manager right away only if we are triggered from an event
390            # when we're just updating from a manual poll, the player manager
391            # will detect changes to the player object itself
392            self.mass.loop.call_soon_threadsafe(self.update_state)
393
394    def log_subscription_result(self, result: Any, event: str, level: int = logging.DEBUG) -> None:
395        """Log a message if a subscription action (create/renew/stop) results in an exception."""
396        if not isinstance(result, Exception):
397            return
398
399        if isinstance(result, asyncio.exceptions.TimeoutError):
400            message = "Request timed out"
401            exc_info = None
402        else:
403            message = str(result)
404            exc_info = result if not str(result) else None
405
406        self.logger.log(
407            level,
408            "%s failed for %s: %s",
409            event,
410            self.display_name,
411            message,
412            exc_info=exc_info if self.logger.isEnabledFor(10) else None,
413        )
414
415    async def subscribe(self) -> None:
416        """Initiate event subscriptions under an async lock."""
417        async with self._subscription_lock:
418            try:
419                # Create event subscriptions.
420                subscriptions = [
421                    self._subscribe_target(getattr(self.soco, service), self._handle_event)
422                    for service in self.missing_subscriptions
423                ]
424                if not subscriptions:
425                    return
426                self.logger.log(
427                    VERBOSE_LOG_LEVEL, "Creating subscriptions for %s", self.display_name
428                )
429                results = await asyncio.gather(*subscriptions, return_exceptions=True)
430                for result in results:
431                    self.log_subscription_result(result, "Creating subscription", logging.WARNING)
432                if any(isinstance(result, Exception) for result in results):
433                    raise SonosSubscriptionsFailed
434            except SonosSubscriptionsFailed:
435                self.logger.warning("Creating subscriptions failed for %s", self.display_name)
436                # the subscription lock is already held here and is not reentrant
437                await self._offline()
438
439    async def unsubscribe(self) -> None:
440        """Cancel all subscriptions."""
441        if not self._subscriptions:
442            return
443        self.logger.log(VERBOSE_LOG_LEVEL, "Unsubscribing from events for %s", self.display_name)
444        # drop the subscriptions before awaiting: if the unsubscribe is cancelled midway
445        # they would stay behind as stale entries, and subscribe() skips the services it
446        # believes are still subscribed, leaving the speaker without events entirely
447        subscriptions, self._subscriptions = self._subscriptions, []
448        results = await asyncio.gather(
449            *(subscription.unsubscribe() for subscription in subscriptions),
450            return_exceptions=True,
451        )
452        for result in results:
453            self.log_subscription_result(result, "Unsubscribe")
454
455    def update_groups(self) -> None:
456        """Update group topology when polling."""
457        asyncio.run_coroutine_threadsafe(self.create_update_groups_coro(), self.mass.loop)
458
459    def create_update_groups_coro(
460        self, event: SonosEvent | None = None
461    ) -> Coroutine[Any, Any, None]:
462        """Handle callback for topology change event."""
463
464        def _get_soco_group() -> list[str]:
465            """Ask SoCo cache for existing topology."""
466            coordinator_uid = self.soco.uid
467            joined_uids = []
468            with contextlib.suppress(OSError, SoCoException):
469                if self.soco.group and self.soco.group.coordinator:
470                    coordinator_uid = self.soco.group.coordinator.uid
471                    joined_uids = [
472                        p.uid
473                        for p in self.soco.group.members
474                        if p.uid != coordinator_uid and p.is_visible
475                    ]
476
477            return [coordinator_uid, *joined_uids]
478
479        async def _extract_group(event: SonosEvent | None) -> list[str]:
480            """Extract group layout from a topology event."""
481            group = event and event.zone_player_uui_ds_in_group
482            if group:
483                assert isinstance(group, str)
484                return group.split(",")
485            return await asyncio.to_thread(_get_soco_group)
486
487        def _regroup(group: list[str]) -> None:
488            """Rebuild internal group layout (async safe)."""
489            if group == [self.soco.uid] and not self._attr_group_members:
490                # Skip updating existing single speakers in polling mode
491                return
492
493            group_members_ids = []
494
495            for uid in group:
496                speaker = self.mass.players.get_player(uid)
497                if speaker:
498                    group_members_ids.append(uid)
499                else:
500                    self.logger.debug(
501                        "%s group member unavailable (%s), will try again",
502                        self.display_name,
503                        uid,
504                    )
505                    return
506
507            if self._attr_group_members == group_members_ids:
508                # Useful in polling mode for speakers with stereo pairs or surrounds
509                # as those "invisible" speakers will bypass the single speaker check
510                return
511
512            self._attr_group_members = group_members_ids
513            self.mass.loop.call_soon_threadsafe(self.update_state)
514
515            self.logger.debug("Regrouped %s: %s", self.display_name, self._attr_group_members)
516            self.update_player()
517
518        async def _handle_group_event(event: SonosEvent | None) -> None:
519            """Get async lock and handle event."""
520            if self._unloaded:
521                return
522            _provider = cast("SonosPlayerProvider", self._provider)
523            async with _provider.topology_condition:
524                group = await _extract_group(event)
525                if self.soco.uid == group[0]:
526                    _regroup(group)
527                    _provider.topology_condition.notify_all()
528
529        return _handle_group_event(event)
530
531    async def wait_for_groups(self, groups: list[list[SonosPlayer]]) -> None:
532        """Wait until all groups are present, or timeout."""
533
534        def _test_groups(groups: list[list[SonosPlayer]]) -> bool:
535            """Return whether all groups exist now."""
536            for group in groups:
537                coordinator = group[0]
538
539                # Test that coordinator is coordinating
540                current_group = coordinator.group_members
541                if coordinator != current_group[0]:
542                    return False
543
544                # Test that joined members match
545                if set(group[1:]) != set(current_group[1:]):
546                    return False
547
548            return True
549
550        _provider = cast("SonosPlayerProvider", self._provider)
551        try:
552            async with asyncio.timeout(5):
553                while not _test_groups(groups):
554                    await _provider.topology_condition.wait()
555        except TimeoutError:
556            self.logger.warning("Timeout waiting for target groups %s", groups)
557
558        if players := self.mass.players.all_players(provider_filter=_provider.instance_id):
559            any_speaker = cast("SonosPlayer", players[0])
560            any_speaker.soco.zone_group_state.clear_cache()
561
562    @soco_error()
563    async def _join(self, coordinator: SoCo) -> None:
564        """
565        Join this speaker to the group of the given coordinator.
566
567        :param coordinator: The SoCo instance of the speaker leading the group.
568        """
569        await asyncio.to_thread(self.soco.join, coordinator)
570        self.schedule_poll()
571
572    @soco_error()
573    async def _unjoin(self) -> None:
574        """Remove this speaker from the group it is currently in."""
575        await asyncio.to_thread(self.soco.unjoin)
576        self.schedule_poll()
577
578    def _extract_mac_from_player_id(self) -> str | None:
579        """
580        Extract MAC address from Sonos player_id.
581
582        Sonos player_ids follow the format RINCON_XXXXXXXXXXXX01400 where
583        the middle 12 hex characters represent the MAC address.
584
585        :return: MAC address string in XX:XX:XX:XX:XX:XX format, or None if not extractable.
586        """
587        # Remove RINCON_ prefix if present
588        player_id = self.player_id
589        player_id = player_id.removeprefix("RINCON_")
590
591        # Remove the 01400 suffix (or similar) - should be last 5 chars
592        if len(player_id) >= 17:  # 12 hex chars for MAC + 5 chars suffix
593            mac_hex = player_id[:12]
594        else:
595            return None
596
597        # Validate it looks like a MAC (all hex characters)
598        try:
599            int(mac_hex, 16)
600        except ValueError:
601            return None
602
603        # Format as XX:XX:XX:XX:XX:XX
604        return ":".join(mac_hex[i : i + 2].upper() for i in range(0, 12, 2))
605
606    async def _check_availability(self) -> None:
607        """Check if the player is still available."""
608        try:
609            await asyncio.to_thread(self.ping)
610            self._speaker_activity("ping")
611        except SonosUpdateError:
612            if not self._attr_available:
613                return
614            self.logger.warning(
615                "No recent activity and cannot reach %s, marking unavailable",
616                self.display_name,
617            )
618            await self.offline()
619
620    @soco_error()
621    def _poll_track_info(self) -> dict[str, Any]:
622        """
623        Poll the speaker for current track info.
624
625        Add converted position values (NOT async fiendly).
626        """
627        track_info: dict[str, Any] = self.soco.get_current_track_info()
628        track_info[DURATION_SECONDS] = _timespan_secs(track_info.get("duration"))
629        track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position"))
630        return track_info
631
632    async def _offline(self) -> None:
633        """Handle removal of speaker when unavailable; caller must hold the subscription lock."""
634        if not self._attr_available:
635            return
636
637        if self._resub_cooldown_expires_at is None and not self.mass.closing:
638            self._resub_cooldown_expires_at = time.monotonic() + RESUB_COOLDOWN_SECONDS
639            self.logger.debug("Starting resubscription cooldown for %s", self.display_name)
640
641        self._attr_available = False
642
643        self.update_state()
644        await self.unsubscribe()
645
646    async def _subscribe_target(
647        self, target: SubscriptionBase, sub_callback: Callable[[SonosEvent], None]
648    ) -> None:
649        """Create a Sonos subscription for given target."""
650
651        def on_renew_failed(exception: Exception) -> None:
652            """Handle a failed subscription renewal callback."""
653            self.mass.create_task(self._renew_failed(exception))
654
655        # Use events_asyncio which makes subscribe() async-awaitable
656        subscription = await target.subscribe(
657            auto_renew=True, requested_timeout=SUBSCRIPTION_TIMEOUT
658        )
659        subscription.callback = sub_callback
660        subscription.auto_renew_fail = on_renew_failed
661        self._subscriptions.append(subscription)
662
663    async def _renew_failed(self, exception: Exception) -> None:
664        """
665        Mark the speaker as offline after a subscription renewal failure.
666
667        This is to reset the state to allow a future clean subscription attempt.
668        """
669        if not self._attr_available:
670            return
671
672        self.log_subscription_result(exception, "Subscription renewal", logging.WARNING)
673        await self.offline()
674
675    def _handle_event(self, event: SonosEvent) -> None:
676        """Handle SonosEvent callback."""
677        service_type: str = event.service.service_type
678        self._speaker_activity(f"{service_type} subscription")
679        if service_type == "DeviceProperties":
680            self.update_player()
681            return
682        if service_type == "AVTransport":
683            self._handle_avtransport_event(event)
684            return
685        if service_type == "RenderingControl":
686            self._handle_rendering_control_event(event)
687            return
688        if service_type == "ZoneGroupTopology":
689            self._handle_zone_group_topology_event(event)
690            return
691
692    def _handle_avtransport_event(self, event: SonosEvent) -> None:
693        """Update information about currently playing media from an event."""
694        # NOTE: The new coordinator can be provided in a media update event but
695        # before the ZoneGroupState updates. If this happens the playback
696        # state will be incorrect and should be ignored. Switching to the
697        # new coordinator will use its media. The regrouping process will
698        # be completed during the next ZoneGroupState update.
699
700        # Missing transport_state indicates a transient error
701        if (new_status := event.variables.get("transport_state")) is None:
702            return
703
704        # Ignore transitions, we should get the target state soon
705        if new_status == SONOS_STATE_TRANSITIONING:
706            self._attr_poll_interval = TRANSITION_POLL_INTERVAL
707            return
708        self._attr_poll_interval = POLL_INTERVAL
709
710        evars = event.variables
711        new_status = _convert_state(evars["transport_state"])
712        state_changed = new_status != self._attr_playback_state
713
714        self._attr_playback_state = new_status
715
716        track_uri = evars["enqueued_transport_uri"] or evars["current_track_uri"]
717        audio_source = self.soco.music_source_from_uri(track_uri)
718
719        self._set_basic_track_info(update_position=state_changed)
720        ct_md = evars["current_track_meta_data"]
721
722        et_uri_md = evars["enqueued_transport_uri_meta_data"]
723
724        channel = ""
725        if audio_source == MUSIC_SRC_RADIO:
726            if et_uri_md:
727                channel = et_uri_md.title
728
729            # Extra guards for S1 compatibility
730            if ct_md and hasattr(ct_md, "radio_show") and ct_md.radio_show:
731                radio_show = ct_md.radio_show.split(",")[0]
732                channel = " • ".join(filter(None, [channel, radio_show]))
733
734            if isinstance(et_uri_md, DidlAudioBroadcast) and self._attr_current_media:
735                self._attr_current_media.title = self._attr_current_media.title or channel
736
737        self.update_player()
738
739    def _handle_rendering_control_event(self, event: SonosEvent) -> None:
740        """Update information about currently volume settings."""
741        variables = event.variables
742
743        if "volume" in variables:
744            volume = variables["volume"]
745            self._attr_volume_level = int(volume["Master"])
746
747        if mute := variables.get("mute"):
748            self._attr_volume_muted = mute["Master"] == "1"
749
750        self.update_player()
751
752    def _handle_zone_group_topology_event(self, event: SonosEvent) -> None:
753        """Handle callback for topology change event."""
754        if "zone_player_uui_ds_in_group" not in event.variables:
755            return
756        asyncio.run_coroutine_threadsafe(self.create_update_groups_coro(event), self.mass.loop)
757
758    def _update_attributes(self) -> None:
759        """Update attributes of the MA Player from SoCo state."""
760        if not self._attr_available:
761            self._attr_playback_state = PlayerState.IDLE
762            self._attr_group_members.clear()
763            return
764
765    def _set_basic_track_info(self, update_position: bool = False) -> None:
766        """Query the speaker to update media metadata and position info."""
767        try:
768            track_info = self._poll_track_info()
769        except SonosUpdateError as err:
770            self.logger.warning("Fetching track info failed: %s", err)
771            return
772        if not track_info["uri"]:
773            return
774        uri = track_info["uri"]
775
776        audio_source = self.soco.music_source_from_uri(uri)
777        if (source_id := SOURCE_MAPPING.get(audio_source)) and audio_source in LINEIN_SOURCES:
778            self._attr_elapsed_time = None
779            self._attr_elapsed_time_last_updated = None
780            self._attr_active_source = source_id
781            self._attr_current_media = None
782            if source_id not in [x.id for x in self._attr_source_list]:
783                self._attr_source_list.append(PLAYER_SOURCE_MAP[source_id])
784            return
785
786        current_media = PlayerMedia(
787            uri=uri,
788            artist=track_info.get("artist"),
789            album=track_info.get("album"),
790            title=track_info.get("title"),
791            image_url=track_info.get("album_art"),
792        )
793        self._attr_current_media = current_media
794        self._attr_active_source = None
795        self._update_media_position(track_info, force_update=update_position)
796
797    def _update_media_position(
798        self, position_info: dict[str, int], force_update: bool = False
799    ) -> None:
800        """Update state when playing music tracks."""
801        duration = position_info.get(DURATION_SECONDS)
802        current_position = position_info.get(POSITION_SECONDS)
803
804        if not (duration or current_position):
805            self._attr_elapsed_time = None
806            self._attr_elapsed_time_last_updated = None
807            return
808
809        should_update = force_update
810        if self._attr_current_media:
811            self._attr_current_media.duration = duration
812
813        # player started reporting position?
814        if current_position is not None and self._attr_elapsed_time is None:
815            should_update = True
816
817        # position jumped?
818        if current_position is not None and self._attr_elapsed_time is not None:
819            if self._attr_playback_state == PlaybackState.PLAYING:
820                assert self._attr_elapsed_time_last_updated is not None
821                time_diff = time.time() - self._attr_elapsed_time_last_updated
822            else:
823                time_diff = 0
824
825            calculated_position = self._attr_elapsed_time + time_diff
826
827            if abs(calculated_position - current_position) > 1.5:
828                should_update = True
829
830        if current_position is None:
831            self._attr_elapsed_time = None
832            self._attr_elapsed_time_last_updated = None
833        elif should_update:
834            self._attr_elapsed_time = current_position
835            self._attr_elapsed_time_last_updated = time.time()
836
837    def _speaker_activity(self, source: str) -> None:
838        """Track the last activity on this speaker, set availability and resubscribe."""
839        if self._unloaded:
840            return
841        if self._resub_cooldown_expires_at:
842            if time.monotonic() < self._resub_cooldown_expires_at:
843                self.logger.debug(
844                    "Activity on %s from %s while in cooldown, ignoring",
845                    self.display_name,
846                    source,
847                )
848                return
849            self._resub_cooldown_expires_at = None
850
851        self.logger.log(VERBOSE_LOG_LEVEL, "Activity on %s from %s", self.display_name, source)
852        self._last_activity = time.monotonic()
853        was_available = self._attr_available
854        self._attr_available = True
855        if not was_available:
856            self.update_player()
857            self.mass.loop.call_soon_threadsafe(self.mass.create_task, self.subscribe())
858
859
860def _convert_state(sonos_state: str | None) -> PlayerState:
861    """Convert Sonos state to PlayerState."""
862    if sonos_state == "PLAYING":
863        return PlayerState.PLAYING
864    if sonos_state == "TRANSITIONING":
865        return PlayerState.PLAYING
866    if sonos_state == "PAUSED_PLAYBACK":
867        return PlayerState.PAUSED
868    return PlayerState.IDLE
869
870
871def _timespan_secs(timespan: str | None) -> int | None:
872    """Parse a time-span into number of seconds."""
873    if timespan in ("", "NOT_IMPLEMENTED"):
874        return None
875    if timespan is None:
876        return None
877    return int(sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":")))))
878