coordinator.py

Go to the documentation of this file.

"""Coordinator for Spotify."""
 
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
from typing import TYPE_CHECKING
 
from spotifyaio import (
    ContextType,
    PlaybackState,
    Playlist,
    SpotifyClient,
    SpotifyConnectionError,
    SpotifyNotFoundError,
    UserProfile,
)
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
import homeassistant.util.dt as dt_util
 
from .const import DOMAIN
 
if TYPE_CHECKING:
    from .models import SpotifyData
 
_LOGGER = logging.getLogger(__name__)
 
 
type SpotifyConfigEntry = ConfigEntry[SpotifyData]
 
 
@dataclass
class SpotifyCoordinatorData:
    """Class to hold Spotify data."""
 
    current_playback: PlaybackState | None
    position_updated_at: datetime | None
    playlist: Playlist | None
    dj_playlist: bool = False
 
 
# This is a minimal representation of the DJ playlist that Spotify now offers
# The DJ is not fully integrated with the playlist API, so we need to guard
# against trying to fetch it as a regular playlist
SPOTIFY_DJ_PLAYLIST_URI = "spotify:playlist:37i9dQZF1EYkqdzj48dyYq"
 
 
class SpotifyCoordinator(DataUpdateCoordinator[SpotifyCoordinatorData]):
    """Class to manage fetching Spotify data."""
 
    current_user: UserProfile
    config_entry: SpotifyConfigEntry
 
    def __init__(self, hass: HomeAssistant, client: SpotifyClient) -> None:
        """Initialize."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=30),
        )
        self.clientclient = client
        self._playlist_playlist: Playlist | None = None
        self._checked_playlist_id_checked_playlist_id: str | None = None
 
    async def _async_setup(self) -> None:
        """Set up the coordinator."""
        try:
            self.current_usercurrent_user = await self.clientclient.get_current_user()
        except SpotifyConnectionError as err:
            raise UpdateFailed("Error communicating with Spotify API") from err
 
    async def _async_update_data(self) -> SpotifyCoordinatorData:
        try:
            current = await self.clientclient.get_playback()
        except SpotifyConnectionError as err:
            raise UpdateFailed("Error communicating with Spotify API") from err
        if not current:
            return SpotifyCoordinatorData(
                current_playback=None,
                position_updated_at=None,
                playlist=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()
 
        dj_playlist = False
        if (context := current.context) is not None:
            dj_playlist = context.uri == SPOTIFY_DJ_PLAYLIST_URI
            if not (
                context.uri
                in (
                    self._checked_playlist_id_checked_playlist_id,
                    SPOTIFY_DJ_PLAYLIST_URI,
                )
                or (self._playlist_playlist is None and context.uri == self._checked_playlist_id_checked_playlist_id)
            ):
                self._checked_playlist_id_checked_playlist_id = context.uri
                self._playlist_playlist = None
                if context.context_type == ContextType.PLAYLIST:
                    # Make sure any playlist lookups don't break the current
                    # playback state update
                    try:
                        self._playlist_playlist = await self.clientclient.get_playlist(context.uri)
                    except SpotifyNotFoundError:
                        _LOGGER.debug(
                            "Spotify playlist '%s' not found. "
                            "Most likely a Spotify-created playlist",
                            context.uri,
                        )
                        self._playlist_playlist = None
                    except SpotifyConnectionError:
                        _LOGGER.debug(
                            "Unable to load spotify playlist '%s'. "
                            "Continuing without playlist data",
                            context.uri,
                        )
                        self._playlist_playlist = None
                        self._checked_playlist_id_checked_playlist_id = None
        return SpotifyCoordinatorData(
            current_playback=current,
            position_updated_at=position_updated_at,
            playlist=self._playlist_playlist,
            dj_playlist=dj_playlist,
        )