Add typing for radiobrowser

music_assistant/server/controllers/cache.py

@@ -8,8 +8,8 @@
 import os
 import time
 from collections import OrderedDict
-from collections.abc import Iterator, MutableMapping
-from typing import TYPE_CHECKING, Any
+from collections.abc import Callable, Iterator, MutableMapping
+from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
 from music_assistant.common.helpers.json import json_dumps, json_loads
 from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
@@ -235,12 +235,18 @@ def __schedule_cleanup_task(self) -> None:
         self.mass.loop.call_later(3600, self.__schedule_cleanup_task)
 
 
-def use_cache(expiration=86400 * 30):
+Param = ParamSpec("Param")
+RetType = TypeVar("RetType")
+
+
+def use_cache(
+    expiration: int = 86400 * 30,
+) -> Callable[[Callable[Param, RetType]], Callable[Param, RetType]]:
     """Return decorator that can be used to cache a method's result."""
 
-    def wrapper(func):
+    def wrapper(func: Callable[Param, RetType]) -> Callable[Param, RetType]:
         @functools.wraps(func)
-        async def wrapped(*args, **kwargs):
+        async def wrapped(*args: Param.args, **kwargs: Param.kwargs):
             method_class = args[0]
             method_class_name = method_class.__class__.__name__
             cache_key_parts = [method_class_name, func.__name__]

music_assistant/server/models/music_provider.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
 from music_assistant.common.models.enums import MediaType, ProviderFeature
@@ -282,7 +283,7 @@ async def get_item(self, media_type: MediaType, prov_item_id: str) -> MediaItemT
             return await self.get_radio(prov_item_id)
         return await self.get_track(prov_item_id)
 
-    async def browse(self, path: str, offset: int, limit: int) -> list[MediaItemType]:
+    async def browse(self, path: str, offset: int, limit: int) -> Sequence[MediaItemType]:
         """Browse this provider's items.
 
         :param path: The path to browse, (e.g. provider_id://artists).

music_assistant/server/providers/radiobrowser/__init__.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from radios import FilterBy, Order, RadioBrowser, RadioBrowserError
+from radios import FilterBy, Order, RadioBrowser, RadioBrowserError, Station
 
 from music_assistant.common.models.enums import LinkType, ProviderFeature, StreamType
+from music_assistant.common.models.errors import MediaNotFoundError
 from music_assistant.common.models.media_items import (
     AudioFormat,
     BrowseFolder,
@@ -19,6 +21,7 @@
     ProviderMapping,
     Radio,
     SearchResults,
+    UniqueList,
 )
 from music_assistant.common.models.streamdetails import StreamDetails
 from music_assistant.server.controllers.cache import use_cache
@@ -83,7 +86,7 @@ async def handle_async_init(self) -> None:
             self.logger.exception("%s", err)
 
     async def search(
-        self, search_query: str, media_types=list[MediaType], limit: int = 10
+        self, search_query: str, media_types: list[MediaType], limit: int = 10
     ) -> SearchResults:
         """Perform search on musicprovider.
 
