Add type annotations to controllers/__init__.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
+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/__init__.py

@@ -6,25 +6,40 @@
 import abc
 from functools import partial
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Protocol
 
 from ..error import UnsupportedNamespace, ControllerNotRegistered
-from ..response_handler import chain_on_success
+from ..generated.cast_channel_pb2 import CastMessage  # pylint: disable=no-name-in-module
+from ..response_handler import CallbackType, chain_on_success
 
 if TYPE_CHECKING:
     from ..socket_client import SocketClient
 
 
+class SendMessageFunc(Protocol):
+    """Protocol for SocketClient's send message functions."""
+    def __call__(
+        self,
+        namespace: str,
+        message: Any,
+        *,
+        inc_session_id: bool = False,
+        callback_function: CallbackType | None = None,
+        no_add_request_id: bool = False,
+    ) -> None:
+        ...
+
+
 class BaseController(abc.ABC):
     """ABC for namespace controllers."""
 
     def __init__(
         self,
-        namespace,
-        supporting_app_id=None,
-        target_platform=False,
-        app_must_match=False,
-    ):
+        namespace: str,
+        supporting_app_id: str | None = None,
+        target_platform: bool = False,
+        app_must_match: bool = False,
+    ) -> None:
         """
         Initialize the controller.
 
@@ -42,30 +57,43 @@ def __init__(
         self.target_platform = target_platform
 
         self._socket_client: SocketClient | None = None
-        self._message_func = None
+        self._message_func: SendMessageFunc | None = None
 
         self.logger = logging.getLogger(__name__)
 
     @property
-    def is_active(self):
+    def is_active(self) -> bool:
         """True if the controller is connected to a socket client and the
         Chromecast is running an app that supports this controller."""
         return (
             self._socket_client is not None
             and self.namespace in self._socket_client.app_namespaces
         )
 
-    def launch(self, *, callback_function=None, force_launch=False):
+    def launch(
+        self,
+        *,
+        callback_function: CallbackType | None = None,
+        force_launch: bool = False,
+    ) -> None:
         """If set, launches app related to the controller."""
-        self._check_registered()
+        if self.supporting_app_id is None:
+            self.logger.debug(
+                "%s: Can't launch app with no supporting app_id",
+                self.__class__.__name__,
+            )
+            return
 
-        self._socket_client.receiver_controller.launch_app(
+        if self._socket_client is None:
+            raise ControllerNotRegistered
+
+        self._socket_client.receiver_controller.launch_app(  # type: ignore[no-untyped-call]
             self.supporting_app_id,
             force_launch=force_launch,
             callback_function=callback_function,
         )
 
-    def registered(self, socket_client):
+    def registered(self, socket_client: SocketClient) -> None:
         """Called when a controller is registered."""
         self._socket_client = socket_client
 
@@ -74,31 +102,32 @@ def registered(self, socket_client):
         else:
             self._message_func = self._socket_client.send_app_message
 
-    def unregistered(self):
+    def unregistered(self) -> None:
         """Called when a controller is unregistered."""
         self._message_func = None
 
-    def channel_connected(self):
+    def channel_connected(self) -> None:
         """Called when a channel has been openend that supports the
         namespace of this controller."""
 
-    def channel_disconnected(self):
+    def channel_disconnected(self) -> None:
         """Called when a channel is disconnected."""
 
     def send_message(
         self,
-        data,
+        data: Any,
         *,
-        inc_session_id=False,
-        callback_function=None,
-        no_add_request_id=False,
-    ):
+        inc_session_id: bool = False,
+        callback_function: CallbackType | None = None,
+        no_add_request_id: bool = False,
+    ) -> None:
         """
         Send a message on this namespace to the Chromecast. Ensures app is loaded.
 
         Will raise a NotConnected exception if not connected.
         """
-        self._check_registered()
+        if self._socket_client is None:
+            raise ControllerNotRegistered
 
         receiver_ctrl = self._socket_client.receiver_controller
 
@@ -135,13 +164,16 @@ def send_message(
 
     def send_message_nocheck(
         self,
-        data,
+        data: Any,
         *,
-        inc_session_id=False,
-        callback_function=None,
-        no_add_request_id=False,
-    ):
+        inc_session_id: bool = False,
+        callback_function: CallbackType | None = None,
+        no_add_request_id: bool = False,
+    ) -> None:
         """Send a message."""
+        if TYPE_CHECKING:
+            assert self._message_func
+
         self._message_func(
             self.namespace,
             data,
@@ -150,25 +182,15 @@ def send_message_nocheck(
             no_add_request_id=no_add_request_id,
         )
 
-    def receive_message(self, _message, _data: dict):
+    def receive_message(self, _message: CastMessage, _data: dict) -> bool:
         """
         Called when a message is received that matches the namespace.
         Returns boolean indicating if message was handled.
         data is message.payload_utf8 interpreted as a JSON dict.
         """
         return False
 
-    def tear_down(self):
+    def tear_down(self) -> None:
         """Called when we are shutting down."""
         self._socket_client = None
         self._message_func = None
-
-    def _check_registered(self):
-        """Helper method to see if we are registered with a Cast object."""
-        if self._socket_client is None:
-            raise ControllerNotRegistered(
-                (
-                    "Trying to use the controller without it being registered "
-                    "with a Cast object."
-                )
-            )