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