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.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