diff --git a/.bin/run-mypy b/.bin/run-mypy new file mode 100755 index 0000000..33e359b --- /dev/null +++ b/.bin/run-mypy @@ -0,0 +1,3 @@ +#!/bin/sh + +poetry run mypy src/uvcclient diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 44b94c6..f49b9b1 100644 --- a/.pre-commit-config.yaml +++ b/.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)$ diff --git a/poetry.lock b/poetry.lock index fdcbda6..4b8f026 100644 --- a/poetry.lock +++ b/poetry.lock @@ -137,6 +137,64 @@ files = [ {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"}, ] +[[package]] +name = "mypy" +version = "1.11.1" +description = "Optional static typing for Python" +optional = false +python-versions = ">=3.8" +files = [ + {file = "mypy-1.11.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:a32fc80b63de4b5b3e65f4be82b4cfa362a46702672aa6a0f443b4689af7008c"}, + {file = "mypy-1.11.1-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:c1952f5ea8a5a959b05ed5f16452fddadbaae48b5d39235ab4c3fc444d5fd411"}, + {file = "mypy-1.11.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e1e30dc3bfa4e157e53c1d17a0dad20f89dc433393e7702b813c10e200843b03"}, + {file = "mypy-1.11.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:2c63350af88f43a66d3dfeeeb8d77af34a4f07d760b9eb3a8697f0386c7590b4"}, + {file = "mypy-1.11.1-cp310-cp310-win_amd64.whl", hash = "sha256:a831671bad47186603872a3abc19634f3011d7f83b083762c942442d51c58d58"}, + {file = "mypy-1.11.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:7b6343d338390bb946d449677726edf60102a1c96079b4f002dedff375953fc5"}, + {file = "mypy-1.11.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e4fe9f4e5e521b458d8feb52547f4bade7ef8c93238dfb5bbc790d9ff2d770ca"}, + {file = "mypy-1.11.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:886c9dbecc87b9516eff294541bf7f3655722bf22bb898ee06985cd7269898de"}, + {file = "mypy-1.11.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:fca4a60e1dd9fd0193ae0067eaeeb962f2d79e0d9f0f66223a0682f26ffcc809"}, + {file = "mypy-1.11.1-cp311-cp311-win_amd64.whl", hash = "sha256:0bd53faf56de9643336aeea1c925012837432b5faf1701ccca7fde70166ccf72"}, + {file = "mypy-1.11.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:f39918a50f74dc5969807dcfaecafa804fa7f90c9d60506835036cc1bc891dc8"}, + {file = "mypy-1.11.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:0bc71d1fb27a428139dd78621953effe0d208aed9857cb08d002280b0422003a"}, + {file = "mypy-1.11.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b868d3bcff720dd7217c383474008ddabaf048fad8d78ed948bb4b624870a417"}, + {file = "mypy-1.11.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:a707ec1527ffcdd1c784d0924bf5cb15cd7f22683b919668a04d2b9c34549d2e"}, + {file = "mypy-1.11.1-cp312-cp312-win_amd64.whl", hash = "sha256:64f4a90e3ea07f590c5bcf9029035cf0efeae5ba8be511a8caada1a4893f5525"}, + {file = "mypy-1.11.1-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:749fd3213916f1751fff995fccf20c6195cae941dc968f3aaadf9bb4e430e5a2"}, + {file = "mypy-1.11.1-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:b639dce63a0b19085213ec5fdd8cffd1d81988f47a2dec7100e93564f3e8fb3b"}, + {file = "mypy-1.11.1-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4c956b49c5d865394d62941b109728c5c596a415e9c5b2be663dd26a1ff07bc0"}, + {file = "mypy-1.11.1-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:45df906e8b6804ef4b666af29a87ad9f5921aad091c79cc38e12198e220beabd"}, + {file = "mypy-1.11.1-cp38-cp38-win_amd64.whl", hash = "sha256:d44be7551689d9d47b7abc27c71257adfdb53f03880841a5db15ddb22dc63edb"}, + {file = "mypy-1.11.1-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:2684d3f693073ab89d76da8e3921883019ea8a3ec20fa5d8ecca6a2db4c54bbe"}, + {file = "mypy-1.11.1-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:79c07eb282cb457473add5052b63925e5cc97dfab9812ee65a7c7ab5e3cb551c"}, + {file = "mypy-1.11.1-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:11965c2f571ded6239977b14deebd3f4c3abd9a92398712d6da3a772974fad69"}, + {file = "mypy-1.11.1-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:a2b43895a0f8154df6519706d9bca8280cda52d3d9d1514b2d9c3e26792a0b74"}, + {file = "mypy-1.11.1-cp39-cp39-win_amd64.whl", hash = "sha256:1a81cf05975fd61aec5ae16501a091cfb9f605dc3e3c878c0da32f250b74760b"}, + {file = "mypy-1.11.1-py3-none-any.whl", hash = "sha256:0624bdb940255d2dd24e829d99a13cfeb72e4e9031f9492148f410ed30bcab54"}, + {file = "mypy-1.11.1.tar.gz", hash = "sha256:f404a0b069709f18bbdb702eb3dcfe51910602995de00bd39cea3050b5772d08"}, +] + +[package.dependencies] +mypy-extensions = ">=1.0.0" +tomli = {version = ">=1.1.0", markers = "python_version < \"3.11\""} +typing-extensions = ">=4.6.0" + +[package.extras] +dmypy = ["psutil (>=4.0)"] +install-types = ["pip"] +mypyc = ["setuptools (>=50)"] +reports = ["lxml"] + +[[package]] +name = "mypy-extensions" +version = "1.0.0" +description = "Type system extensions for programs checked with the mypy type checker." +optional = false +python-versions = ">=3.5" +files = [ + {file = "mypy_extensions-1.0.0-py3-none-any.whl", hash = "sha256:4392f6c0eb8a5668a69e23d168ffa70f0be9ccfd32b5cc2d26a34ae5b844552d"}, + {file = "mypy_extensions-1.0.0.tar.gz", hash = "sha256:75dbf8955dc00442a438fc4d0666508a9a97b6bd41aa2f0ffe9d2f2725af0782"}, +] + [[package]] name = "packaging" version = "24.1" @@ -330,7 +388,18 @@ files = [ {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"}, ] +[[package]] +name = "typing-extensions" +version = "4.12.2" +description = "Backported and Experimental Type Hints for Python 3.8+" +optional = false +python-versions = ">=3.8" +files = [ + {file = "typing_extensions-4.12.2-py3-none-any.whl", hash = "sha256:04e5ca0351e0f3f85c6853954072df659d0d13fac324d0072316b67d7794700d"}, + {file = "typing_extensions-4.12.2.tar.gz", hash = "sha256:1a7ead55c7e559dd4dee8856e3a88b41225abfe1ce8df57b7c13915fe121ffb8"}, +] + [metadata] lock-version = "2.0" python-versions = ">=3.10" -content-hash = "283afb2b43d0d11e08076449c4ab8e99429122e287f87a744fb1d023b64a7d07" +content-hash = "4edb71e2be2173b21e1d488fbfc65436a311e24b0035290041f1df619ccdb539" diff --git a/pyproject.toml b/pyproject.toml index 6fe8e8f..3e48a45 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/uvcclient/camera.py b/src/uvcclient/camera.py index db60255..2eddf69 100644 --- a/src/uvcclient/camera.py +++ b/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,7 +29,7 @@ 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 @@ -46,7 +37,7 @@ def __init__(self, host, username, password, port=80): 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,7 +74,7 @@ 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 @@ -94,22 +82,22 @@ def _cfgwrite(self, setting, value): 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,7 +106,7 @@ 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): @@ -126,7 +114,7 @@ def reboot(self): 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) diff --git a/src/uvcclient/main.py b/src/uvcclient/main.py index fc8970c..63bc928 100644 --- a/src/uvcclient/main.py +++ b/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 diff --git a/src/uvcclient/nvr.py b/src/uvcclient/nvr.py index 1551778..4de1d21 100755 --- a/src/uvcclient/nvr.py +++ b/src/uvcclient/nvr.py @@ -20,17 +20,10 @@ import logging import os import pprint +import urllib.parse as urlparse import zlib - -# Python3 compatibility -try: - import httplib -except ImportError: - from http import client as httplib -try: - import urlparse -except ImportError: - import urllib.parse as urlparse +from http import client as httplib +from typing import Any, Literal class Invalid(Exception): @@ -54,7 +47,9 @@ class UVCRemote: CHANNEL_NAMES = ["high", "medium", "low"] - def __init__(self, host, port, apikey, path="/", ssl=False): + def __init__( + self, host: str, port: int, apikey: str, path: str = "/", ssl: bool = False + ) -> None: self._host = host self._port = port self._path = path @@ -68,7 +63,7 @@ def __init__(self, host, port, apikey, path="/", ssl=False): self._log.debug(f"Server version is {version}") @property - def server_version(self): + def server_version(self) -> tuple[int, int, int]: version = self._bootstrap["systemInfo"]["version"].split(".") major = int(version[0]) minor = int(version[1]) @@ -79,19 +74,19 @@ def server_version(self): return (major, minor, rev) @property - def camera_identifier(self): + def camera_identifier(self) -> str: if self.server_version >= (3, 2, 0): return "id" else: return "uuid" - def _get_http_connection(self): + def _get_http_connection(self) -> httplib.HTTPConnection: if self._ssl: return httplib.HTTPSConnection(self._host, self._port) else: return httplib.HTTPConnection(self._host, self._port) - def _safe_request(self, *args, **kwargs): + def _safe_request(self, *args: Any, **kwargs: Any) -> httplib.HTTPResponse: try: conn = self._get_http_connection() conn.request(*args, **kwargs) @@ -101,7 +96,7 @@ def _safe_request(self, *args, **kwargs): except httplib.HTTPException as ex: raise CameraConnectionError(f"Error connecting to camera: {ex!s}") from ex - def _uvc_request(self, *args, **kwargs): + def _uvc_request(self, *args: Any, **kwargs: Any) -> dict[str, Any]: try: return self._uvc_request_safe(*args, **kwargs) except OSError as ex: @@ -110,8 +105,12 @@ def _uvc_request(self, *args, **kwargs): raise NvrError(f"Error connecting to camera: {ex!s}") from ex def _uvc_request_safe( - self, path, method="GET", data=None, mimetype="application/json" - ): + self, + path: str, + method: str = "GET", + data: dict[str, Any] | None = None, + mimetype: str = "application/json", + ) -> dict[str, Any]: conn = self._get_http_connection() if "?" in path: url = f"{path}&apiKey={self._apikey}" @@ -124,7 +123,10 @@ def _uvc_request_safe( "Accept-Encoding": "gzip, deflate, sdch", } self._log.debug(f"{method} {url} headers={headers} data={data!r}") - conn.request(method, url, data, headers) + body = None + if data: + body = json.dumps(data) + conn.request(method, url, body, headers) resp = conn.getresponse() headers = dict(resp.getheaders()) self._log.debug(f"{method} {url} Result: {resp.status} {resp.reason}") @@ -133,23 +135,23 @@ def _uvc_request_safe( if resp.status / 100 != 2: raise NvrError(f"Request failed: {resp.status}") - data = resp.read() + res = resp.read() if ( headers.get("content-encoding") == "gzip" or headers.get("Content-Encoding") == "gzip" ): - data = zlib.decompress(data, 32 + zlib.MAX_WBITS) - return json.loads(data.decode()) + res = zlib.decompress(res, 32 + zlib.MAX_WBITS) + return json.loads(res.decode()) - def _get_bootstrap(self): + def _get_bootstrap(self) -> dict[str, Any]: return self._uvc_request("/api/2.0/bootstrap")["data"][0] - def dump(self, uuid): + def dump(self, uuid: str) -> None: """Dump information for a camera by UUID.""" data = self._uvc_request(f"/api/2.0/camera/{uuid}") pprint.pprint(data) - def set_recordmode(self, uuid, mode, chan=None): + def set_recordmode(self, uuid: str, mode: str, chan: str | None = None) -> bool: """ Set the recording mode for a camera by UUID. @@ -181,7 +183,7 @@ def set_recordmode(self, uuid, mode, chan=None): updated = data["data"][0]["recordingSettings"] return settings == updated - def get_recordmode(self, uuid): + def get_recordmode(self, uuid: str) -> Literal["none", "full", "motion"]: url = f"/api/2.0/camera/{uuid}" data = self._uvc_request(url) recmodes = data["data"][0]["recordingSettings"] @@ -192,12 +194,14 @@ def get_recordmode(self, uuid): else: return "none" - def get_picture_settings(self, uuid): + def get_picture_settings(self, uuid: str) -> dict[str, Any]: url = f"/api/2.0/camera/{uuid}" data = self._uvc_request(url) return data["data"][0]["ispSettings"] - def set_picture_settings(self, uuid, settings): + def set_picture_settings( + self, uuid: str, settings: dict[str, Any] + ) -> dict[str, Any]: url = f"/api/2.0/camera/{uuid}" data = self._uvc_request(url) for key in settings: @@ -211,18 +215,18 @@ def set_picture_settings(self, uuid, settings): data = self._uvc_request(url, "PUT", json.dumps(data["data"][0])) return data["data"][0]["ispSettings"] - def prune_zones(self, uuid): + def prune_zones(self, uuid: str) -> None: url = f"/api/2.0/camera/{uuid}" data = self._uvc_request(url) data["data"][0]["zones"] = [data["data"][0]["zones"][0]] self._uvc_request(url, "PUT", json.dumps(data["data"][0])) - def list_zones(self, uuid): + def list_zones(self, uuid: str) -> list[dict[str, Any]]: url = f"/api/2.0/camera/{uuid}" data = self._uvc_request(url) return data["data"][0]["zones"] - def index(self): + def index(self) -> list[dict[str, Any]]: """ Return an index of available cameras. @@ -241,7 +245,7 @@ def index(self): if not x["deleted"] ] - def name_to_uuid(self, name): + def name_to_uuid(self, name: str) -> str | None: """ Attempt to convert a camera name to its UUID. @@ -256,10 +260,10 @@ def name_to_uuid(self, name): cams_by_name = {x["name"]: x["uuid"] for x in cameras} return cams_by_name.get(name) - def get_camera(self, uuid): + def get_camera(self, uuid: str) -> dict[str, Any]: return self._uvc_request(f"/api/2.0/camera/{uuid}")["data"][0] - def get_snapshot(self, uuid): + def get_snapshot(self, uuid: str) -> bytes: url = f"/api/2.0/snapshot/camera/{uuid}?force=true&apiKey={self._apikey}" print(url) resp = self._safe_request("GET", url) @@ -268,7 +272,7 @@ def get_snapshot(self, uuid): return resp.read() -def get_auth_from_env(): +def get_auth_from_env() -> tuple[str | None, int, str | None, str]: """ Attempt to get UVC NVR connection information from the environment. @@ -289,16 +293,17 @@ def get_auth_from_env(): # http://192.168.1.1:7080/apikey result = urlparse.urlparse(combined) if ":" in result.netloc: - host, port = result.netloc.split(":", 1) - port = int(port) + host, found_port = result.netloc.split(":", 1) + port = int(found_port) else: host = result.netloc port = 7080 apikey = urlparse.parse_qs(result.query)["apiKey"][0] path = result.path + return host, port, apikey, path else: - host = os.getenv("UVC_HOST") - port = int(os.getenv("UVC_PORT", 7080)) - apikey = os.getenv("UVC_APIKEY") - path = "/" - return host, port, apikey, path + env_host = os.getenv("UVC_HOST") + env_port = int(os.getenv("UVC_PORT", 7080)) + env_apikey = os.getenv("UVC_APIKEY") + env_path = "/" + return env_host, env_port, env_apikey, env_path diff --git a/src/uvcclient/py.typed b/src/uvcclient/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/uvcclient/store.py b/src/uvcclient/store.py index 5d87784..d833eb0 100644 --- a/src/uvcclient/store.py +++ b/src/uvcclient/store.py @@ -2,6 +2,7 @@ import json import logging import os +from typing import Any LOG = logging.getLogger(__name__) _INFO_STORE = None @@ -12,13 +13,15 @@ class UnableToManageStore(Exception): class InfoStore: - def __init__(self, path=None): + _data: dict[str, Any] + + def __init__(self, path: str | None = None) -> None: if path is None: path = os.path.expanduser(os.path.join("~", ".uvcclient")) self._path = path self.load() - def load(self): + def load(self) -> None: try: with open(self._path) as f: self._data = json.loads(base64.b64decode(f.read()).decode()) @@ -29,7 +32,7 @@ def load(self): LOG.error("Failed to read store data: %s", ex) raise UnableToManageStore("Unable to write to store") from ex - def save(self): + def save(self) -> None: try: with open(self._path, "w") as f: f.write(base64.b64encode(json.dumps(self._data).encode()).decode()) @@ -38,20 +41,20 @@ def save(self): LOG.error("Unable to write store: %s", str(ex)) raise UnableToManageStore("Unable to write to store") from ex - def get_camera_passwords(self): + def get_camera_passwords(self) -> dict[str, str]: return self._data.get("camera_passwords", {}) - def get_camera_password(self, uuid): + def get_camera_password(self, uuid: str) -> str | None: return self.get_camera_passwords().get(uuid) - def set_camera_password(self, uuid, password): + def set_camera_password(self, uuid: str, password: str) -> None: if "camera_passwords" not in self._data: self._data["camera_passwords"] = {} self._data["camera_passwords"][uuid] = password self.save() -def get_info_store(path=None): +def get_info_store(path: str | None = None) -> InfoStore: global _INFO_STORE if _INFO_STORE is None: _INFO_STORE = InfoStore(path) diff --git a/tests/test_camera.py b/tests/test_camera.py index da6f261..205849e 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -1,10 +1,6 @@ -try: - import httplib -except ImportError: - from http import client as httplib - import json import unittest +from http import client as httplib from unittest import mock from uvcclient import camera diff --git a/tests/test_client.py b/tests/test_client.py index 584e229..0793e7e 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,11 +1,7 @@ -try: - import httplib -except ImportError: - from http import client as httplib - import json import unittest import zlib +from http import client as httplib from unittest import mock from uvcclient import nvr @@ -56,7 +52,7 @@ def test_uvc_request_put(self): resp = conn.getresponse.return_value resp.status = 200 resp.read.return_value = json.dumps({}).encode() - result = client._uvc_request("/bar?foo=bar", method="PUT", data="foobar") + result = client._uvc_request("/bar?foo=bar", method="PUT", data={"foo": "bar"}) self.assertEqual({}, result) headers = { "Content-Type": "application/json", @@ -64,7 +60,7 @@ def test_uvc_request_put(self): "Accept-Encoding": "gzip, deflate, sdch", } conn.request.assert_called_once_with( - "PUT", "/bar?foo=bar&apiKey=key", "foobar", headers + "PUT", "/bar?foo=bar&apiKey=key", '{"foo": "bar"}', headers ) def test_uvc_request_failed(self): @@ -73,7 +69,7 @@ def test_uvc_request_failed(self): resp = conn.getresponse.return_value resp.status = 404 self.assertRaises( - nvr.NvrError, client._uvc_request, "/bar", method="PUT", data="foobar" + nvr.NvrError, client._uvc_request, "/bar", method="PUT", data={"foo": "bar"} ) def test_uvc_request_failed_noauth(self): @@ -82,7 +78,11 @@ def test_uvc_request_failed_noauth(self): resp = conn.getresponse.return_value resp.status = 401 self.assertRaises( - nvr.NotAuthorized, client._uvc_request, "/bar", method="PUT", data="foobar" + nvr.NotAuthorized, + client._uvc_request, + "/bar", + method="PUT", + data={"foo": "bar"}, ) def test_uvc_request_deflated(self): diff --git a/tests/test_store.py b/tests/test_store.py index 5136240..9067dcd 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -1,14 +1,10 @@ +import builtins import contextlib import unittest from unittest import mock from uvcclient import store -try: - import __builtin__ as builtins -except ImportError: - import builtins - class OpenHelper: def __init__(self):