Refactor Sonos media metadata handling (#66840)

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

.coveragerc

@@ -1122,6 +1122,7 @@ omit =
     homeassistant/components/sonos/favorites.py
     homeassistant/components/sonos/helpers.py
     homeassistant/components/sonos/household_coordinator.py
+    homeassistant/components/sonos/media.py
     homeassistant/components/sonos/media_browser.py
     homeassistant/components/sonos/media_player.py
     homeassistant/components/sonos/speaker.py

homeassistant/components/sonos/const.py

@@ -159,13 +159,16 @@ SONOS_CREATE_MEDIA_PLAYER = "sonos_create_media_player"
 SONOS_FALLBACK_POLL = "sonos_fallback_poll"
 SONOS_ALARMS_UPDATED = "sonos_alarms_updated"
 SONOS_FAVORITES_UPDATED = "sonos_favorites_updated"
+SONOS_MEDIA_UPDATED = "sonos_media_updated"
 SONOS_SPEAKER_ACTIVITY = "sonos_speaker_activity"
 SONOS_SPEAKER_ADDED = "sonos_speaker_added"
 SONOS_STATE_UPDATED = "sonos_state_updated"
 SONOS_REBOOTED = "sonos_rebooted"
 SONOS_VANISHED = "sonos_vanished"
 
+SOURCE_AIRPLAY = "AirPlay"
 SOURCE_LINEIN = "Line-in"
+SOURCE_SPOTIFY_CONNECT = "Spotify Connect"
 SOURCE_TV = "TV"
 
 AVAILABILITY_CHECK_INTERVAL = datetime.timedelta(minutes=1)

homeassistant/components/sonos/media.py

@@ -0,0 +1,234 @@
+"""Support for media metadata handling."""
+from __future__ import annotations
+
+import datetime
+import logging
+from typing import Any
+
+from soco.core import (
+    MUSIC_SRC_AIRPLAY,
+    MUSIC_SRC_LINE_IN,
+    MUSIC_SRC_RADIO,
+    MUSIC_SRC_SPOTIFY_CONNECT,
+    MUSIC_SRC_TV,
+    SoCo,
+)
+from soco.data_structures import DidlAudioBroadcast, DidlPlaylistContainer
+from soco.music_library import MusicLibrary
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.config_validation import time_period_str
+from homeassistant.helpers.dispatcher import dispatcher_send
+from homeassistant.util import dt as dt_util
+
+from .const import (
+    SONOS_MEDIA_UPDATED,
+    SONOS_STATE_PLAYING,
+    SONOS_STATE_TRANSITIONING,
+    SOURCE_AIRPLAY,
+    SOURCE_LINEIN,
+    SOURCE_SPOTIFY_CONNECT,
+    SOURCE_TV,
+)
+from .helpers import soco_error
+
+LINEIN_SOURCES = (MUSIC_SRC_TV, MUSIC_SRC_LINE_IN)
+SOURCE_MAPPING = {
+    MUSIC_SRC_AIRPLAY: SOURCE_AIRPLAY,
+    MUSIC_SRC_TV: SOURCE_TV,
+    MUSIC_SRC_LINE_IN: SOURCE_LINEIN,
+    MUSIC_SRC_SPOTIFY_CONNECT: SOURCE_SPOTIFY_CONNECT,
+}
+UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None}
+DURATION_SECONDS = "duration_in_s"
+POSITION_SECONDS = "position_in_s"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _timespan_secs(timespan: str | None) -> None | float:
+    """Parse a time-span into number of seconds."""
+    if timespan in UNAVAILABLE_VALUES:
+        return None
+    return time_period_str(timespan).total_seconds()  # type: ignore[arg-type]
+
+
+class SonosMedia:
+    """Representation of the current Sonos media."""
+
+    def __init__(self, hass: HomeAssistant, soco: SoCo) -> None:
+        """Initialize a SonosMedia."""
+        self.hass = hass
+        self.soco = soco
+        self.play_mode: str | None = None
+        self.playback_status: str | None = None
+
+        # This block is reset with clear()
+        self.album_name: str | None = None
+        self.artist: str | None = None
+        self.channel: str | None = None
+        self.duration: float | None = None
+        self.image_url: str | None = None
+        self.queue_position: int | None = None
+        self.queue_size: int | None = None
+        self.playlist_name: str | None = None
+        self.source_name: str | None = None
+        self.title: str | None = None
+        self.uri: str | None = None
+
+        self.position: float | None = None
+        self.position_updated_at: datetime.datetime | None = None
+
+    def clear(self) -> None:
+        """Clear basic media info."""
+        self.album_name = None
+        self.artist = None
+        self.channel = None
+        self.duration = None
+        self.image_url = None
+        self.playlist_name = None
+        self.queue_position = None
+        self.queue_size = None
+        self.source_name = None
+        self.title = None
+        self.uri = None
+
+    def clear_position(self) -> None:
+        """Clear the position attributes."""
+        self.position = None
+        self.position_updated_at = None
+
+    @property
+    def library(self) -> MusicLibrary:
+        """Return the soco MusicLibrary instance."""
+        return self.soco.music_library
+
+    @soco_error()
+    def poll_track_info(self) -> dict[str, Any]:
+        """Poll the speaker for current track info, add converted position values, and return."""
+        track_info = self.soco.get_current_track_info()
+        track_info[DURATION_SECONDS] = _timespan_secs(track_info.get("duration"))
+        track_info[POSITION_SECONDS] = _timespan_secs(track_info.get("position"))
+        return track_info
+
+    def write_media_player_states(self) -> None:
+        """Send a signal to media player(s) to write new states."""
+        dispatcher_send(self.hass, SONOS_MEDIA_UPDATED, self.soco.uid)
+
+    def set_basic_track_info(self, update_position: bool = False) -> None:
+        """Query the speaker to update media metadata and position info."""
+        self.clear()
+
+        track_info = self.poll_track_info()
+        self.uri = track_info["uri"]
+
+        audio_source = self.soco.music_source_from_uri(self.uri)
+        if source := SOURCE_MAPPING.get(audio_source):
+            self.source_name = source
+            if audio_source in LINEIN_SOURCES:
+                self.clear_position()
+                self.title = source
+                return
+
+        self.artist = track_info.get("artist")
+        self.album_name = track_info.get("album")
+        self.title = track_info.get("title")
+        self.image_url = track_info.get("album_art")
+
+        playlist_position = int(track_info.get("playlist_position"))
+        if playlist_position > 0:
+            self.queue_position = playlist_position
+
+        self.update_media_position(track_info, force_update=update_position)
+
+    def update_media_from_event(self, evars: dict[str, Any]) -> None:
+        """Update information about currently playing media using an event payload."""
+        new_status = evars["transport_state"]
+        state_changed = new_status != self.playback_status
+
+        self.play_mode = evars["current_play_mode"]
+        self.playback_status = new_status
+
+        track_uri = evars["enqueued_transport_uri"] or evars["current_track_uri"]
+        audio_source = self.soco.music_source_from_uri(track_uri)
+
+        self.set_basic_track_info(update_position=state_changed)
+
+        if ct_md := evars["current_track_meta_data"]:
+            if not self.image_url:
+                if album_art_uri := getattr(ct_md, "album_art_uri", None):
+                    self.image_url = self.library.build_album_art_full_uri(
+                        album_art_uri
+                    )
+
+        et_uri_md = evars["enqueued_transport_uri_meta_data"]
+        if isinstance(et_uri_md, DidlPlaylistContainer):
+            self.playlist_name = et_uri_md.title
+
+        if queue_size := evars.get("number_of_tracks", 0):
+            self.queue_size = int(queue_size)
+
+        if audio_source == MUSIC_SRC_RADIO:
+            self.channel = et_uri_md.title
+
+            if ct_md and ct_md.radio_show:
+                radio_show = ct_md.radio_show.split(",")[0]
+                self.channel = " • ".join(filter(None, [self.channel, radio_show]))
+
+            if isinstance(et_uri_md, DidlAudioBroadcast):
+                self.title = self.title or self.channel
+
+        self.write_media_player_states()
+
+    @soco_error()
+    def poll_media(self) -> None:
+        """Poll information about currently playing media."""
+        transport_info = self.soco.get_current_transport_info()
+        new_status = transport_info["current_transport_state"]
+
+        if new_status == SONOS_STATE_TRANSITIONING:
+            return
+
+        update_position = new_status != self.playback_status
+        self.playback_status = new_status
+        self.play_mode = self.soco.play_mode
+
+        self.set_basic_track_info(update_position=update_position)
+
+        self.write_media_player_states()
+
+    def update_media_position(
+        self, position_info: dict[str, int], force_update: bool = False
+    ) -> None:
+        """Update state when playing music tracks."""
+        if (duration := position_info.get(DURATION_SECONDS)) == 0:
+            self.clear_position()
+            return
+
+        should_update = force_update
+        self.duration = duration
+        current_position = position_info.get(POSITION_SECONDS)
+
+        # player started reporting position?
+        if current_position is not None and self.position is None:
+            should_update = True
+
+        # position jumped?
+        if current_position is not None and self.position is not None:
+            if self.playback_status == SONOS_STATE_PLAYING:
+                assert self.position_updated_at is not None
+                time_delta = dt_util.utcnow() - self.position_updated_at
+                time_diff = time_delta.total_seconds()
+            else:
+                time_diff = 0
+
+            calculated_position = self.position + time_diff
+
+            if abs(calculated_position - current_position) > 1.5:
+                should_update = True
+
+        if current_position is None:
+            self.clear_position()
+        elif should_update:
+            self.position = current_position
+            self.position_updated_at = dt_util.utcnow()

