music-assistant-server

1"""Playback backend contract for the Spotify music provider."""
2
3from __future__ import annotations
4
5from abc import ABC, abstractmethod
6from typing import TYPE_CHECKING
7
8if TYPE_CHECKING:
9    from collections.abc import AsyncGenerator
10
11    from music_assistant_models.enums import MediaType
12    from music_assistant_models.media_items import AudioFormat
13    from music_assistant_models.streamdetails import StreamDetails
14
15    from music_assistant.helpers.json import SerializableType
16    from music_assistant.providers.spotify.provider import SpotifyProvider
17
18
19class SpotifyPlaybackBackend(ABC):
20    """
21    One way to fetch Spotify audio, item by item.
22
23    The SpotifyProvider owns everything catalog/Web API related; a backend owns
24    the playback session and the per-item audio fetch. All URIs passed to a
25    backend use Spotify's canonical form (``spotify:track:<id>`` /
26    ``spotify:episode:<id>``).
27    """
28
29    def __init__(self, provider: SpotifyProvider) -> None:
30        """
31        Initialize the backend (cheap; real setup happens in ``setup``).
32
33        :param provider: The owning Spotify provider instance.
34        """
35        self.provider = provider
36        self.mass = provider.mass
37        self.logger = provider.logger
38
39    @abstractmethod
40    def source_audio_format(self, media_type: MediaType) -> AudioFormat:
41        """
42        Return the format of the Spotify source, for StreamDetails and display.
43
44        :param media_type: What is being streamed; Spotify serves music and
45            spoken content at different qualities.
46        """
47
48    @property
49    def handoff_audio_format(self) -> AudioFormat | None:
50        """
51        Return the format this backend actually hands over, when it differs.
52
53        None means the source arrives untouched, so the source format describes
54        the bytes as well.
55        """
56        return None
57
58    @property
59    def is_realtime(self) -> bool:
60        """Return whether this backend delivers audio at playback pace (no read-ahead)."""
61        return False
62
63    @abstractmethod
64    async def setup(self) -> None:
65        """
66        Validate availability and prepare the playback session.
67
68        :raises LoginFailed: When the stored playback authorization is missing or
69            unusable, requiring the user to re-run the setup flow.
70        """
71
72    async def unload(self) -> None:  # noqa: B027
73        """Release any resources held by the backend."""
74
75    @abstractmethod
76    def stream_spotify_uri(
77        self,
78        spotify_uri: str,
79        seek_position: int = 0,
80        *,
81        streamdetails: StreamDetails | None = None,
82    ) -> AsyncGenerator[bytes]:
83        """
84        Yield the audio for one Spotify URI in this backend's audio format.
85
86        :param spotify_uri: Canonical Spotify URI (``spotify:track:<id>`` or
87            ``spotify:episode:<id>``).
88        :param seek_position: Position in seconds to start from.
89        :param streamdetails: The StreamDetails the audio is requested for.
90            Backends that fetch each item on its own ignore these; a backend
91            that keeps one session needs them to know which queue and which
92            item of it this audio belongs to.
93        """
94
95    async def get_diagnostics(self) -> dict[str, SerializableType]:
96        """Return diagnostic details about the backend (never any secret)."""
97        return {}
98