Add type annotations to controllers/multizone.py

mypy.ini

@@ -19,4 +19,4 @@ disallow_untyped_decorators = true
 disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
-files = pychromecast/config.py, pychromecast/const.py, pychromecast/dial.py, pychromecast/discovery.py, pychromecast/error.py, pychromecast/models.py, pychromecast/response_handler.py, pychromecast/controllers/__init__.py, pychromecast/controllers/media.py, pychromecast/controllers/receiver.py
+files = pychromecast/config.py, pychromecast/const.py, pychromecast/dial.py, pychromecast/discovery.py, pychromecast/error.py, pychromecast/models.py, pychromecast/response_handler.py, pychromecast/controllers/__init__.py, pychromecast/controllers/media.py, pychromecast/controllers/multizone.py, pychromecast/controllers/receiver.py

pychromecast/controllers/multizone.py

@@ -1,19 +1,31 @@
 """
 Controller to monitor audio group members.
 """
+from __future__ import annotations
+
 import abc
 import logging
+from typing import TYPE_CHECKING, TypedDict
+from uuid import UUID
 
 from . import BaseController
-from .media import MediaStatus
-from .receiver import CastStatus
+from .media import MediaController, MediaStatusListener, MediaStatus
+from .receiver import CastStatus, CastStatusListener
 from ..const import MESSAGE_TYPE
+
+# pylint: disable-next=no-name-in-module
+from ..generated.cast_channel_pb2 import CastMessage
 from ..socket_client import (
     CONNECTION_STATUS_CONNECTED,
     CONNECTION_STATUS_DISCONNECTED,
     CONNECTION_STATUS_LOST,
+    ConnectionStatus,
+    ConnectionStatusListener,
 )
 
+if TYPE_CHECKING:
+    from .. import Chromecast
+
 _LOGGER = logging.getLogger(__name__)
 
 MULTIZONE_NAMESPACE = "urn:x-cast:com.google.cast.multizone"
@@ -27,10 +39,71 @@
 TYPE_SESSION_UPDATED = "PLAYBACK_SESSION_UPDATED"
 
 
-class Listener:
+class GroupInfo(TypedDict):
+    """Chromecast connection and listener for a group."""
+
+    chromecast: Chromecast
+    listener: Listener
+
+
+class GroupMemberInfo(TypedDict):
+    """Group memberships and listener for a group."""
+
+    group_memberships: set[str]
+    listeners: list[MultiZoneManagerListener]
+
+
+class MultiZoneControllerListener(abc.ABC):
+    """Listener for receiving audio group events."""
+
+    @abc.abstractmethod
+    def multizone_member_added(self, group_uuid: str) -> None:
+        """The cast has been added to group identified by group_uuid."""
+
+    @abc.abstractmethod
+    def multizone_member_removed(self, group_uuid: str) -> None:
+        """The cast has been removed from group identified by group_uuid."""
+
+    @abc.abstractmethod
+    def multizone_status_received(self) -> None:
+        """Multizone status has been updated."""
+
+
+class MultiZoneManagerListener(abc.ABC):
+    """Listener for receiving audio group events for a cast device."""
+
+    @abc.abstractmethod
+    def added_to_multizone(self, group_uuid: str) -> None:
+        """The cast has been added to group identified by group_uuid."""
+
+    @abc.abstractmethod
+    def removed_from_multizone(self, group_uuid: str) -> None:
+        """The cast has been removed from group identified by group_uuid."""
+
+    @abc.abstractmethod
+    def multizone_new_media_status(
+        self, group_uuid: str, media_status: MediaStatus
+    ) -> None:
+        """The group identified by group_uuid, of which the cast is a member, has new media status."""
+
+    @abc.abstractmethod
+    def multizone_new_cast_status(
+        self, group_uuid: str, cast_status: CastStatus
+    ) -> None:
+        """The group identified by group_uuid, of which the cast is a member, has new status."""
+
+
+class Listener(
+    CastStatusListener,
+    ConnectionStatusListener,
+    MediaStatusListener,
+    MultiZoneControllerListener,
+):
     """Callback handler."""
 
-    def __init__(self, group_cast, casts):
+    def __init__(
+        self, group_cast: Chromecast, casts: dict[str, GroupMemberInfo]
+    ) -> None:
         """Initialize the listener."""
         self._casts = casts
         group_cast.register_status_listener(self)
@@ -41,106 +114,90 @@ def __init__(self, group_cast, casts):
         self._group_uuid = str(group_cast.uuid)
         group_cast.register_handler(self._mz)
 
-    def new_cast_status(self, cast_status):
+    def new_cast_status(self, status: CastStatus) -> None:
         """Handle reception of a new CastStatus."""
         casts = self._casts
         group_members = self._mz.members
         for member_uuid in group_members:
             if member_uuid not in casts:
                 continue
             for listener in list(casts[member_uuid]["listeners"]):
-                listener.multizone_new_cast_status(self._group_uuid, cast_status)
+                listener.multizone_new_cast_status(self._group_uuid, status)
 
