From: Jc2k Date: Fri, 10 Jan 2025 23:04:37 +0000 (+0000) Subject: chore: mypy for sonos s1 (#1854) X-Git-Url: https://git.kitaultman.com/?a=commitdiff_plain;h=834864623c5269f87a5e20813012f45fdd65ee30;p=music-assistant-server.git chore: mypy for sonos s1 (#1854) --- diff --git a/music_assistant/mass.py b/music_assistant/mass.py index 4b9c8298..7f921ae4 100644 --- a/music_assistant/mass.py +++ b/music_assistant/mass.py @@ -6,7 +6,7 @@ import asyncio import logging import os from collections.abc import Awaitable, Callable, Coroutine -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, TypeVar from uuid import uuid4 import aiofiles @@ -74,6 +74,8 @@ LOGGER = logging.getLogger(MASS_LOGGER_NAME) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) PROVIDERS_PATH = os.path.join(BASE_DIR, "providers") +_R = TypeVar("_R") + class MusicAssistant: """Main MusicAssistant (Server) object.""" @@ -317,12 +319,12 @@ class MusicAssistant: def create_task( self, - target: Coroutine | Awaitable | Callable, + target: Coroutine[Any, Any, _R] | Awaitable[_R] | Callable[..., _R], *args: Any, task_id: str | None = None, abort_existing: bool = False, **kwargs: Any, - ) -> asyncio.Task | asyncio.Future: + ) -> asyncio.Task[_R] | asyncio.Future[_R]: """Create Task on (main) event loop from Coroutine(function). Tasks created by this helper will be properly cancelled on stop. diff --git a/music_assistant/providers/sonos_s1/__init__.py b/music_assistant/providers/sonos_s1/__init__.py index ed62855a..3fcfe3f2 100644 --- a/music_assistant/providers/sonos_s1/__init__.py +++ b/music_assistant/providers/sonos_s1/__init__.py @@ -13,7 +13,7 @@ import asyncio import logging from collections import OrderedDict from dataclasses import dataclass, field -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from music_assistant_models.config_entries import ConfigEntry, ConfigValueType from music_assistant_models.enums import ( @@ -50,7 +50,7 @@ if TYPE_CHECKING: from music_assistant_models.provider import ProviderManifest from soco.core import SoCo - from music_assistant import MusicAssistant + from music_assistant.mass import MusicAssistant from music_assistant.models import ProviderInstanceType @@ -133,10 +133,14 @@ class UnjoinData: class SonosPlayerProvider(PlayerProvider): """Sonos Player provider.""" - sonosplayers: dict[str, SonosPlayer] | None = None _discovery_running: bool = False _discovery_reschedule_timer: asyncio.TimerHandle | None = None + def __init__(self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig): + """Handle initialization of the provider.""" + super().__init__(mass, manifest, config) + self.sonosplayers: OrderedDict[str, SonosPlayer] = OrderedDict() + @property def supported_features(self) -> set[ProviderFeature]: """Return the features supported by this Provider.""" @@ -144,7 +148,6 @@ class SonosPlayerProvider(PlayerProvider): async def handle_async_init(self) -> None: """Handle async initialization of the provider.""" - self.sonosplayers: OrderedDict[str, SonosPlayer] = OrderedDict() self.topology_condition = asyncio.Condition() self.boot_counts: dict[str, int] = {} self.mdns_names: dict[str, str] = {} @@ -166,7 +169,7 @@ class SonosPlayerProvider(PlayerProvider): await asyncio.gather(*(player.offline() for player in self.sonosplayers.values())) if events_asyncio.event_listener: await events_asyncio.event_listener.async_stop() - self.sonosplayers = None + self.sonosplayers = OrderedDict() async def get_player_config_entries( self, @@ -287,6 +290,7 @@ class SonosPlayerProvider(PlayerProvider): """Handle PLAY MEDIA on given player.""" sonos_player = self.sonosplayers[player_id] mass_player = self.mass.players.get(player_id) + assert mass_player if sonos_player.sync_coordinator: # this should be already handled by the player manager, but just in case... msg = ( @@ -308,7 +312,7 @@ class SonosPlayerProvider(PlayerProvider): media.uri = media.uri.replace(".flac", ".mp3") didl_metadata = create_didl_metadata(media) # set crossfade according to player setting - crossfade = await self.mass.config.get_player_config_value(player_id, CONF_CROSSFADE) + crossfade = bool(await self.mass.config.get_player_config_value(player_id, CONF_CROSSFADE)) if sonos_player.crossfade != crossfade: def set_crossfade() -> None: @@ -375,11 +379,13 @@ class SonosPlayerProvider(PlayerProvider): self._discovery_running = True try: self.logger.debug("Sonos discovery started...") - discovered_devices: set[SoCo] = discover( - timeout=30, household_id=household_id, allow_network_scan=allow_network_scan + discovered_devices: set[SoCo] = ( + discover( + timeout=30, household_id=household_id, allow_network_scan=allow_network_scan + ) + or set() ) - if discovered_devices is None: - discovered_devices = set() + # process new players for soco in discovered_devices: try: @@ -463,7 +469,7 @@ class SonosPlayerProvider(PlayerProvider): async def discover_household_ids(mass: MusicAssistant, prefer_s1: bool = True) -> list[str]: """Discover the HouseHold ID of S1 speaker(s) the network.""" if cache := await mass.cache.get("sonos_household_ids"): - return cache + return cast(list[str], cache) household_ids: list[str] = [] def get_all_sonos_ips() -> set[SoCo]: diff --git a/music_assistant/providers/sonos_s1/helpers.py b/music_assistant/providers/sonos_s1/helpers.py index b9476562..eda5acf6 100644 --- a/music_assistant/providers/sonos_s1/helpers.py +++ b/music_assistant/providers/sonos_s1/helpers.py @@ -11,7 +11,7 @@ from soco import SoCo from soco.exceptions import SoCoException, SoCoUPnPException if TYPE_CHECKING: - from . import SonosPlayer + from .player import SonosPlayer UID_PREFIX = "RINCON_" @@ -81,11 +81,11 @@ def _find_target_identifier(instance: Any, fallback_soco: SoCo | None) -> str | """Extract the best available target identifier from the provided instance object.""" if zone_name := getattr(instance, "zone_name", None): # SonosPlayer instance - return zone_name + return str(zone_name) if soco := getattr(instance, "soco", fallback_soco): # Holds a SoCo instance attribute # Only use attributes with no I/O - return soco._player_name or soco.ip_address + return str(soco._player_name or soco.ip_address) return None @@ -105,4 +105,4 @@ def sync_get_visible_zones(soco: SoCo) -> set[SoCo]: """Ensure I/O attributes are cached and return visible zones.""" _ = soco.household_id _ = soco.uid - return soco.visible_zones + return soco.visible_zones or set() diff --git a/music_assistant/providers/sonos_s1/player.py b/music_assistant/providers/sonos_s1/player.py index a66406ce..339abdfd 100644 --- a/music_assistant/providers/sonos_s1/player.py +++ b/music_assistant/providers/sonos_s1/player.py @@ -139,7 +139,7 @@ class SonosPlayer: """Return zone name.""" if self.mass_player: return self.mass_player.display_name - return self.soco.speaker_info["zone_name"] + return str(self.soco.speaker_info["zone_name"]) @property def subscription_address(self) -> str: @@ -322,7 +322,7 @@ class SonosPlayer: async def poll_speaker(self) -> None: """Poll the speaker for updates.""" - def _poll(): + def _poll() -> None: """Poll the speaker for updates (NOT async friendly).""" self.update_groups() self.poll_media() @@ -346,7 +346,9 @@ class SonosPlayer: self._set_basic_track_info(update_position=update_position) self.update_player() - async def _subscribe_target(self, target: SubscriptionBase, sub_callback: Callable) -> None: + async def _subscribe_target( + self, target: SubscriptionBase, sub_callback: Callable[[SonosEvent], None] + ) -> None: """Create a Sonos subscription for given target.""" subscription = await target.subscribe( auto_renew=True, requested_timeout=SUBSCRIPTION_TIMEOUT @@ -502,7 +504,9 @@ class SonosPlayer: self.logger.debug("%s was missing, adding to %s group", missing_zone, self.zone_name) self.update_groups() - def create_update_groups_coro(self, event: SonosEvent | None = None) -> Coroutine: + def create_update_groups_coro( + self, event: SonosEvent | None = None + ) -> Coroutine[Any, Any, None]: """Handle callback for topology change event.""" def _get_soco_group() -> list[str]: @@ -563,7 +567,7 @@ class SonosPlayer: self.mass.loop.call_soon_threadsafe(self.mass.players.update, self.player_id) for joined_uid in group[1:]: - joined_speaker: SonosPlayer = self.sonos_prov.sonosplayers.get(joined_uid) + joined_speaker = self.sonos_prov.sonosplayers.get(joined_uid) if joined_speaker: joined_speaker.sync_coordinator = self joined_speaker.group_members = group_members @@ -763,7 +767,7 @@ class SonosPlayer: @soco_error() def _join(self, members: list[SonosPlayer]) -> list[SonosPlayer]: if self.sync_coordinator: - self.unjoin() + self._unjoin() group = [self] else: group = self.group_members.copy() @@ -795,7 +799,7 @@ class SonosPlayer: return track_info -def _convert_state(sonos_state: str) -> PlayerState: +def _convert_state(sonos_state: str | None) -> PlayerState: """Convert Sonos state to PlayerState.""" if sonos_state == "PLAYING": return PlayerState.PLAYING @@ -806,8 +810,10 @@ def _convert_state(sonos_state: str) -> PlayerState: return PlayerState.IDLE -def _timespan_secs(timespan): +def _timespan_secs(timespan: str | None) -> int | None: """Parse a time-span into number of seconds.""" - if timespan in ("", "NOT_IMPLEMENTED", None): + if timespan in ("", "NOT_IMPLEMENTED"): + return None + if timespan is None: return None - return sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":")))) + return int(sum(60 ** x[0] * int(x[1]) for x in enumerate(reversed(timespan.split(":"))))) diff --git a/pyproject.toml b/pyproject.toml index a4d60c8c..8df8f154 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -142,7 +142,6 @@ exclude = [ '^music_assistant/providers/siriusxm/.*$', '^music_assistant/providers/slimproto/.*$', '^music_assistant/providers/sonos/.*$', - '^music_assistant/providers/sonos_s1/.*$', '^music_assistant/providers/soundcloud/.*$', '^music_assistant/providers/snapcast/.*$', '^music_assistant/providers/spotify/.*$',