@@ -102,7 +105,7 @@ async def search(
 
         return result
 
-    async def browse(self, path: str, offset: int, limit: int) -> list[MediaItemType]:
+    async def browse(self, path: str, offset: int, limit: int) -> Sequence[MediaItemType]:
         """Browse this provider's items.
 
         :param path: The path to browse, (e.g. provid://artists).
@@ -168,14 +171,16 @@ async def browse(self, path: str, offset: int, limit: int) -> list[MediaItemType
                     path=path + "/" + country.code.lower(),
                     name=country.name,
                 )
-                folder.metadata.images = [
-                    MediaItemImage(
-                        type=ImageType.THUMB,
-                        path=country.favicon,
-                        provider=self.instance_id,
-                        remotely_accessible=True,
-                    )
-                ]
+                folder.metadata.images = UniqueList(
+                    [
+                        MediaItemImage(
+                            type=ImageType.THUMB,
+                            path=country.favicon,
+                            provider=self.instance_id,
+                            remotely_accessible=True,
+                        )
+                    ]
+                )
                 items.append(folder)
             return items
 
@@ -187,7 +192,7 @@ async def browse(self, path: str, offset: int, limit: int) -> list[MediaItemType
         return []
 
     @use_cache(3600 * 24)
-    async def get_tag_names(self):
+    async def get_tag_names(self) -> Sequence[str]:
         """Get a list of tag names."""
         tags = await self.radios.tags(
             hide_broken=True,
@@ -202,7 +207,7 @@ async def get_tag_names(self):
         return tag_names
 
     @use_cache(3600 * 24)
-    async def get_country_codes(self):
+    async def get_country_codes(self) -> Sequence[str]:
         """Get a list of country names."""
         countries = await self.radios.countries(order=Order.NAME, hide_broken=True)
         country_codes = []
@@ -211,7 +216,7 @@ async def get_country_codes(self):
         return country_codes
 
     @use_cache(3600)
-    async def get_by_popularity(self):
+    async def get_by_popularity(self) -> Sequence[Radio]:
         """Get radio stations by popularity."""
         stations = await self.radios.stations(
             hide_broken=True,
@@ -225,7 +230,7 @@ async def get_by_popularity(self):
         return items
 
     @use_cache(3600)
-    async def get_by_tag(self, tag: str):
+    async def get_by_tag(self, tag: str) -> Sequence[Radio]:
         """Get radio stations by tag."""
         items = []
         stations = await self.radios.stations(
@@ -240,7 +245,7 @@ async def get_by_tag(self, tag: str):
         return items
 
     @use_cache(3600)
-    async def get_by_country(self, country_code: str):
+    async def get_by_country(self, country_code: str) -> list[Radio]:
         """Get radio stations by country."""
         items = []
         stations = await self.radios.stations(
@@ -257,9 +262,11 @@ async def get_by_country(self, country_code: str):
     async def get_radio(self, prov_radio_id: str) -> Radio:
         """Get radio station details."""
         radio = await self.radios.station(uuid=prov_radio_id)
+        if not radio:
+            raise MediaNotFoundError(f"Radio station {prov_radio_id} not found")
         return await self._parse_radio(radio)
 
-    async def _parse_radio(self, radio_obj: dict) -> Radio:
+    async def _parse_radio(self, radio_obj: Station) -> Radio:
         """Parse Radio object from json obj returned from api."""
         radio = Radio(
             item_id=radio_obj.uuid,
@@ -273,23 +280,26 @@ async def _parse_radio(self, radio_obj: dict) -> Radio:
                 )
             },
         )
-        radio.metadata.label = radio_obj.tags
         radio.metadata.popularity = radio_obj.votes
-        radio.metadata.links = [MediaItemLink(type=LinkType.WEBSITE, url=radio_obj.homepage)]
-        radio.metadata.images = [
-            MediaItemImage(
-                type=ImageType.THUMB,
-                path=radio_obj.favicon,
-                provider=self.instance_id,
-                remotely_accessible=True,
-            )
-        ]
+        radio.metadata.links = {MediaItemLink(type=LinkType.WEBSITE, url=radio_obj.homepage)}
+        radio.metadata.images = UniqueList(
+            [
+                MediaItemImage(
+                    type=ImageType.THUMB,
+                    path=radio_obj.favicon,
+                    provider=self.instance_id,
+                    remotely_accessible=True,
+                )
+            ]
+        )
 
         return radio
 
     async def get_stream_details(self, item_id: str) -> StreamDetails:
         """Get streamdetails for a radio station."""
         stream = await self.radios.station(uuid=item_id)
+        if not stream:
+            raise MediaNotFoundError(f"Radio station {item_id} not found")
         await self.radios.station_click(uuid=item_id)
         return StreamDetails(
             provider=self.domain,

mypy.ini

@@ -21,4 +21,4 @@ disallow_untyped_decorators = true
 disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
-packages=tests,music_assistant.client,music_assistant.common,music_assistant.server.providers.jellyfin
+packages=tests,music_assistant.client,music_assistant.common,music_assistant.server.providers.jellyfin,music_assistant.server.providers.radiobrowser