feat: add typing

.bin/run-mypy

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+poetry run mypy src/uvcclient

.pre-commit-config.yaml

@@ -6,6 +6,7 @@ default_stages: [commit]
 ci:
   autofix_commit_msg: "chore(pre-commit.ci): auto fixes"
   autoupdate_commit_msg: "chore(pre-commit.ci): pre-commit autoupdate"
+  skip: [mypy]
 
 repos:
   - repo: https://github.com/commitizen-tools/commitizen
@@ -40,3 +41,12 @@ repos:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
       - id: ruff-format
+  - repo: local
+    hooks:
+      - id: mypy
+        name: mypy
+        language: script
+        entry: ./.bin/run-mypy
+        types_or: [python, pyi]
+        require_serial: true
+        files: ^(src/uvcclient)/.+\.(py|pyi)$

pyproject.toml

@@ -39,6 +39,7 @@ pytest-sugar = "1.0.0"
 pytest-timeout = "2.3.1"
 pytest-xdist = "3.6.1"
 pytest-cov = "^5.0.0"
+mypy = "^1.11.1"
 
 [tool.semantic_release]
 version_toml = ["pyproject.toml:tool.poetry.version"]
@@ -134,3 +135,24 @@ select = [
 
 [tool.ruff.lint.isort]
 known-first-party = ["uvcclient", "tests"]
+
+[tool.mypy]
+disable_error_code = "import-untyped,unused-ignore"
+check_untyped_defs = true
+ignore_missing_imports = true
+disallow_any_generics = true
+disallow_incomplete_defs = true
+disallow_untyped_defs = true
+mypy_path = "src/"
+no_implicit_optional = true
+show_error_codes = true
+warn_unreachable = true
+warn_unused_ignores = true
+
+[[tool.mypy.overrides]]
+module = "tests.*"
+allow_untyped_defs = true
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"

src/uvcclient/camera.py

@@ -15,18 +15,9 @@
 
 import json
 import logging
-
-# Python3 compatibility
-try:
-    import httplib
-except ImportError:
-    from http import client as httplib
-try:
-    import urllib
-
-    import urlparse
-except ImportError:
-    import urllib.parse as urlparse
+import urllib.parse as urlparse
+from http import client as httplib
+from typing import Any
 
 
 class CameraConnectError(Exception):
@@ -38,15 +29,15 @@ class CameraAuthError(Exception):
 
 
 class UVCCameraClient:
-    def __init__(self, host, username, password, port=80):
+    def __init__(self, host: str, username: str, password: str, port: int = 80) -> None:
         self._host = host
         self._port = port
         self._username = username
         self._password = password
         self._cookie = ""
         self._log = logging.getLogger(f"UVCCamera({self._host})")
 
-    def _safe_request(self, *args, **kwargs):
+    def _safe_request(self, *args: Any, **kwargs: Any) -> httplib.HTTPResponse:
         try:
             conn = httplib.HTTPConnection(self._host, self._port)
             conn.request(*args, **kwargs)
@@ -56,7 +47,7 @@ def _safe_request(self, *args, **kwargs):
         except httplib.HTTPException as ex:
             raise CameraConnectError(f"Error connecting to camera: {ex!s}") from ex
 
-    def login(self):
+    def login(self) -> None:
         resp = self._safe_request("GET", "/")
         headers = dict(resp.getheaders())
         try:
@@ -65,10 +56,7 @@ def login(self):
             self._cookie = headers["set-cookie"]
         session = self._cookie.split("=")[1].split(";")[0]
 
-        try:
-            urlencode = urllib.urlencode
-        except AttributeError:
-            urlencode = urlparse.urlencode
+        urlencode = urlparse.urlencode
 
         data = urlencode(
             {
@@ -86,30 +74,30 @@ def login(self):
         if resp.status != 200:
             raise CameraAuthError(f"Failed to login: {resp.reason}")
 
-    def _cfgwrite(self, setting, value):
+    def _cfgwrite(self, setting: str, value: str | int) -> bool:
         headers = {"Cookie": self._cookie}
         resp = self._safe_request(
             "GET", f"/cfgwrite.cgi?{setting}={value}", headers=headers
         )
         self._log.debug(f"Setting {setting}={value}: {resp.status} {resp.reason}")
         return resp.status == 200
 
-    def set_led(self, enabled):
+    def set_led(self, enabled: bool) -> bool:
         return self._cfgwrite("led.front.status", int(enabled))
 
     @property
-    def snapshot_url(self):
+    def snapshot_url(self) -> str:
         return "/snapshot.cgi"
 
     @property
-    def reboot_url(self):
+    def reboot_url(self) -> str:
         return "/api/1.1/reboot"
 
     @property
-    def status_url(self):
+    def status_url(self) -> str:
         return "/api/1.1/status"
 
-    def get_snapshot(self):
+    def get_snapshot(self) -> bytes:
         headers = {"Cookie": self._cookie}
         resp = self._safe_request("GET", self.snapshot_url, headers=headers)
         if resp.status in (401, 403, 302):
@@ -118,15 +106,15 @@ def get_snapshot(self):
             raise CameraConnectError(f"Snapshot failed: {resp.status}")
         return resp.read()
 
-    def reboot(self):
+    def reboot(self) -> None:
         headers = {"Cookie": self._cookie}
         resp = self._safe_request("GET", self.reboot_url, headers=headers)
         if resp.status in (401, 403, 302):
             raise CameraAuthError("Not logged in")
         elif resp.status != 200:
             raise CameraConnectError(f"Reboot failed: {resp.status}")
 
-    def get_status(self):
+    def get_status(self) -> dict[str, Any]:
         headers = {"Cookie": self._cookie}
         resp = self._safe_request("GET", self.status_url, headers=headers)
         if resp.status in (401, 403, 302):
@@ -138,10 +126,10 @@ def get_status(self):
 
 class UVCCameraClientV320(UVCCameraClient):
     @property
-    def snapshot_url(self):
+    def snapshot_url(self) -> str:
         return "/snap.jpeg"
 
-    def login(self):
+    def login(self) -> None:
         headers = {"Content-Type": "application/json"}
         data = json.dumps({"username": self._username, "password": self._password})
         resp = self._safe_request("POST", "/api/1.1/login", data, headers=headers)

src/uvcclient/main.py

@@ -18,14 +18,15 @@
 import logging
 import optparse
 import sys
+from typing import Any
 
 from . import camera, nvr, store
-from .nvr import Invalid
+from .nvr import Invalid, UVCRemote
 
 INFO_STORE = store.get_info_store()
 
 
-def do_led(camera_info, enabled):
+def do_led(camera_info: dict[str, Any], enabled: bool) -> None:
     password = INFO_STORE.get_camera_password(camera_info["uuid"]) or "ubnt"
     cam_client = camera.UVCCameraClient(
         camera_info["host"], camera_info["username"], password
@@ -34,8 +35,9 @@ def do_led(camera_info, enabled):
     cam_client.set_led(enabled)
 
 
-def do_snapshot(client, camera_info):
+def do_snapshot(client: UVCRemote, camera_info: dict[str, Any]) -> bytes:
     password = INFO_STORE.get_camera_password(camera_info["uuid"]) or "ubnt"
+    cam_client: camera.UVCCameraClient
     if client.server_version >= (3, 2, 0):
         cam_client = camera.UVCCameraClientV320(
             camera_info["host"], camera_info["username"], password
@@ -52,8 +54,9 @@ def do_snapshot(client, camera_info):
         return client.get_snapshot(camera_info["uuid"])
 
 
-def do_reboot(client, camera_info):
+def do_reboot(client: UVCRemote, camera_info: dict[str, Any]) -> None:
     password = INFO_STORE.get_camera_password(camera_info["uuid"]) or "ubnt"
+    cam_client: camera.UVCCameraClient
     if client.server_version >= (3, 2, 0):
         cam_client = camera.UVCCameraClientV320(
             camera_info["host"], camera_info["username"], password
@@ -73,7 +76,7 @@ def do_reboot(client, camera_info):
         print(f"Failed to reboot: {e}")
 
 
-def do_set_password(opts):
+def do_set_password(opts: optparse.Values) -> None:
     print("This will store the administrator password for a camera ")
     print("for later use. It will be stored on disk obscured, but ")
     print("NOT ENCRYPTED! If this is not okay, cancel now.")
@@ -87,7 +90,7 @@ def do_set_password(opts):
     print("Password set")
 
 
-def main():
+def main() -> int:
     host, port, apikey, path = nvr.get_auth_from_env()
 
     parser = optparse.OptionParser()
@@ -160,7 +163,7 @@ def main():
 
     if not all([host, port, apikey]):
         print("Host, port, and apikey are required")
-        return
+        return 1
 
     if opts.verbose:
         level = logging.DEBUG
@@ -174,7 +177,7 @@ def main():
         opts.uuid = client.name_to_uuid(opts.name)
         if not opts.uuid:
             print(f"`{opts.name}' is not a valid name")
-            return
+            return 1
 
     if opts.dump:
         client.dump(opts.uuid)
@@ -211,9 +214,9 @@ def main():
         if not opts.uuid:
             print("Name or UUID is required")
             return 1
-        r = client.get_recordmode(opts.uuid)
-        print(r)
-        return r == "none"
+        res = client.get_recordmode(opts.uuid)
+        print(res)
+        return res == "none"
     elif opts.get_picture_settings:
         settings = client.get_picture_settings(opts.uuid)
         print(",".join([f"{k}={v}" for k, v in settings.items()]))
@@ -262,10 +265,7 @@ def main():
         if not camera:
             print("No such camera")
             return 1
-        if hasattr(sys.stdout, "buffer"):
-            sys.stdout.buffer.write(do_snapshot(client, camera))
-        else:
-            sys.stdout.write(do_snapshot(client, camera))
+        sys.stdout.buffer.write(do_snapshot(client, camera))
     elif opts.reboot:
         camera = client.get_camera(opts.uuid)
         if not camera:
@@ -276,3 +276,4 @@ def main():
         do_set_password(opts)
     else:
         print("No action specified; try --help")
+    return 0