chore: mypy for sonos s1

music_assistant/mass.py

@@ -315,14 +315,14 @@ def remove_listener() -> None:
 
         return remove_listener
 
-    def create_task(
+    def create_task[_R](
         self,
         target: Coroutine | Awaitable | Callable,
         *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.

music_assistant/providers/sonos_s1/__init__.py

@@ -13,7 +13,7 @@
 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 @@
     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,18 +133,21 @@ 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."""
         return {ProviderFeature.SYNC_PLAYERS}
 
     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 @@ async def unload(self, is_removed: bool = False) -> None:
         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 @@ async def play_media(
         """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 @@ async def enqueue_next_media(self, player_id: str, media: PlayerMedia) -> None:
             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 @@ def do_discover() -> None:
             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 @@ def _add_player(self, soco: SoCo) -> None:
 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]:

music_assistant/providers/sonos_s1/helpers.py

@@ -11,7 +11,7 @@
 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()

music_assistant/providers/sonos_s1/player.py

@@ -139,7 +139,7 @@ def zone_name(self) -> str:
         """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 @@ def update_player(self, signal_update: bool = True) -> None:
     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 @@ def poll_media(self) -> None:
         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 @@ def update_group_for_uid(self, uid: str) -> None:
         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 @@ def _regroup(group: list[str]) -> None:
             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 @@ def _speaker_activity(self, source: str) -> None:
     @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 @@ def _poll_track_info(self) -> dict[str, Any]:
         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(":")))))

pyproject.toml

@@ -143,7 +143,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/.*$',