Add audio feature sensors to Spotify (#128785)

homeassistant/components/spotify/__init__.py

@@ -29,7 +29,7 @@ from .util import (
     spotify_uri_from_media_browser_url,
 )
 
-PLATFORMS = [Platform.MEDIA_PLAYER]
+PLATFORMS = [Platform.MEDIA_PLAYER, Platform.SENSOR]
 
 __all__ = [
     "async_browse_media",

homeassistant/components/spotify/coordinator.py

@@ -6,12 +6,14 @@ import logging
 
 from spotifyaio import (
     ContextType,
+    ItemType,
     PlaybackState,
     Playlist,
     SpotifyClient,
     SpotifyConnectionError,
     UserProfile,
 )
+from spotifyaio.models import AudioFeatures
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@@ -29,6 +31,7 @@ class SpotifyCoordinatorData:
     current_playback: PlaybackState | None
     position_updated_at: datetime | None
     playlist: Playlist | None
+    audio_features: AudioFeatures | None
     dj_playlist: bool = False
 
 
@@ -53,6 +56,7 @@ class SpotifyCoordinator(DataUpdateCoordinator[SpotifyCoordinatorData]):
         )
         self.client = client
         self._playlist: Playlist | None = None
+        self._currently_loaded_track: str | None = None
 
     async def _async_setup(self) -> None:
         """Set up the coordinator."""
@@ -65,12 +69,22 @@ class SpotifyCoordinator(DataUpdateCoordinator[SpotifyCoordinatorData]):
         current = await self.client.get_playback()
         if not current:
             return SpotifyCoordinatorData(
-                current_playback=None, position_updated_at=None, playlist=None
+                current_playback=None,
+                position_updated_at=None,
+                playlist=None,
+                audio_features=None,
             )
         # Record the last updated time, because Spotify's timestamp property is unreliable
         # and doesn't actually return the fetch time as is mentioned in the API description
         position_updated_at = dt_util.utcnow()
 
+        audio_features: AudioFeatures | None = None
+        if (item := current.item) is not None and item.type == ItemType.TRACK:
+            if item.uri != self._currently_loaded_track:
+                self._currently_loaded_track = item.uri
+                audio_features = await self.client.get_audio_features(item.uri)
+            else:
+                audio_features = self.data.audio_features
         dj_playlist = False
         if (context := current.context) is not None:
             if self._playlist is None or self._playlist.uri != context.uri:
@@ -93,5 +107,6 @@ class SpotifyCoordinator(DataUpdateCoordinator[SpotifyCoordinatorData]):
             current_playback=current,
             position_updated_at=position_updated_at,
             playlist=self._playlist,
+            audio_features=audio_features,
             dj_playlist=dj_playlist,
         )

homeassistant/components/spotify/sensor.py

@@ -0,0 +1,85 @@
+"""Sensor platform for Spotify."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from spotifyaio.models import AudioFeatures
+
+from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import DOMAIN, SpotifyConfigEntry
+from .coordinator import SpotifyCoordinator
+
+
+@dataclass(frozen=True, kw_only=True)
+class SpotifyAudioFeaturesSensorEntityDescription(SensorEntityDescription):
+    """Describes Spotify sensor entity."""
+
+    value_fn: Callable[[AudioFeatures], float]
+
+
+AUDIO_FEATURE_SENSORS: tuple[SpotifyAudioFeaturesSensorEntityDescription, ...] = (
+    SpotifyAudioFeaturesSensorEntityDescription(
+        key="bpm",
+        translation_key="song_tempo",
+        native_unit_of_measurement="bpm",
+        suggested_display_precision=0,
+        value_fn=lambda audio_features: audio_features.tempo,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: SpotifyConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Spotify sensor based on a config entry."""
+    coordinator = entry.runtime_data.coordinator
+
+    user_id = entry.unique_id
+
+    assert user_id is not None
+
+    async_add_entities(
+        SpotifyAudioFeatureSensor(coordinator, description, user_id, entry.title)
+        for description in AUDIO_FEATURE_SENSORS
+    )
+
+
+class SpotifyAudioFeatureSensor(CoordinatorEntity[SpotifyCoordinator], SensorEntity):
+    """Representation of a Spotify sensor."""
+
+    _attr_has_entity_name = True
+    entity_description: SpotifyAudioFeaturesSensorEntityDescription
+
+    def __init__(
+        self,
+        coordinator: SpotifyCoordinator,
+        entity_description: SpotifyAudioFeaturesSensorEntityDescription,
+        user_id: str,
+        name: str,
+    ) -> None:
+        """Initialize."""
+        super().__init__(coordinator)
+        self._attr_unique_id = f"{user_id}_{entity_description.key}"
+        self.entity_description = entity_description
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, user_id)},
+            manufacturer="Spotify AB",
+            model=f"Spotify {coordinator.current_user.product}",
+            name=f"Spotify {name}",
+            entry_type=DeviceEntryType.SERVICE,
+            configuration_url="https://open.spotify.com",
+        )
+
+    @property
+    def native_value(self) -> float | None:
+        """Return the state of the sensor."""
+        if (audio_features := self.coordinator.data.audio_features) is None:
+            return None
+        return self.entity_description.value_fn(audio_features)

