music-assistant-server
12.1 KB•PY
__init__.py
12.1 KB • 357 lines • python
1"""SiriusXM Music Provider for Music Assistant."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Sequence
6from typing import TYPE_CHECKING, Any
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
9from music_assistant_models.enums import (
10 ConfigEntryType,
11 ContentType,
12 ImageType,
13 LinkType,
14 MediaType,
15 ProviderFeature,
16 StreamType,
17)
18from music_assistant_models.errors import LoginFailed, MediaNotFoundError
19from music_assistant_models.media_items import (
20 AudioFormat,
21 BrowseFolder,
22 ItemMapping,
23 MediaItemImage,
24 MediaItemLink,
25 MediaItemType,
26 ProviderMapping,
27 Radio,
28 UniqueList,
29)
30from music_assistant_models.streamdetails import StreamDetails
31from tenacity import RetryError
32
33from music_assistant.controllers.cache import use_cache
34from music_assistant.helpers.util import select_free_port
35from music_assistant.helpers.webserver import Webserver
36from music_assistant.models.music_provider import MusicProvider
37
38if TYPE_CHECKING:
39 from music_assistant_models.config_entries import ProviderConfig
40 from music_assistant_models.provider import ProviderManifest
41
42 from music_assistant import MusicAssistant
43 from music_assistant.models import ProviderInstanceType
44
45import sxm.http
46from sxm import SXMClientAsync
47from sxm.models import QualitySize, RegionChoice, XMChannel, XMLiveChannel
48
49CONF_SXM_USERNAME = "sxm_email_address"
50CONF_SXM_PASSWORD = "sxm_password"
51CONF_SXM_REGION = "sxm_region"
52
53SUPPORTED_FEATURES = {
54 ProviderFeature.BROWSE,
55 ProviderFeature.LIBRARY_RADIOS,
56}
57
58
59async def setup(
60 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
61) -> ProviderInstanceType:
62 """Initialize provider(instance) with given configuration."""
63 return SiriusXMProvider(mass, manifest, config, SUPPORTED_FEATURES)
64
65
66async def get_config_entries(
67 mass: MusicAssistant,
68 instance_id: str | None = None,
69 action: str | None = None,
70 values: dict[str, ConfigValueType] | None = None,
71) -> tuple[ConfigEntry, ...]:
72 """
73 Return Config entries to setup this provider.
74
75 instance_id: id of an existing provider instance (None if new instance setup).
76 action: [optional] action key called from config entries UI.
77 values: the (intermediate) raw values for config entries sent with the action.
78 """
79 # ruff: noqa: ARG001
80 return (
81 ConfigEntry(
82 key=CONF_SXM_USERNAME,
83 type=ConfigEntryType.STRING,
84 label="Username",
85 required=True,
86 ),
87 ConfigEntry(
88 key=CONF_SXM_PASSWORD,
89 type=ConfigEntryType.SECURE_STRING,
90 label="Password",
91 required=True,
92 ),
93 ConfigEntry(
94 key=CONF_SXM_REGION,
95 type=ConfigEntryType.STRING,
96 default_value="US",
97 options=[
98 ConfigValueOption(title="United States", value="US"),
99 ConfigValueOption(title="Canada", value="CA"),
100 ],
101 label="Region",
102 required=True,
103 ),
104 )
105
106
107class SiriusXMProvider(MusicProvider):
108 """SiriusXM Music Provider."""
109
110 _username: str
111 _password: str
112 _region: str
113 _client: SXMClientAsync
114
115 _channels: list[XMChannel]
116
117 _sxm_server: Webserver
118 _base_url: str
119
120 _current_stream_details: StreamDetails | None = None
121
122 async def handle_async_init(self) -> None:
123 """Handle async initialization of the provider."""
124 username = self.config.get_value(CONF_SXM_USERNAME)
125 assert isinstance(username, str) # for type checker
126 password = self.config.get_value(CONF_SXM_PASSWORD)
127 assert isinstance(password, str) # for type checker
128
129 region: RegionChoice = (
130 RegionChoice.US if self.config.get_value(CONF_SXM_REGION) == "US" else RegionChoice.CA
131 )
132
133 self._client = SXMClientAsync(
134 username,
135 password,
136 region,
137 quality=QualitySize.LARGE_256k,
138 update_handler=self._channel_updated,
139 )
140
141 self.logger.info("Authenticating with SiriusXM")
142 try:
143 if not await self._client.authenticate():
144 raise LoginFailed("Could not login to SiriusXM")
145 except RetryError:
146 # It looks like there's a bug in the sxm-client code
147 # where it won't return False if there's bad credentials.
148 # Due to the retry logic, it's attempting to log in multiple
149 # times and then finally raises an unrelated exception,
150 # rather than returning False or raising the package's
151 # AuthenticationError.
152 # Therefore, we're resorting to catching the RetryError
153 # here and recognizing it as a login failure.
154 raise LoginFailed("Could not login to SiriusXM")
155
156 self.logger.info("Successfully authenticated")
157
158 await self._refresh_channels()
159
160 # Set up the sxm server for streaming
161 bind_ip = "127.0.0.1"
162 bind_port = await select_free_port(8100, 9999)
163
164 self._base_url = f"{bind_ip}:{bind_port}"
165 http_handler = sxm.http.make_http_handler(self._client)
166
167 self._sxm_server = Webserver(self.logger)
168
169 await self._sxm_server.setup(
170 bind_ip=bind_ip,
171 bind_port=bind_port,
172 base_url=self._base_url,
173 static_routes=[
174 ("*", "/{tail:.*}", http_handler),
175 ],
176 )
177
178 self.logger.debug(f"SXM Proxy server running at {bind_ip}:{bind_port}")
179
180 async def unload(self, is_removed: bool = False) -> None:
181 """
182 Handle unload/close of the provider.
183
184 Called when provider is deregistered (e.g. MA exiting or config reloading).
185 """
186 await self._sxm_server.close()
187
188 @property
189 def is_streaming_provider(self) -> bool:
190 """
191 Return True if the provider is a streaming provider.
192
193 This literally means that the catalog is not the same as the library contents.
194 For local based providers (files, plex), the catalog is the same as the library content.
195 It also means that data is if this provider is NOT a streaming provider,
196 data cross instances is unique, the catalog and library differs per instance.
197
198 Setting this to True will only query one instance of the provider for search and lookups.
199 Setting this to False will query all instances of this provider for search and lookups.
200 """
201 return True
202
203 async def get_library_radios(self) -> AsyncGenerator[Radio, None]:
204 """Retrieve library/subscribed radio stations from the provider."""
205 for channel in self._channels_by_id.values():
206 if channel.is_favorite:
207 yield self._parse_radio(channel)
208
209 @use_cache(3600 * 24 * 14) # Cache for 14 days
210 async def get_radio(self, prov_radio_id: str) -> Radio:
211 """Get full radio details by id."""
212 if prov_radio_id not in self._channels_by_id:
213 raise MediaNotFoundError("Station not found")
214
215 return self._parse_radio(self._channels_by_id[prov_radio_id])
216
217 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
218 """Get streamdetails for a track/radio."""
219 # There's a chance that the SiriusXM auth session has expired
220 # by the time the user clicks to play a station. The sxm-client
221 # will attempt to reauthenticate automatically, but this causes
222 # a delay in streaming, and ffmpeg raises a TimeoutError.
223 # To prevent this, we're going to explicitly authenticate with
224 # SiriusXM proactively when a station has been chosen to avoid
225 # this.
226 await self._client.authenticate()
227
228 hls_path = f"http://{self._base_url}/{item_id}.m3u8"
229
230 # Keep a reference to the current `StreamDetails` object so that we can
231 # update the `stream_title` attribute as callbacks come in from the
232 # sxm-client with the channel's live data.
233 # See `_channel_updated` for where this is handled.
234 self._current_stream_details = StreamDetails(
235 item_id=item_id,
236 provider=self.instance_id,
237 audio_format=AudioFormat(
238 content_type=ContentType.AAC,
239 ),
240 stream_type=StreamType.HLS,
241 media_type=MediaType.RADIO,
242 path=hls_path,
243 can_seek=False,
244 allow_seek=False,
245 )
246
247 return self._current_stream_details
248
249 @use_cache(3600 * 3) # Cache for 3 hours
250 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
251 """Browse this provider's items.
252
253 :param path: The path to browse, (e.g. provider_id://artists).
254 """
255 return [self._parse_radio(channel) for channel in self._channels]
256
257 def _channel_updated(self, live_channel_raw: dict[str, Any]) -> None:
258 """Handle a channel update event."""
259 live_data = XMLiveChannel.from_dict(live_channel_raw)
260
261 self.logger.debug(f"Got update for SiriusXM channel {live_data.id}")
262
263 if self._current_stream_details is None:
264 return
265
266 current_channel = self._current_stream_details.item_id
267
268 if live_data.id != current_channel:
269 # This can happen when changing channels
270 self.logger.debug(
271 f"Received update for channel {live_data.id}, current channel is {current_channel}"
272 )
273 return
274
275 latest_cut_marker = live_data.get_latest_cut()
276
277 if latest_cut_marker:
278 latest_cut = latest_cut_marker.cut
279 title = latest_cut.title
280 artist = ", ".join([a.name for a in latest_cut.artists])
281 self._current_stream_details.stream_title = f"{artist} - {title}"
282
283 async def _refresh_channels(self) -> bool:
284 self._channels = await self._client.channels
285
286 self._channels_by_id = {}
287
288 for channel in self._channels:
289 self._channels_by_id[channel.id] = channel
290
291 return True
292
293 def _parse_radio(self, channel: XMChannel) -> Radio:
294 radio = Radio(
295 provider=self.instance_id,
296 item_id=channel.id,
297 name=channel.name,
298 provider_mappings={
299 ProviderMapping(
300 provider_domain=self.domain,
301 provider_instance=self.instance_id,
302 item_id=channel.id,
303 )
304 },
305 )
306
307 icon = next((i.url for i in channel.images if i.width == 300 and i.height == 300), None)
308 banner = next(
309 (i.url for i in channel.images if i.name in ("channel hero image", "background")), None
310 )
311
312 images: list[MediaItemImage] = []
313
314 if icon is not None:
315 images.append(
316 MediaItemImage(
317 provider=self.instance_id,
318 type=ImageType.THUMB,
319 path=icon,
320 remotely_accessible=True,
321 )
322 )
323 images.append(
324 MediaItemImage(
325 provider=self.instance_id,
326 type=ImageType.LOGO,
327 path=icon,
328 remotely_accessible=True,
329 )
330 )
331
332 if banner is not None:
333 images.append(
334 MediaItemImage(
335 provider=self.instance_id,
336 type=ImageType.BANNER,
337 path=banner,
338 remotely_accessible=True,
339 )
340 )
341 images.append(
342 MediaItemImage(
343 provider=self.instance_id,
344 type=ImageType.LANDSCAPE,
345 path=banner,
346 remotely_accessible=True,
347 )
348 )
349
350 radio.metadata.images = UniqueList(images) if images else None
351 radio.metadata.links = {MediaItemLink(type=LinkType.WEBSITE, url=channel.url)}
352 radio.metadata.description = channel.medium_description
353 radio.metadata.explicit = bool(channel.is_mature)
354 radio.metadata.genres = {cat.name for cat in channel.categories}
355
356 return radio
357