-    def new_media_status(self, media_status):
+    def new_media_status(self, status: MediaStatus) -> None:
         """Handle reception of a new MediaStatus."""
         casts = self._casts
         group_members = self._mz.members
         for member_uuid in group_members:
             if member_uuid not in casts:
                 continue
             for listener in list(casts[member_uuid]["listeners"]):
-                listener.multizone_new_media_status(self._group_uuid, media_status)
+                listener.multizone_new_media_status(self._group_uuid, status)
 
-    def new_connection_status(self, conn_status):
+    def load_media_failed(self, item: int, error_code: int) -> None:
+        """Called when load media failed."""
+
+    def new_connection_status(self, status: ConnectionStatus) -> None:
         """Handle reception of a new ConnectionStatus."""
-        if conn_status.status == CONNECTION_STATUS_CONNECTED:
+        if status.status == CONNECTION_STATUS_CONNECTED:
             self._mz.update_members()
-        if conn_status.status in (
+        if status.status in (
             CONNECTION_STATUS_DISCONNECTED,
             CONNECTION_STATUS_LOST,
         ):
             self._mz.reset_members()
 
-    def multizone_member_added(self, member_uuid):
+    def multizone_member_added(self, group_uuid: str) -> None:
         """Handle added audio group member."""
         casts = self._casts
-        if member_uuid not in casts:
-            casts[member_uuid] = {"listeners": [], "groups": set()}
-        casts[member_uuid]["groups"].add(self._group_uuid)
-        for listener in list(casts[member_uuid]["listeners"]):
+        if group_uuid not in casts:
+            casts[group_uuid] = {"listeners": [], "group_memberships": set()}
+        casts[group_uuid]["group_memberships"].add(self._group_uuid)
+        for listener in list(casts[group_uuid]["listeners"]):
             listener.added_to_multizone(self._group_uuid)
 
-    def multizone_member_removed(self, member_uuid):
+    def multizone_member_removed(self, group_uuid: str) -> None:
         """Handle removed audio group member."""
         casts = self._casts
-        if member_uuid not in casts:
-            casts[member_uuid] = {"listeners": [], "groups": set()}
-        casts[member_uuid]["groups"].discard(self._group_uuid)
-        for listener in list(casts[member_uuid]["listeners"]):
+        if group_uuid not in casts:
+            casts[group_uuid] = {"listeners": [], "group_memberships": set()}
+        casts[group_uuid]["group_memberships"].discard(self._group_uuid)
+        for listener in list(casts[group_uuid]["listeners"]):
             listener.removed_from_multizone(self._group_uuid)
 
-    def multizone_status_received(self):
+    def multizone_status_received(self) -> None:
         """Handle reception of audio group status."""
 
 
-class MultiZoneManagerListener(abc.ABC):
-    """Listener for receiving audio group events for a cast device."""
-
-    @abc.abstractmethod
-    def added_to_multizone(self, group_uuid: str):
-        """The cast has been added to group identified by group_uuid."""
-
-    @abc.abstractmethod
-    def removed_from_multizone(self, group_uuid: str):
-        """The cast has been removed from group identified by group_uuid."""
-
-    @abc.abstractmethod
-    def multizone_new_media_status(self, group_uuid: str, media_status: MediaStatus):
-        """The group identified by group_uuid, of which the cast is a member, has new media status."""
-
-    @abc.abstractmethod
-    def multizone_new_cast_status(self, group_uuid: str, cast_status: CastStatus):
-        """The group identified by group_uuid, of which the cast is a member, has new status."""
-
-
 class MultizoneManager:
     """Manage audio groups."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         # Protect self._casts because it will be accessed from callbacks from
         # the casts' socket_client thread
-        self._casts = {}
-        self._groups = {}
+        self._casts: dict[str, GroupMemberInfo] = {}
+        self._groups: dict[str, GroupInfo] = {}
 
-    def add_multizone(self, group_cast):
+    def add_multizone(self, group_cast: Chromecast) -> None:
         """Start managing a group"""
         self._groups[str(group_cast.uuid)] = {
             "chromecast": group_cast,
             "listener": Listener(group_cast, self._casts),
-            "members": set(),
         }
 
-    def remove_multizone(self, group_uuid):
+    def remove_multizone(self, group_uuid: UUID) -> None:
         """Stop managing a group"""
-        group_uuid = str(group_uuid)
-        group = self._groups.pop(group_uuid, None)
+        group_uuid_str = str(group_uuid)
+        group = self._groups.pop(group_uuid_str, None)
         # Inform all group members that they are no longer members
         if group is not None:
             group["listener"]._mz.reset_members()  # pylint: disable=protected-access
         for member in self._casts.values():
-            member["groups"].discard(group_uuid)
+            member["group_memberships"].discard(group_uuid_str)
 
-    def register_listener(self, member_uuid, listener: MultiZoneManagerListener):
+    def register_listener(
+        self, member_uuid: UUID, listener: MultiZoneManagerListener
+    ) -> None:
         """Register a listener for audio group changes of cast uuid.
         On update will call:
         listener.added_to_multizone(group_uuid)
@@ -152,50 +209,36 @@ def register_listener(self, member_uuid, listener: MultiZoneManagerListener):
         listener.multizone_new_cast_status(group_uuid, cast_status)
             The group uuid, of which the cast is a member, has new status
         """
-        member_uuid = str(member_uuid)
-        if member_uuid not in self._casts:
-            self._casts[member_uuid] = {"listeners": [], "groups": set()}
-        self._casts[member_uuid]["listeners"].append(listener)
-
-    def deregister_listener(self, member_uuid, listener):
+        member_uuid_str = str(member_uuid)
+        if member_uuid_str not in self._casts:
+            self._casts[member_uuid_str] = {"listeners": [], "group_memberships": set()}
+        self._casts[member_uuid_str]["listeners"].append(listener)
+
+    def deregister_listener(
+        self, member_uuid: UUID, listener: MultiZoneManagerListener
+    ) -> None:
         """Deregister listener for audio group changes of cast uuid."""
         self._casts[str(member_uuid)]["listeners"].remove(listener)
 
-    def get_multizone_memberships(self, member_uuid):
+    def get_multizone_memberships(self, member_uuid: UUID) -> list[str]:
         """Return a list of audio groups in which cast member_uuid is a member"""
-        return list(self._casts[str(member_uuid)]["groups"])
+        return list(self._casts[str(member_uuid)]["group_memberships"])
 
-    def get_multizone_mediacontroller(self, group_uuid):
+    def get_multizone_mediacontroller(self, group_uuid: UUID) -> MediaController:
         """Get mediacontroller of a group"""
-        return self._groups[str(group_uuid)]["chromecast"].media_controller
-
-
-class MultiZoneControllerListener(abc.ABC):
-    """Listener for receiving audio group events."""
-
-    @abc.abstractmethod
-    def multizone_member_added(self, group_uuid: str):
-        """The cast has been added to group identified by group_uuid."""
-
-    @abc.abstractmethod
-    def multizone_member_removed(self, group_uuid: str):
-        """The cast has been removed from group identified by group_uuid."""
-
-    @abc.abstractmethod
-    def multizone_status_received(self):
-        """Multizone status has been updated."""
+        return self._groups[str(group_uuid)]["chromecast"].media_controller  # type: ignore[no-any-return]
 
 
 class MultizoneController(BaseController):
     """Controller to monitor audio group members."""
 
-    def __init__(self, uuid):
-        self._members = {}
-        self._status_listeners = []
+    def __init__(self, uuid: UUID) -> None:
+        self._members: dict[str, str] = {}
+        self._status_listeners: list[MultiZoneControllerListener] = []
         self._uuid = str(uuid)
         super().__init__(MULTIZONE_NAMESPACE, target_platform=True)
 
-    def _add_member(self, uuid, name):
+    def _add_member(self, uuid: str, name: str) -> None:
         if uuid not in self._members:
             self._members[uuid] = name
             _LOGGER.debug(
@@ -208,7 +251,7 @@ def _add_member(self, uuid, name):
             for listener in list(self._status_listeners):
                 listener.multizone_member_added(uuid)
 
-    def _remove_member(self, uuid):
+    def _remove_member(self, uuid: str) -> None:
         name = self._members.pop(uuid, "<Unknown>")
         _LOGGER.debug(
             "(%s) Removed member %s(%s), members: %s",
@@ -220,7 +263,7 @@ def _remove_member(self, uuid):
         for listener in list(self._status_listeners):
             listener.multizone_member_removed(uuid)
 
-    def register_listener(self, listener: MultiZoneControllerListener):
+    def register_listener(self, listener: MultiZoneControllerListener) -> None:
         """Register a listener for audio group changes. On update will call:
         listener.multizone_member_added(uuid)
         listener.multizone_member_removed(uuid)
@@ -229,26 +272,26 @@ def register_listener(self, listener: MultiZoneControllerListener):
         self._status_listeners.append(listener)
 
     @property
-    def members(self):
+    def members(self) -> list[str]:
         """Return a list of audio group members."""
         return list(self._members.keys())
 
-    def reset_members(self):
+    def reset_members(self) -> None:
         """Reset audio group members."""
         for uuid in list(self._members):
             self._remove_member(uuid)
 
-    def update_members(self):
+    def update_members(self) -> None:
         """Update audio group members."""
         self.send_message({MESSAGE_TYPE: TYPE_GET_STATUS})
 
-    def get_casting_groups(self):
+    def get_casting_groups(self) -> None:
         """Send GET_CASTING_GROUPS message."""
         self.send_message({MESSAGE_TYPE: TYPE_GET_CASTING_GROUPS})
 
-    def receive_message(
-        self, _message, data: dict
-    ):  # pylint: disable=too-many-return-statements
+    def receive_message(  # pylint: disable=too-many-return-statements
+        self, _message: CastMessage, data: dict
+    ) -> bool:
         """Called when a multizone message is received."""
         if data[MESSAGE_TYPE] == TYPE_DEVICE_ADDED:
             uuid = data["device"]["deviceId"]
@@ -299,7 +342,7 @@ def receive_message(
 
         return False
 
-    def tear_down(self):
+    def tear_down(self) -> None:
         """Called when controller is destroyed."""
         super().tear_down()