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