homeassistant/components/spotify/strings.json

@@ -30,5 +30,12 @@
     "info": {
       "api_endpoint_reachable": "Spotify API endpoint reachable"
     }
+  },
+  "entity": {
+    "sensor": {
+      "song_tempo": {
+        "name": "Song tempo"
+      }
+    }
   }
 }

tests/components/spotify/conftest.py

@@ -9,6 +9,7 @@ from spotifyaio.models import (
     Album,
     Artist,
     ArtistResponse,
+    AudioFeatures,
     CategoriesResponse,
     Category,
     CategoryPlaylistResponse,
@@ -132,6 +133,7 @@ def mock_spotify() -> Generator[AsyncMock]:
             ("album.json", "get_album", Album),
             ("artist.json", "get_artist", Artist),
             ("show.json", "get_show", Show),
+            ("audio_features.json", "get_audio_features", AudioFeatures),
         ):
             getattr(client, method).return_value = obj.from_json(
                 load_fixture(fixture, DOMAIN)

tests/components/spotify/fixtures/audio_features.json

@@ -0,0 +1,20 @@
+{
+  "danceability": 0.696,
+  "energy": 0.905,
+  "key": 2,
+  "loudness": -2.743,
+  "mode": 1,
+  "speechiness": 0.103,
+  "acousticness": 0.011,
+  "instrumentalness": 0.000905,
+  "liveness": 0.302,
+  "valence": 0.625,
+  "tempo": 114.944,
+  "type": "audio_features",
+  "id": "11dFghVXANMlKmJXsNCbNl",
+  "uri": "spotify:track:11dFghVXANMlKmJXsNCbNl",
+  "track_href": "https://api.spotify.com/v1/tracks/11dFghVXANMlKmJXsNCbNl",
+  "analysis_url": "https://api.spotify.com/v1/audio-analysis/11dFghVXANMlKmJXsNCbNl",
+  "duration_ms": 207960,
+  "time_signature": 4
+}

tests/components/spotify/snapshots/test_diagnostics.ambr

@@ -14,6 +14,20 @@
       }),
     ]),
     'playback': dict({
+      'audio_features': dict({
+        'acousticness': 0.011,
+        'danceability': 0.696,
+        'energy': 0.905,
+        'instrumentalness': 0.000905,
+        'key': 2,
+        'liveness': 0.302,
+        'loudness': -2.743,
+        'mode': 1,
+        'speechiness': 0.103,
+        'tempo': 114.944,
+        'time_signature': 4,
+        'valence': 0.625,
+      }),
       'current_playback': dict({
         'context': dict({
           'context_type': 'playlist',

tests/components/spotify/snapshots/test_sensor.ambr

@@ -0,0 +1,51 @@
+# serializer version: 1
+# name: test_entities[sensor.spotify_spotify_1_song_tempo-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'sensor',
+    'entity_category': None,
+    'entity_id': 'sensor.spotify_spotify_1_song_tempo',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+      'sensor': dict({
+        'suggested_display_precision': 0,
+      }),
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Song tempo',
+    'platform': 'spotify',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'song_tempo',
+    'unique_id': '1112264111_bpm',
+    'unit_of_measurement': 'bpm',
+  })
+# ---
+# name: test_entities[sensor.spotify_spotify_1_song_tempo-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'Spotify spotify_1 Song tempo',
+      'unit_of_measurement': 'bpm',
+    }),
+    'context': <ANY>,
+    'entity_id': 'sensor.spotify_spotify_1_song_tempo',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': '114.944',
+  })
+# ---

