music-assistant-server

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