/
/
/
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
8from music_assistant_models.errors import AudioError
9
10if TYPE_CHECKING:
11 from collections.abc import AsyncGenerator
12
13 from music_assistant_models.enums import MediaType
14 from music_assistant_models.media_items import AudioFormat
15 from music_assistant_models.streamdetails import StreamDetails
16
17 from music_assistant.helpers.json import SerializableType
18 from music_assistant.providers.spotify.provider import SpotifyProvider
19
20
21class StreamSupersededError(AudioError):
22 """
23 Raised when Music Assistant replaced the stream that was delivering an item.
24
25 A backend that serves a queue from one session cuts the stream of an item it
26 is asked to serve from a new one - a seek - and refuses it the session again,
27 whichever part of the item it comes back for: nothing beyond the cut is that
28 stream's to deliver.
29 """
30
31
32class SpotifyPlaybackBackend(ABC):
33 """
34 One way to fetch Spotify audio, item by item.
35
36 The SpotifyProvider owns everything catalog/Web API related; a backend owns
37 the playback session and the per-item audio fetch. All URIs passed to a
38 backend use Spotify's canonical form (``spotify:track:<id>`` /
39 ``spotify:episode:<id>``).
40 """
41
42 def __init__(self, provider: SpotifyProvider) -> None:
43 """
44 Initialize the backend (cheap; real setup happens in ``setup``).
45
46 :param provider: The owning Spotify provider instance.
47 """
48 self.provider = provider
49 self.mass = provider.mass
50 self.logger = provider.logger
51
52 @abstractmethod
53 def source_audio_format(self, media_type: MediaType) -> AudioFormat:
54 """
55 Return the format of the Spotify source, for StreamDetails and display.
56
57 :param media_type: What is being streamed; Spotify serves music and
58 spoken content at different qualities.
59 """
60
61 @property
62 def handoff_audio_format(self) -> AudioFormat | None:
63 """
64 Return the format this backend actually hands over, when it differs.
65
66 None means the source arrives untouched, so the source format describes
67 the bytes as well.
68 """
69 return None
70
71 @property
72 def is_realtime(self) -> bool:
73 """Return whether this backend delivers audio at playback pace (no read-ahead)."""
74 return False
75
76 @abstractmethod
77 async def setup(self) -> None:
78 """
79 Validate availability and prepare the playback session.
80
81 :raises LoginFailed: When the stored playback authorization is missing or
82 unusable, requiring the user to re-run the setup flow.
83 """
84
85 async def unload(self) -> None: # noqa: B027
86 """Release any resources held by the backend."""
87
88 @abstractmethod
89 def stream_spotify_uri(
90 self,
91 spotify_uri: str,
92 seek_position: int = 0,
93 *,
94 streamdetails: StreamDetails | None = None,
95 continuation: bool = False,
96 ) -> AsyncGenerator[bytes]:
97 """
98 Yield the audio for one Spotify URI in this backend's audio format.
99
100 :param spotify_uri: Canonical Spotify URI (``spotify:track:<id>`` or
101 ``spotify:episode:<id>``).
102 :param seek_position: Position in seconds to start from.
103 :param streamdetails: The StreamDetails the audio is requested for.
104 Backends that fetch each item on its own ignore these; a backend
105 that keeps one session needs them to know which queue and which
106 item of it this audio belongs to.
107 :param continuation: Whether this URI continues an item stream already
108 under way - a later chapter of the audiobook being streamed - rather
109 than starting one.
110 :raises StreamSupersededError: When Music Assistant took the item off
111 this stream, leaving it nothing to deliver.
112 """
113
114 async def get_diagnostics(self) -> dict[str, SerializableType]:
115 """Return diagnostic details about the backend (never any secret)."""
116 return {}
117