tests/components/spotify/test_media_player.py

@@ -45,6 +45,7 @@ from homeassistant.const import (
     SERVICE_SHUFFLE_SET,
     SERVICE_VOLUME_SET,
     STATE_UNAVAILABLE,
+    Platform,
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
@@ -70,7 +71,10 @@ async def test_entities(
 ) -> None:
     """Test the Spotify entities."""
     freezer.move_to("2023-10-21")
-    with patch("secrets.token_hex", return_value="mock-token"):
+    with (
+        patch("secrets.token_hex", return_value="mock-token"),
+        patch("homeassistant.components.spotify.PLATFORMS", [Platform.MEDIA_PLAYER]),
+    ):
         await setup_integration(hass, mock_config_entry)
 
         await snapshot_platform(
@@ -92,7 +96,10 @@ async def test_podcast(
     mock_spotify.return_value.get_playback.return_value = PlaybackState.from_json(
         load_fixture("playback_episode.json", DOMAIN)
     )
-    with patch("secrets.token_hex", return_value="mock-token"):
+    with (
+        patch("secrets.token_hex", return_value="mock-token"),
+        patch("homeassistant.components.spotify.PLATFORMS", [Platform.MEDIA_PLAYER]),
+    ):
         await setup_integration(hass, mock_config_entry)
 
         await snapshot_platform(

tests/components/spotify/test_sensor.py

@@ -0,0 +1,65 @@
+"""Tests for the Spotify sensor platform."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from spotifyaio import PlaybackState
+from syrupy import SnapshotAssertion
+
+from homeassistant.components.spotify import DOMAIN
+from homeassistant.const import STATE_UNKNOWN, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from . import setup_integration
+
+from tests.common import MockConfigEntry, load_fixture, snapshot_platform
+
+
+@pytest.mark.usefixtures("setup_credentials")
+async def test_entities(
+    hass: HomeAssistant,
+    mock_spotify: MagicMock,
+    mock_config_entry: MockConfigEntry,
+    entity_registry: er.EntityRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the Spotify entities."""
+    with patch("homeassistant.components.spotify.PLATFORMS", [Platform.SENSOR]):
+        await setup_integration(hass, mock_config_entry)
+
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+
+
+@pytest.mark.usefixtures("setup_credentials")
+async def test_audio_features_unavailable(
+    hass: HomeAssistant,
+    mock_spotify: MagicMock,
+    mock_config_entry: MockConfigEntry,
+    entity_registry: er.EntityRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the Spotify entities."""
+    mock_spotify.return_value.get_audio_features.return_value = None
+
+    await setup_integration(hass, mock_config_entry)
+
+    assert hass.states.get("sensor.spotify_spotify_1_song_tempo").state == STATE_UNKNOWN
+
+
+@pytest.mark.usefixtures("setup_credentials")
+async def test_audio_features_unknown_during_podcast(
+    hass: HomeAssistant,
+    mock_spotify: MagicMock,
+    mock_config_entry: MockConfigEntry,
+    entity_registry: er.EntityRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the Spotify audio features sensor during a podcast."""
+    mock_spotify.return_value.get_playback.return_value = PlaybackState.from_json(
+        load_fixture("playback_episode.json", DOMAIN)
+    )
+
+    await setup_integration(hass, mock_config_entry)
+
+    assert hass.states.get("sensor.spotify_spotify_1_song_tempo").state == STATE_UNKNOWN