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