Improve typing and ship py.typed

Author: lawtancool

pyControl4/account.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
-from typing import AsyncGenerator
+from typing import Any, AsyncGenerator
 
 import aiohttp
 import asyncio
@@ -153,8 +153,9 @@ async def get_account_bearer_token(self) -> str:
         data = await self._send_account_auth_request()
         json_dict = json.loads(data)
         try:
-            self.account_bearer_token = json_dict["authToken"]["token"]
-            return self.account_bearer_token
+            token: str = json_dict["authToken"]["token"]
+            self.account_bearer_token = token
+            return token
         except KeyError:
             msg = (
                 "Did not receive an account bearer token. "
@@ -163,7 +164,7 @@ async def get_account_bearer_token(self) -> str:
             _LOGGER.error(msg + data)
             raise
 
-    async def get_account_controllers(self) -> dict:
+    async def get_account_controllers(self) -> dict[str, Any]:
         """Returns a dictionary of the information for all controllers registered
         to an account.
 
@@ -179,13 +180,14 @@ async def get_account_controllers(self) -> dict:
         data = await self._send_account_get_request(GET_CONTROLLERS_ENDPOINT)
         json_dict = json.loads(data)
         try:
-            return json_dict["account"]
+            result: dict[str, Any] = json_dict["account"]
+            return result
         except KeyError:
             msg = "Did not receive account information from the Control4 API."
             _LOGGER.error(msg + " Response: " + data)
             raise
 
-    async def get_controller_info(self, controller_href: str) -> dict:
+    async def get_controller_info(self, controller_href: str) -> dict[str, Any]:
         """Returns a dictionary of the information of a specific controller.
 
         Parameters:
@@ -228,7 +230,7 @@ async def get_controller_info(self, controller_href: str) -> dict:
             ```
         """
         data = await self._send_account_get_request(controller_href)
-        json_dict = json.loads(data)
+        json_dict: dict[str, Any] = json.loads(data)
         return json_dict
 
     async def get_controller_os_version(self, controller_href: str) -> str:
@@ -241,13 +243,14 @@ async def get_controller_os_version(self, controller_href: str) -> str:
         data = await self._send_account_get_request(controller_href + "/controller")
         json_dict = json.loads(data)
         try:
-            return json_dict["osVersion"]
+            version: str = json_dict["osVersion"]
+            return version
         except KeyError:
             msg = "Did not receive OS version from the Control4 API."
             _LOGGER.error(msg + " Response: " + data)
             raise
 
-    async def get_director_bearer_token(self, controller_common_name: str) -> dict:
+    async def get_director_bearer_token(self, controller_common_name: str) -> dict[str, Any]:
         """Returns a dictionary with a director bearer token for making Control4
         Director API requests, and its time valid in seconds (usually 86400 seconds)
 

pyControl4/director.py

@@ -76,7 +76,7 @@ async def send_get_request(self, uri: str) -> str:
                     return text
 
     async def send_post_request(
-        self, uri: str, command: str, params: dict, is_async: bool = True
+        self, uri: str, command: str, params: dict[str, Any], is_async: bool = True
     ) -> str:
         """Sends a POST request to the specified API URI. Used to send commands
            to the Director.
@@ -104,7 +104,7 @@ async def send_post_request(
                     check_response_for_error(text)
                     return text
 
-    async def get_all_items_by_category(self, category: str) -> list[dict]:
+    async def get_all_items_by_category(self, category: str) -> list[dict[str, Any]]:
         """Returns a list of items related to a particular category.
 
         Parameters:
@@ -115,23 +115,26 @@ async def get_all_items_by_category(self, category: str) -> list[dict]:
                          outlet_wireless_dimmer, voice-scene
         """
         data = await self.send_get_request(f"/api/v1/categories/{category}")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_all_item_info(self) -> list[dict]:
+    async def get_all_item_info(self) -> list[dict[str, Any]]:
         """Returns a list of all the items on the Director."""
         data = await self.send_get_request("/api/v1/items")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_item_info(self, item_id: int) -> list[dict]:
+    async def get_item_info(self, item_id: int) -> list[dict[str, Any]]:
         """Returns a list of the details of the specified item.
 
         Parameters:
             `item_id` - The Control4 item ID.
         """
         data = await self.send_get_request(f"/api/v1/items/{item_id}")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_item_setup(self, item_id: int) -> dict:
+    async def get_item_setup(self, item_id: int) -> dict[str, Any]:
         """Returns the setup info of the specified item.
 
         Parameters:
@@ -140,16 +143,18 @@ async def get_item_setup(self, item_id: int) -> dict:
         data = await self.send_post_request(
             f"/api/v1/items/{item_id}/commands", "GET_SETUP", {}, False
         )
-        return json.loads(data)
+        result: dict[str, Any] = json.loads(data)
+        return result
 
-    async def get_item_variables(self, item_id: int) -> list[dict]:
+    async def get_item_variables(self, item_id: int) -> list[dict[str, Any]]:
         """Returns a list of the variables available for the specified item.
 
         Parameters:
             `item_id` - The Control4 item ID.
         """
         data = await self.send_get_request(f"/api/v1/items/{item_id}/variables")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
     async def get_item_variable_value(
         self, item_id: int, var_name: str | list[str] | tuple[str, ...] | set[str]
@@ -207,40 +212,43 @@ async def get_all_item_variable_value(
                 f"Empty response received from Director! The variable {var_name} "
                 f"doesn't seem to exist for any items."
             )
-        json_dict = json.loads(data)
+        json_dict: list[dict[str, Any]] = json.loads(data)
         for item in json_dict:
             if item.get("value") == "Undefined":
                 item["value"] = None
         return json_dict
 
-    async def get_item_commands(self, item_id: int) -> list[dict]:
+    async def get_item_commands(self, item_id: int) -> list[dict[str, Any]]:
         """Returns the commands available for the specified item.
 
         Parameters:
             `item_id` - The Control4 item ID.
         """
         data = await self.send_get_request(f"/api/v1/items/{item_id}/commands")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_item_network(self, item_id: int) -> list[dict]:
+    async def get_item_network(self, item_id: int) -> list[dict[str, Any]]:
         """Returns the network information for the specified item.
 
         Parameters:
             `item_id` - The Control4 item ID.
         """
         data = await self.send_get_request(f"/api/v1/items/{item_id}/network")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_item_bindings(self, item_id: int) -> list[dict]:
+    async def get_item_bindings(self, item_id: int) -> list[dict[str, Any]]:
         """Returns the bindings information for the specified item.
 
         Parameters:
             `item_id` - The Control4 item ID.
         """
         data = await self.send_get_request(f"/api/v1/items/{item_id}/bindings")
-        return json.loads(data)
+        result: list[dict[str, Any]] = json.loads(data)
+        return result
 
-    async def get_ui_configuration(self) -> dict:
+    async def get_ui_configuration(self) -> dict[str, Any]:
         """Returns a dictionary of the Control4 App UI Configuration enumerating
         rooms and capabilities
 
@@ -323,4 +331,5 @@ async def get_ui_configuration(self) -> dict:
             }
         """
         data = await self.send_get_request("/api/v1/agents/ui_configuration")
-        return json.loads(data)
+        result: dict[str, Any] = json.loads(data)
+        return result

pyControl4/error_handling.py

@@ -2,14 +2,16 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import json
 import xmltodict
 
 
 class C4Exception(Exception):
     """Base error for pyControl4."""
 
-    def __init__(self, message):
+    def __init__(self, message: str) -> None:
         self.message = message
 
 
@@ -88,7 +90,7 @@ def _check_response_format(response_text: str) -> str:
     return "JSON"
 
 
-def _extract_error_info(dictionary: dict) -> dict | None:
+def _extract_error_info(dictionary: dict[str, Any]) -> dict[str, Any] | None:
     """Extract error information from a parsed Control4 response.
 
     Returns a dict with 'details', 'code', or 'error' key, or None if no error found.
@@ -121,7 +123,7 @@ def _extract_error_info(dictionary: dict) -> dict | None:
     return None
 
 
-def _raise_error(error_info: dict, response_text: str) -> None:
+def _raise_error(error_info: dict[str, Any], response_text: str) -> None:
     """Raise appropriate exception based on error info."""
     details = error_info.get("details")
     code = error_info.get("code")

pyControl4/py.typed

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from typing import Any
 
 from pyControl4 import C4Entity
 
@@ -139,7 +140,7 @@ async def set_stop(self) -> None:
             {},
         )
 
-    async def get_audio_devices(self) -> dict:
+    async def get_audio_devices(self) -> dict[str, Any]:
         """
         Note: As tested in OS 3.2.3 this doesn't work, but may work in previous versions
 
@@ -152,9 +153,10 @@ async def get_audio_devices(self) -> dict:
         data = await self.director.send_get_request(
             f"/api/v1/locations/rooms/{self.item_id}/audio_devices"
         )
-        return json.loads(data)
+        result: dict[str, Any] = json.loads(data)
+        return result
 
-    async def get_video_devices(self) -> dict:
+    async def get_video_devices(self) -> dict[str, Any]:
         """
         Note: As tested in OS 3.2.3 this doesn't work, but may work in previous versions
 
@@ -167,4 +169,5 @@ async def get_video_devices(self) -> dict:
         data = await self.director.send_get_request(
             f"/api/v1/locations/rooms/{self.item_id}/video_devices"
         )
-        return json.loads(data)
+        result: dict[str, Any] = json.loads(data)
+        return result

pyControl4/room.py

@@ -9,7 +9,7 @@
 
 import aiohttp
 import asyncio
-import socketio_v4 as socketio
+import socketio_v4 as socketio  # type: ignore[import-untyped]
 import logging
 
 from .error_handling import check_response_for_error
@@ -21,14 +21,14 @@
 _STATUS_ACK_MESSAGE = "2"
 
 
-class _C4DirectorNamespace(socketio.AsyncClientNamespace):
+class _C4DirectorNamespace(socketio.AsyncClientNamespace):  # type: ignore[misc]
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.url: str = kwargs.pop("url")
         self.token: str = kwargs.pop("token")
-        self.callback: Callable = kwargs.pop("callback")
+        self.callback: Callable[..., Any] = kwargs.pop("callback")
         self.session: aiohttp.ClientSession | None = kwargs.pop("session")
-        self.connect_callback: Callable | None = kwargs.pop("connect_callback")
-        self.disconnect_callback: Callable | None = kwargs.pop("disconnect_callback")
+        self.connect_callback: Callable[..., Any] | None = kwargs.pop("connect_callback")
+        self.disconnect_callback: Callable[..., Any] | None = kwargs.pop("disconnect_callback")
         super().__init__(*args, **kwargs)
         self.uri = _NAMESPACE_URI
         self.subscription_id: str | None = None
@@ -99,8 +99,8 @@ def __init__(
         self,
         ip: str,
         session_no_verify_ssl: aiohttp.ClientSession | None = None,
-        connect_callback: Callable | None = None,
-        disconnect_callback: Callable | None = None,
+        connect_callback: Callable[..., Any] | None = None,
+        disconnect_callback: Callable[..., Any] | None = None,
     ):
         """Creates a Control4 Websocket object.
 
@@ -128,21 +128,21 @@ def __init__(
         self.base_url: str = f"https://{ip}"
         self.wss_url: str = f"wss://{ip}"
         self.session: aiohttp.ClientSession | None = session_no_verify_ssl
-        self.connect_callback: Callable | None = connect_callback
-        self.disconnect_callback: Callable | None = disconnect_callback
+        self.connect_callback: Callable[..., Any] | None = connect_callback
+        self.disconnect_callback: Callable[..., Any] | None = disconnect_callback
 
-        self._item_callbacks: dict[int, list[Callable]] = dict()
+        self._item_callbacks: dict[int, list[Callable[..., Any]]] = dict()
         self._sio: socketio.AsyncClient | None = None
 
     @property
-    def item_callbacks(self) -> MappingProxyType[int, list[Callable]]:
+    def item_callbacks(self) -> MappingProxyType[int, list[Callable[..., Any]]]:
         """Returns a read-only view of registered item ids (key) and their
         callbacks (value). Use add_item_callback() or remove_item_callback()
         to modify callbacks.
         """
         return MappingProxyType(self._item_callbacks)
 
-    def add_item_callback(self, item_id: int, callback: Callable) -> None:
+    def add_item_callback(self, item_id: int, callback: Callable[..., Any]) -> None:
         """Register a callback to receive updates about an item.
         Parameters:
             `item_id` - The Control4 item ID.
@@ -159,7 +159,7 @@ def add_item_callback(self, item_id: int, callback: Callable) -> None:
             self._item_callbacks[item_id].append(callback)
 
     def remove_item_callback(
-        self, item_id: int, callback: Callable | None = None
+        self, item_id: int, callback: Callable[..., Any] | None = None
     ) -> None:
         """Unregister callback(s) for an item.
         Parameters:

pyControl4/websocket.py

@@ -4,3 +4,4 @@ python-socketio-v4
 websocket-client
 pdoc3
 pytest-asyncio
+types-xmltodict

requirements-dev.txt

@@ -5,14 +5,15 @@
 
 setuptools.setup(
     name="pyControl4",  # Replace with your own username
-    version="2.0.0",
+    version="2.0.1",
     author="lawtancool",
     author_email="contact@lawrencetan.ca",
     description="Python 3 asyncio package for interacting with Control4 systems",
     long_description=long_description,
     long_description_content_type="text/markdown",
     url="https://github.com/lawtancool/pyControl4",
     packages=setuptools.find_packages(),
+    package_data={"pyControl4": ["py.typed"]},
     classifiers=[
         "Programming Language :: Python :: 3",
         "License :: OSI Approved :: Apache Software License",