homeassistant/components/sonos/media_player.py

@@ -23,6 +23,7 @@ from homeassistant.components.media_player import (
     async_process_play_media_url,
 )
 from homeassistant.components.media_player.const import (
+    ATTR_INPUT_SOURCE,
     ATTR_MEDIA_ENQUEUE,
     MEDIA_TYPE_ALBUM,
     MEDIA_TYPE_ARTIST,
@@ -65,6 +66,7 @@ from .const import (
     MEDIA_TYPES_TO_SONOS,
     PLAYABLE_MEDIA_TYPES,
     SONOS_CREATE_MEDIA_PLAYER,
+    SONOS_MEDIA_UPDATED,
     SONOS_STATE_PLAYING,
     SONOS_STATE_TRANSITIONING,
     SOURCE_LINEIN,
@@ -255,6 +257,23 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
         self._attr_unique_id = self.soco.uid
         self._attr_name = self.speaker.zone_name
 
+    async def async_added_to_hass(self) -> None:
+        """Handle common setup when added to hass."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SONOS_MEDIA_UPDATED,
+                self.async_write_media_state,
+            )
+        )
+
+    @callback
+    def async_write_media_state(self, uid: str) -> None:
+        """Write media state if the provided UID is coordinator of this speaker."""
+        if self.coordinator.uid == uid:
+            self.async_write_ha_state()
+
     @property
     def coordinator(self) -> SonosSpeaker:
         """Return the current coordinator SonosSpeaker."""
@@ -295,7 +314,7 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
         self.speaker.update_groups()
         self.speaker.update_volume()
         if self.speaker.is_coordinator:
-            self.speaker.update_media()
+            self.media.poll_media()
 
     @property
     def volume_level(self) -> float | None:
@@ -660,6 +679,12 @@ class SonosMediaPlayerEntity(SonosEntity, MediaPlayerEntity):
         if self.media.queue_position is not None:
             attributes[ATTR_QUEUE_POSITION] = self.media.queue_position
 
+        if self.media.queue_size:
+            attributes["queue_size"] = self.media.queue_size
+
+        if self.source:
+            attributes[ATTR_INPUT_SOURCE] = self.source
+
         return attributes
 
     async def async_get_browse_image(

homeassistant/components/sonos/speaker.py

@@ -9,15 +9,12 @@ from functools import partial
 import logging
 import time
 from typing import Any
-import urllib.parse
 
 import async_timeout
 import defusedxml.ElementTree as ET
-from soco.core import MUSIC_SRC_LINE_IN, MUSIC_SRC_RADIO, MUSIC_SRC_TV, SoCo
-from soco.data_structures import DidlAudioBroadcast, DidlPlaylistContainer
+from soco.core import SoCo
 from soco.events_base import Event as SonosEvent, SubscriptionBase
 from soco.exceptions import SoCoException, SoCoUPnPException
-from soco.music_library import MusicLibrary
 from soco.plugins.plex import PlexPlugin
 from soco.plugins.sharelink import ShareLinkPlugin
 from soco.snapshot import Snapshot
@@ -58,12 +55,11 @@ from .const import (
     SONOS_STATE_TRANSITIONING,
     SONOS_STATE_UPDATED,
     SONOS_VANISHED,
-    SOURCE_LINEIN,
-    SOURCE_TV,
     SUBSCRIPTION_TIMEOUT,
 )
 from .favorites import SonosFavorites
 from .helpers import soco_error
+from .media import SonosMedia
 from .statistics import ActivityStatistics, EventStatistics
 
 NEVER_TIME = -1200.0
@@ -80,7 +76,6 @@ SUBSCRIPTION_SERVICES = [
     "zoneGroupTopology",
 ]
 SUPPORTED_VANISH_REASONS = ("sleeping", "upgrade")
-UNAVAILABLE_VALUES = {"", "NOT_IMPLEMENTED", None}
 UNUSED_DEVICE_KEYS = ["SPID", "TargetRoomName"]
 
 
@@ -97,57 +92,6 @@ def fetch_battery_info_or_none(soco: SoCo) -> dict[str, Any] | None:
         return soco.get_battery_info()
 
 
-def _timespan_secs(timespan: str | None) -> None | float:
-    """Parse a time-span into number of seconds."""
-    if timespan in UNAVAILABLE_VALUES:
-        return None
-
-    assert timespan is not None
-    return sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":"))))
-
-
-class SonosMedia:
-    """Representation of the current Sonos media."""
-
-    def __init__(self, soco: SoCo) -> None:
-        """Initialize a SonosMedia."""
-        self.library = MusicLibrary(soco)
-        self.play_mode: str | None = None
-        self.playback_status: str | None = None
-
-        self.album_name: str | None = None
-        self.artist: str | None = None
-        self.channel: str | None = None
-        self.duration: float | None = None
-        self.image_url: str | None = None
-        self.queue_position: int | None = None
-        self.playlist_name: str | None = None
-        self.source_name: str | None = None
-        self.title: str | None = None
-        self.uri: str | None = None
-
-        self.position: float | None = None
-        self.position_updated_at: datetime.datetime | None = None
-
-    def clear(self) -> None:
-        """Clear basic media info."""
-        self.album_name = None
-        self.artist = None
-        self.channel = None
-        self.duration = None
-        self.image_url = None
-        self.playlist_name = None
-        self.queue_position = None
-        self.source_name = None
-        self.title = None
-        self.uri = None
-
-    def clear_position(self) -> None:
-        """Clear the position attributes."""
-        self.position = None
-        self.position_updated_at = None
-
-
 class SonosSpeaker:
     """Representation of a Sonos speaker."""
 
@@ -158,7 +102,7 @@ class SonosSpeaker:
         self.hass = hass
         self.soco = soco
         self.household_id: str = soco.household_id
-        self.media = SonosMedia(soco)
+        self.media = SonosMedia(hass, soco)
         self._plex_plugin: PlexPlugin | None = None
         self._share_link_plugin: ShareLinkPlugin | None = None
         self.available = True
@@ -512,7 +456,18 @@ class SonosSpeaker:
         if crossfade := event.variables.get("current_crossfade_mode"):
             self.cross_fade = bool(int(crossfade))
 
-        self.hass.async_add_executor_job(self.update_media, event)
+        # Missing transport_state indicates a transient error
+        if (new_status := event.variables.get("transport_state")) is None:
+            return
+
+        # Ignore transitions, we should get the target state soon
+        if new_status == SONOS_STATE_TRANSITIONING:
+            return
+
+        self.event_stats.process(event)
+        self.hass.async_add_executor_job(
+            self.media.update_media_from_event, event.variables
+        )
 
     @callback
     def async_update_volume(self, event: SonosEvent) -> None:
@@ -1064,176 +1019,3 @@ class SonosSpeaker:
         """Update information about current volume settings."""
         self.volume = self.soco.volume
         self.muted = self.soco.mute
-
-    @soco_error()
-    def update_media(self, event: SonosEvent | None = None) -> None:
-        """Update information about currently playing media."""
-        variables = event.variables if event else {}
-
-        if "transport_state" in variables:
-            # If the transport has an error then transport_state will
-            # not be set
-            new_status = variables["transport_state"]
-        else:
-            transport_info = self.soco.get_current_transport_info()
-            new_status = transport_info["current_transport_state"]
-
-        # Ignore transitions, we should get the target state soon
-        if new_status == SONOS_STATE_TRANSITIONING:
-            return
-
-        if event:
-            self.event_stats.process(event)
-
-        self.media.clear()
-        update_position = new_status != self.media.playback_status
-        self.media.playback_status = new_status
-
-        if "transport_state" in variables:
-            self.media.play_mode = variables["current_play_mode"]
-            track_uri = (
-                variables["enqueued_transport_uri"] or variables["current_track_uri"]
-            )
-            music_source = self.soco.music_source_from_uri(track_uri)
-            if uri_meta_data := variables.get("enqueued_transport_uri_meta_data"):
-                if isinstance(uri_meta_data, DidlPlaylistContainer):
-                    self.media.playlist_name = uri_meta_data.title
-        else:
-            self.media.play_mode = self.soco.play_mode
-            music_source = self.soco.music_source
-
-        if music_source == MUSIC_SRC_TV:
-            self.update_media_linein(SOURCE_TV)
-        elif music_source == MUSIC_SRC_LINE_IN:
-            self.update_media_linein(SOURCE_LINEIN)
-        else:
-            track_info = self.soco.get_current_track_info()
-            if not track_info["uri"]:
-                self.media.clear_position()
-            else:
-                self.media.uri = track_info["uri"]
-                self.media.artist = track_info.get("artist")
-                self.media.album_name = track_info.get("album")
-                self.media.title = track_info.get("title")
-
-                if music_source == MUSIC_SRC_RADIO:
-                    self.update_media_radio(variables)
-                else:
-                    self.update_media_music(track_info)
-                self.update_media_position(update_position, track_info)
-
-        self.write_entity_states()
-
-        # Also update slaves
-        speakers = self.hass.data[DATA_SONOS].discovered.values()
-        for speaker in speakers:
-            if speaker.coordinator == self:
-                speaker.write_entity_states()
-
-    def update_media_linein(self, source: str) -> None:
-        """Update state when playing from line-in/tv."""
-        self.media.clear_position()
-
-        self.media.title = source
-        self.media.source_name = source
-
-    def update_media_radio(self, variables: dict) -> None:
-        """Update state when streaming radio."""
-        self.media.clear_position()
-        radio_title = None
-
-        if current_track_metadata := variables.get("current_track_meta_data"):
-            if album_art_uri := getattr(current_track_metadata, "album_art_uri", None):
-                self.media.image_url = self.media.library.build_album_art_full_uri(
-                    album_art_uri
-                )
-            if not self.media.artist:
-                self.media.artist = getattr(current_track_metadata, "creator", None)
-
-            # A missing artist implies metadata is incomplete, try a different method
-            if not self.media.artist:
-                radio_show = None
-                stream_content = None
-                if current_track_metadata.radio_show:
-                    radio_show = current_track_metadata.radio_show.split(",")[0]
-                if not current_track_metadata.stream_content.startswith(
-                    ("ZPSTR_", "TYPE=")
-                ):
-                    stream_content = current_track_metadata.stream_content
-                radio_title = " • ".join(filter(None, [radio_show, stream_content]))
-
-        if radio_title:
-            # Prefer the radio title created above
-            self.media.title = radio_title
-        elif uri_meta_data := variables.get("enqueued_transport_uri_meta_data"):
-            if isinstance(uri_meta_data, DidlAudioBroadcast) and (
-                self.soco.music_source_from_uri(self.media.title) == MUSIC_SRC_RADIO
-                or (
-                    isinstance(self.media.title, str)
-                    and isinstance(self.media.uri, str)
-                    and (
-                        self.media.title in self.media.uri
-                        or self.media.title in urllib.parse.unquote(self.media.uri)
-                    )
-                )
-            ):
-                # Fall back to the radio channel name as a last resort
-                self.media.title = uri_meta_data.title
-
-        media_info = self.soco.get_current_media_info()
-        self.media.channel = media_info["channel"]
-
-        # Check if currently playing radio station is in favorites
-        fav = next(
-            (
-                fav
-                for fav in self.favorites
-                if fav.reference.get_uri() == media_info["uri"]
-            ),
-            None,
-        )
-        if fav:
-            self.media.source_name = fav.title
-
-    def update_media_music(self, track_info: dict) -> None:
-        """Update state when playing music tracks."""
-        self.media.image_url = track_info.get("album_art")
-
-        playlist_position = int(track_info.get("playlist_position"))  # type: ignore
-        if playlist_position > 0:
-            self.media.queue_position = playlist_position - 1
-
-    def update_media_position(
-        self, update_media_position: bool, track_info: dict
-    ) -> None:
-        """Update state when playing music tracks."""
-        self.media.duration = _timespan_secs(track_info.get("duration"))
-        current_position = _timespan_secs(track_info.get("position"))
-
-        if self.media.duration == 0:
-            self.media.clear_position()
-            return
-
-        # player started reporting position?
-        if current_position is not None and self.media.position is None:
-            update_media_position = True
-
-        # position jumped?
-        if current_position is not None and self.media.position is not None:
-            if self.media.playback_status == SONOS_STATE_PLAYING:
-                assert self.media.position_updated_at is not None
-                time_delta = dt_util.utcnow() - self.media.position_updated_at
-                time_diff = time_delta.total_seconds()
-            else:
-                time_diff = 0
-
-            calculated_position = self.media.position + time_diff
-
-            if abs(calculated_position - current_position) > 1.5:
-                update_media_position = True
-
-        if current_position is None:
-            self.media.clear_position()
-        elif update_media_position:
-            self.media.position = current_position
-            self.media.position_updated_at = dt_util.utcnow()

tests/components/sonos/conftest.py

@@ -284,9 +284,11 @@ def no_media_event_fixture(soco):
         "current_crossfade_mode": "0",
         "current_play_mode": "NORMAL",
         "current_section": "0",
+        "current_track_meta_data": "",
         "current_track_uri": "",
         "enqueued_transport_uri": "",
         "enqueued_transport_uri_meta_data": "",
+        "number_of_tracks": "0",
         "transport_state": "STOPPED",
     }
     return SonosMockEvent(soco, soco.avTransport, variables)

tests/components/sonos/test_speaker.py

@@ -19,9 +19,7 @@ async def test_fallback_to_polling(
     caplog.clear()
 
     # Ensure subscriptions are cancelled and polling methods are called when subscriptions time out
-    with patch(
-        "homeassistant.components.sonos.speaker.SonosSpeaker.update_media"
-    ), patch(
+    with patch("homeassistant.components.sonos.media.SonosMedia.poll_media"), patch(
         "homeassistant.components.sonos.speaker.SonosSpeaker.subscription_address"
     ):
         async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)