From 8a5f9b644b80a2f739bc5d9720e316150e938ab6 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 9 Jun 2024 12:42:02 -0500 Subject: [PATCH] fix: Only pyupgrade non-typer code --- pyproject.toml | 1 + src/uiprotect/api.py | 128 ++++---- src/uiprotect/cli/__init__.py | 2 +- src/uiprotect/data/base.py | 148 ++++----- src/uiprotect/data/bootstrap.py | 30 +- src/uiprotect/data/convert.py | 6 +- src/uiprotect/data/devices.py | 434 +++++++++++++-------------- src/uiprotect/data/nvr.py | 302 +++++++++---------- src/uiprotect/data/types.py | 10 +- src/uiprotect/data/user.py | 42 +-- src/uiprotect/data/websocket.py | 16 +- src/uiprotect/stream.py | 12 +- src/uiprotect/test_util/__init__.py | 58 ++-- src/uiprotect/test_util/anonymize.py | 10 +- src/uiprotect/utils.py | 32 +- src/uiprotect/websocket.py | 8 +- 16 files changed, 620 insertions(+), 619 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 153c7f0f..9949488d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -131,6 +131,7 @@ ignore = [ "B008", # Do not perform function call "S110", # `try`-`except`-`pass` detected, consider logging the exception "D106", # Missing docstring in public nested class + "UP007", # typer needs Optional syntax ] select = [ "B", # flake8-bugbear diff --git a/src/uiprotect/api.py b/src/uiprotect/api.py index 70519939..fd95f069 100644 --- a/src/uiprotect/api.py +++ b/src/uiprotect/api.py @@ -14,7 +14,7 @@ from http.cookies import Morsel, SimpleCookie from ipaddress import IPv4Address, IPv6Address from pathlib import Path -from typing import Any, Literal, Optional, Union, cast +from typing import Any, Literal, cast from urllib.parse import urljoin from uuid import UUID @@ -159,12 +159,12 @@ class BaseApiClient: _last_update: float = NEVER_RAN _last_ws_status: bool = False _last_token_cookie: Morsel[str] | None = None - _last_token_cookie_decode: Optional[dict[str, Any]] = None - _session: Optional[aiohttp.ClientSession] = None + _last_token_cookie_decode: dict[str, Any] | None = None + _session: aiohttp.ClientSession | None = None _loaded_session: bool = False - headers: Optional[dict[str, str]] = None - _websocket: Optional[Websocket] = None + headers: dict[str, str] | None = None + _websocket: Websocket | None = None api_path: str = "/proxy/protect/api/" ws_path: str = "/proxy/protect/ws/updates" @@ -180,10 +180,10 @@ def __init__( username: str, password: str, verify_ssl: bool = True, - session: Optional[aiohttp.ClientSession] = None, + session: aiohttp.ClientSession | None = None, ws_timeout: int = 30, - cache_dir: Optional[Path] = None, - config_dir: Optional[Path] = None, + cache_dir: Path | None = None, + config_dir: Path | None = None, store_sessions: bool = True, ) -> None: self._auth_lock = asyncio.Lock() @@ -243,7 +243,7 @@ async def get_session(self) -> aiohttp.ClientSession: async def get_websocket(self) -> Websocket: """Gets or creates current Websocket.""" - async def _auth(force: bool) -> Optional[dict[str, str]]: + async def _auth(force: bool) -> dict[str, str] | None: if force: if self._session is not None: self._session.cookie_jar.clear() @@ -349,7 +349,7 @@ async def api_request_raw( require_auth: bool = True, raise_exception: bool = True, **kwargs: Any, - ) -> Optional[bytes]: + ) -> bytes | None: """Make a request to UniFi Protect API""" url = urljoin(self.api_path, url) response = await self.request( @@ -373,7 +373,7 @@ async def api_request_raw( _LOGGER.debug(msg, url, response.status, reason) return None - data: Optional[bytes] = await response.read() + data: bytes | None = await response.read() response.release() return data @@ -390,7 +390,7 @@ async def api_request( require_auth: bool = True, raise_exception: bool = True, **kwargs: Any, - ) -> Optional[Union[list[Any], dict[str, Any]]]: + ) -> list[Any] | dict[str, Any] | None: data = await self.api_request_raw( url=url, method=method, @@ -400,7 +400,7 @@ async def api_request( ) if data is not None: - json_data: Union[list[Any], dict[str, Any]] = orjson.loads(data) + json_data: list[Any] | dict[str, Any] = orjson.loads(data) return json_data return None @@ -647,7 +647,7 @@ def check_ws(self) -> bool: def _process_ws_message(self, msg: aiohttp.WSMessage) -> None: raise NotImplementedError - def _get_last_update_id(self) -> Optional[UUID]: + def _get_last_update_id(self) -> UUID | None: raise NotImplementedError @@ -687,9 +687,9 @@ class ProtectApiClient(BaseApiClient): _subscribed_models: set[ModelType] _ignore_stats: bool _ws_subscriptions: list[Callable[[WSSubscriptionMessage], None]] - _bootstrap: Optional[Bootstrap] = None - _last_update_dt: Optional[datetime] = None - _connection_host: Optional[Union[IPv4Address, IPv6Address, str]] = None + _bootstrap: Bootstrap | None = None + _last_update_dt: datetime | None = None + _connection_host: IPv4Address | IPv6Address | str | None = None ignore_unadopted: bool @@ -700,14 +700,14 @@ def __init__( username: str, password: str, verify_ssl: bool = True, - session: Optional[aiohttp.ClientSession] = None, + session: aiohttp.ClientSession | None = None, ws_timeout: int = 30, - cache_dir: Optional[Path] = None, - config_dir: Optional[Path] = None, + cache_dir: Path | None = None, + config_dir: Path | None = None, store_sessions: bool = True, override_connection_host: bool = False, minimum_score: int = 0, - subscribed_models: Optional[set[ModelType]] = None, + subscribed_models: set[ModelType] | None = None, ignore_stats: bool = False, ignore_unadopted: bool = True, debug: bool = False, @@ -749,7 +749,7 @@ def bootstrap(self) -> Bootstrap: return self._bootstrap @property - def connection_host(self) -> Union[IPv4Address, IPv6Address, str]: + def connection_host(self) -> IPv4Address | IPv6Address | str: """Connection host to use for generating RTSP URLs""" if self._connection_host is None: # fallback if cannot find user supplied host @@ -767,7 +767,7 @@ def connection_host(self) -> Union[IPv4Address, IPv6Address, str]: return self._connection_host - async def update(self, force: bool = False) -> Optional[Bootstrap]: + async def update(self, force: bool = False) -> Bootstrap | None: """ Updates the state of devices, initalizes `.bootstrap` and connects to UFP Websocket for real time updates @@ -833,7 +833,7 @@ def emit_message(self, msg: WSSubscriptionMessage) -> None: except Exception: _LOGGER.exception("Exception while running subscription handler") - def _get_last_update_id(self) -> Optional[UUID]: + def _get_last_update_id(self) -> UUID | None: if self._bootstrap is None: return None return self._bootstrap.last_update_id @@ -859,7 +859,7 @@ async def _get_event_paginate( params: dict[str, Any], *, start: datetime, - end: Optional[datetime], + end: datetime | None, ) -> list[dict[str, Any]]: start_int = to_js_time(start) end_int = to_js_time(end) if end else None @@ -915,15 +915,15 @@ async def _get_event_paginate( async def get_events_raw( self, *, - start: Optional[datetime] = None, - end: Optional[datetime] = None, - limit: Optional[int] = None, - offset: Optional[int] = None, - types: Optional[list[EventType]] = None, - smart_detect_types: Optional[list[SmartDetectObjectType]] = None, + start: datetime | None = None, + end: datetime | None = None, + limit: int | None = None, + offset: int | None = None, + types: list[EventType] | None = None, + smart_detect_types: list[SmartDetectObjectType] | None = None, sorting: Literal["asc", "desc"] = "asc", descriptions: bool = True, - all_cameras: Optional[bool] = None, + all_cameras: bool | None = None, category: EventCategories | None = None, # used for testing _allow_manual_paginate: bool = True, @@ -1010,12 +1010,12 @@ async def get_events_raw( async def get_events( self, - start: Optional[datetime] = None, - end: Optional[datetime] = None, - limit: Optional[int] = None, - offset: Optional[int] = None, - types: Optional[list[EventType]] = None, - smart_detect_types: Optional[list[SmartDetectObjectType]] = None, + start: datetime | None = None, + end: datetime | None = None, + limit: int | None = None, + offset: int | None = None, + types: list[EventType] | None = None, + smart_detect_types: list[SmartDetectObjectType] | None = None, sorting: Literal["asc", "desc"] = "asc", descriptions: bool = True, category: EventCategories | None = None, @@ -1132,7 +1132,7 @@ async def get_devices_raw(self, model_type: ModelType) -> list[dict[str, Any]]: async def get_devices( self, model_type: ModelType, - expected_type: Optional[type[ProtectModel]] = None, + expected_type: type[ProtectModel] | None = None, ) -> list[ProtectModel]: """Gets a device list given a model_type, converted into Python objects""" objs: list[ProtectModel] = [] @@ -1235,7 +1235,7 @@ async def get_device( self, model_type: ModelType, device_id: str, - expected_type: Optional[type[ProtectModelWithId]] = None, + expected_type: type[ProtectModelWithId] | None = None, ) -> ProtectModelWithId: """Gets a device give the device model_type and id, converted into Python object""" obj = create_from_unifi_dict( @@ -1347,10 +1347,10 @@ async def get_liveview(self, device_id: str) -> Liveview: async def get_camera_snapshot( self, camera_id: str, - width: Optional[int] = None, - height: Optional[int] = None, - dt: Optional[datetime] = None, - ) -> Optional[bytes]: + width: int | None = None, + height: int | None = None, + dt: datetime | None = None, + ) -> bytes | None: """ Gets snapshot for a camera. @@ -1381,10 +1381,10 @@ async def get_camera_snapshot( async def get_package_camera_snapshot( self, camera_id: str, - width: Optional[int] = None, - height: Optional[int] = None, - dt: Optional[datetime] = None, - ) -> Optional[bytes]: + width: int | None = None, + height: int | None = None, + dt: datetime | None = None, + ) -> bytes | None: """ Gets snapshot from the package camera. @@ -1417,8 +1417,8 @@ async def _stream_response( self, response: aiohttp.ClientResponse, chunk_size: int, - iterator_callback: Optional[IteratorCallback] = None, - progress_callback: Optional[ProgressCallback] = None, + iterator_callback: IteratorCallback | None = None, + progress_callback: ProgressCallback | None = None, ) -> None: total = response.content_length or 0 current = 0 @@ -1439,12 +1439,12 @@ async def get_camera_video( end: datetime, channel_index: int = 0, validate_channel_id: bool = True, - output_file: Optional[Path] = None, - iterator_callback: Optional[IteratorCallback] = None, - progress_callback: Optional[ProgressCallback] = None, + output_file: Path | None = None, + iterator_callback: IteratorCallback | None = None, + progress_callback: ProgressCallback | None = None, chunk_size: int = 65536, - fps: Optional[int] = None, - ) -> Optional[bytes]: + fps: int | None = None, + ) -> bytes | None: """ Exports MP4 video from a given camera at a specific time. @@ -1524,7 +1524,7 @@ async def _get_image_with_retry( path: str, retry_timeout: int = RETRY_TIMEOUT, **kwargs: Any, - ) -> Optional[bytes]: + ) -> bytes | None: """ Retries image request until it returns or timesout. Used for event images like thumbnails and heatmaps. @@ -1533,7 +1533,7 @@ async def _get_image_with_retry( """ now = time.monotonic() timeout = now + retry_timeout - data: Optional[bytes] = None + data: bytes | None = None while data is None and now < timeout: data = await self.api_request_raw(path, raise_exception=False, **kwargs) if data is None: @@ -1545,10 +1545,10 @@ async def _get_image_with_retry( async def get_event_thumbnail( self, thumbnail_id: str, - width: Optional[int] = None, - height: Optional[int] = None, + width: int | None = None, + height: int | None = None, retry_timeout: int = RETRY_TIMEOUT, - ) -> Optional[bytes]: + ) -> bytes | None: """ Gets given thumbanail from a given event. @@ -1576,12 +1576,12 @@ async def get_event_thumbnail( async def get_event_animated_thumbnail( self, thumbnail_id: str, - width: Optional[int] = None, - height: Optional[int] = None, + width: int | None = None, + height: int | None = None, *, speedup: int = 10, retry_timeout: int = RETRY_TIMEOUT, - ) -> Optional[bytes]: + ) -> bytes | None: """ Gets given animated thumbanil from a given event. @@ -1613,7 +1613,7 @@ async def get_event_heatmap( self, heatmap_id: str, retry_timeout: int = RETRY_TIMEOUT, - ) -> Optional[bytes]: + ) -> bytes | None: """ Gets given heatmap from a given event. diff --git a/src/uiprotect/cli/__init__.py b/src/uiprotect/cli/__init__.py index efe52d81..369caa27 100644 --- a/src/uiprotect/cli/__init__.py +++ b/src/uiprotect/cli/__init__.py @@ -5,7 +5,7 @@ import logging import sys from pathlib import Path -from typing import cast +from typing import Optional, cast import orjson import typer diff --git a/src/uiprotect/data/base.py b/src/uiprotect/data/base.py index 7ca4837c..2de8cd95 100644 --- a/src/uiprotect/data/base.py +++ b/src/uiprotect/data/base.py @@ -8,7 +8,7 @@ from datetime import datetime, timedelta from functools import cache from ipaddress import IPv4Address -from typing import TYPE_CHECKING, Any, ClassVar, Optional, TypeVar, Union +from typing import TYPE_CHECKING, Any, ClassVar, TypeVar from uuid import UUID from pydantic.v1 import BaseModel @@ -70,22 +70,22 @@ class ProtectBaseObject(BaseModel): * Provides `.unifi_dict` to convert object back into UFP JSON """ - _api: Optional[ProtectApiClient] = PrivateAttr(None) + _api: ProtectApiClient | None = PrivateAttr(None) - _protect_objs: ClassVar[Optional[dict[str, type[ProtectBaseObject]]]] = None - _protect_objs_set: ClassVar[Optional[SetStr]] = None - _protect_lists: ClassVar[Optional[dict[str, type[ProtectBaseObject]]]] = None - _protect_lists_set: ClassVar[Optional[SetStr]] = None - _protect_dicts: ClassVar[Optional[dict[str, type[ProtectBaseObject]]]] = None - _protect_dicts_set: ClassVar[Optional[SetStr]] = None - _to_unifi_remaps: ClassVar[Optional[DictStrAny]] = None + _protect_objs: ClassVar[dict[str, type[ProtectBaseObject]] | None] = None + _protect_objs_set: ClassVar[SetStr | None] = None + _protect_lists: ClassVar[dict[str, type[ProtectBaseObject]] | None] = None + _protect_lists_set: ClassVar[SetStr | None] = None + _protect_dicts: ClassVar[dict[str, type[ProtectBaseObject]] | None] = None + _protect_dicts_set: ClassVar[SetStr | None] = None + _to_unifi_remaps: ClassVar[DictStrAny | None] = None class Config: arbitrary_types_allowed = True validate_assignment = True copy_on_model_validation = "shallow" - def __init__(self, api: Optional[ProtectApiClient] = None, **data: Any) -> None: + def __init__(self, api: ProtectApiClient | None = None, **data: Any) -> None: """ Base class for creating Python objects from UFP JSON data. @@ -97,7 +97,7 @@ def __init__(self, api: Optional[ProtectApiClient] = None, **data: Any) -> None: @classmethod def from_unifi_dict( cls, - api: Optional[ProtectApiClient] = None, + api: ProtectApiClient | None = None, **data: Any, ) -> Self: """ @@ -124,7 +124,7 @@ def from_unifi_dict( return cls.construct(**data) @classmethod - def construct(cls, _fields_set: Optional[set[str]] = None, **values: Any) -> Self: + def construct(cls, _fields_set: set[str] | None = None, **values: Any) -> Self: api = values.pop("api", None) values_set = set(values) @@ -267,7 +267,7 @@ def _get_protect_dicts_set(cls) -> set[str]: return cls._protect_dicts_set @classmethod - def _get_api(cls, api: Optional[ProtectApiClient]) -> Optional[ProtectApiClient]: + def _get_api(cls, api: ProtectApiClient | None) -> ProtectApiClient | None: """Helper method to try to find and the current ProjtectAPIClient instance from given data""" if api is None and isinstance(cls, ProtectBaseObject) and hasattr(cls, "_api"): # type: ignore[unreachable] api = cls._api # type: ignore[unreachable] @@ -279,7 +279,7 @@ def _clean_protect_obj( cls, data: Any, klass: type[ProtectBaseObject], - api: Optional[ProtectApiClient], + api: ProtectApiClient | None, ) -> Any: if isinstance(data, dict): if api is not None: @@ -292,7 +292,7 @@ def _clean_protect_obj_list( cls, items: list[Any], klass: type[ProtectBaseObject], - api: Optional[ProtectApiClient], + api: ProtectApiClient | None, ) -> list[Any]: for index, item in enumerate(items): items[index] = cls._clean_protect_obj(item, klass, api) @@ -303,7 +303,7 @@ def _clean_protect_obj_dict( cls, items: dict[Any, Any], klass: type[ProtectBaseObject], - api: Optional[ProtectApiClient], + api: ProtectApiClient | None, ) -> dict[Any, Any]: for key, value in items.items(): items[key] = cls._clean_protect_obj(value, klass, api) @@ -380,7 +380,7 @@ def _unifi_dict_protect_obj( use_obj: bool, klass: type[ProtectBaseObject], ) -> Any: - value: Optional[Any] = data.get(key) + value: Any | None = data.get(key) if use_obj: value = getattr(self, key) @@ -398,7 +398,7 @@ def _unifi_dict_protect_obj_list( use_obj: bool, klass: type[ProtectBaseObject], ) -> Any: - value: Optional[Any] = data.get(key) + value: Any | None = data.get(key) if use_obj: value = getattr(self, key) @@ -421,7 +421,7 @@ def _unifi_dict_protect_obj_dict( key: str, use_obj: bool, ) -> Any: - value: Optional[Any] = data.get(key) + value: Any | None = data.get(key) if use_obj: value = getattr(self, key) @@ -438,8 +438,8 @@ def _unifi_dict_protect_obj_dict( def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: """ Can either convert current Python object into UFP JSON dict or take the output of a `.dict()` call and convert it. @@ -492,13 +492,13 @@ def unifi_dict( def _inject_api( self, data: dict[str, Any], - api: Optional[ProtectApiClient], + api: ProtectApiClient | None, ) -> dict[str, Any]: data["api"] = api data_set = set(data) for key in self._get_protect_objs_set().intersection(data_set): - unifi_obj: Optional[Any] = getattr(self, key) + unifi_obj: Any | None = getattr(self, key) if unifi_obj is not None and isinstance(unifi_obj, dict): unifi_obj["api"] = api @@ -522,7 +522,7 @@ def update_from_dict(self: ProtectObject, data: dict[str, Any]) -> ProtectObject """Updates current object from a cleaned UFP JSON dict""" data_set = set(data) for key in self._get_protect_objs_set().intersection(data_set): - unifi_obj: Optional[Any] = getattr(self, key) + unifi_obj: Any | None = getattr(self, key) if unifi_obj is not None and isinstance(unifi_obj, ProtectBaseObject): item = data.pop(key) if item is not None: @@ -577,7 +577,7 @@ class ProtectModel(ProtectBaseObject): automatically decoding a `modelKey` object into the correct UFP object and type """ - model: Optional[ModelType] + model: ModelType | None @classmethod @cache @@ -586,8 +586,8 @@ def _get_unifi_remaps(cls) -> dict[str, str]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -614,7 +614,7 @@ def __init__(self, **data: Any) -> None: self._update_event = update_event or asyncio.Event() @classmethod - def construct(cls, _fields_set: Optional[set[str]] = None, **values: Any) -> Self: + def construct(cls, _fields_set: set[str] | None = None, **values: Any) -> Self: update_lock = values.pop("update_lock", None) update_queue = values.pop("update_queue", None) update_event = values.pop("update_event", None) @@ -814,19 +814,19 @@ async def emit_message(self, updated: dict[str, Any]) -> None: class ProtectDeviceModel(ProtectModelWithId): - name: Optional[str] + name: str | None type: str mac: str - host: Optional[Union[IPv4Address, str]] - up_since: Optional[datetime] - uptime: Optional[timedelta] - last_seen: Optional[datetime] - hardware_revision: Optional[str] - firmware_version: Optional[str] + host: IPv4Address | str | None + up_since: datetime | None + uptime: timedelta | None + last_seen: datetime | None + hardware_revision: str | None + firmware_version: str | None is_updating: bool is_ssh_enabled: bool - _callback_ping: Optional[TimerHandle] = PrivateAttr(None) + _callback_ping: TimerHandle | None = PrivateAttr(None) @classmethod @cache @@ -880,38 +880,38 @@ def callback() -> None: class WiredConnectionState(ProtectBaseObject): - phy_rate: Optional[float] + phy_rate: float | None class WirelessConnectionState(ProtectBaseObject): - signal_quality: Optional[int] - signal_strength: Optional[int] + signal_quality: int | None + signal_strength: int | None class BluetoothConnectionState(WirelessConnectionState): - experience_score: Optional[PercentFloat] = None + experience_score: PercentFloat | None = None class WifiConnectionState(WirelessConnectionState): - phy_rate: Optional[float] - channel: Optional[int] - frequency: Optional[int] - ssid: Optional[str] - bssid: Optional[str] = None - tx_rate: Optional[float] = None + phy_rate: float | None + channel: int | None + frequency: int | None + ssid: str | None + bssid: str | None = None + tx_rate: float | None = None # requires 2.7.5+ - ap_name: Optional[str] = None - experience: Optional[str] = None + ap_name: str | None = None + experience: str | None = None # requires 2.7.15+ - connectivity: Optional[str] = None + connectivity: str | None = None class ProtectAdoptableDeviceModel(ProtectDeviceModel): state: StateType - connection_host: Union[IPv4Address, str, None] - connected_since: Optional[datetime] - latest_firmware_version: Optional[str] - firmware_build: Optional[str] + connection_host: IPv4Address | str | None + connected_since: datetime | None + latest_firmware_version: str | None + firmware_build: str | None is_adopting: bool is_adopted: bool is_adopted_by_other: bool @@ -921,23 +921,23 @@ class ProtectAdoptableDeviceModel(ProtectDeviceModel): is_attempting_to_connect: bool is_connected: bool # requires 1.21+ - market_name: Optional[str] + market_name: str | None # requires 2.7.5+ - fw_update_state: Optional[str] = None + fw_update_state: str | None = None # requires 2.8.14+ - nvr_mac: Optional[str] = None + nvr_mac: str | None = None # requires 2.8.22+ - guid: Optional[UUID] = None + guid: UUID | None = None # requires 2.9.20+ - is_restoring: Optional[bool] = None - last_disconnect: Optional[datetime] = None - anonymous_device_id: Optional[UUID] = None + is_restoring: bool | None = None + last_disconnect: datetime | None = None + anonymous_device_id: UUID | None = None - wired_connection_state: Optional[WiredConnectionState] = None - wifi_connection_state: Optional[WifiConnectionState] = None - bluetooth_connection_state: Optional[BluetoothConnectionState] = None - bridge_id: Optional[str] - is_downloading_firmware: Optional[bool] + wired_connection_state: WiredConnectionState | None = None + wifi_connection_state: WifiConnectionState | None = None + bluetooth_connection_state: BluetoothConnectionState | None = None + bridge_id: str | None + is_downloading_firmware: bool | None # TODO: # bridgeCandidates @@ -977,8 +977,8 @@ async def _api_update(self, data: dict[str, Any]) -> None: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -1017,7 +1017,7 @@ def is_bluetooth(self) -> bool: return self.bluetooth_connection_state is not None @property - def bridge(self) -> Optional[Bridge]: + def bridge(self) -> Bridge | None: if self.bridge_id is None: return None @@ -1070,7 +1070,7 @@ async def unadopt(self) -> None: raise NotAuthorized("Do not have permission to unadopt devices") await self.api.unadopt_device(self.model, self.id) - async def adopt(self, name: Optional[str] = None) -> None: + async def adopt(self, name: str | None = None) -> None: """Adopts a device""" if not self.can_adopt: raise BadRequest("Device cannot be adopted") @@ -1085,11 +1085,11 @@ async def adopt(self, name: Optional[str] = None) -> None: class ProtectMotionDeviceModel(ProtectAdoptableDeviceModel): - last_motion: Optional[datetime] + last_motion: datetime | None is_dark: bool # not directly from UniFi - last_motion_event_id: Optional[str] = None + last_motion_event_id: str | None = None @classmethod @cache @@ -1098,8 +1098,8 @@ def _get_read_only_fields(cls) -> set[str]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -1109,7 +1109,7 @@ def unifi_dict( return data @property - def last_motion_event(self) -> Optional[Event]: + def last_motion_event(self) -> Event | None: if self.last_motion_event_id is None: return None diff --git a/src/uiprotect/data/bootstrap.py b/src/uiprotect/data/bootstrap.py index 8644002b..4db31c43 100644 --- a/src/uiprotect/data/bootstrap.py +++ b/src/uiprotect/data/bootstrap.py @@ -7,7 +7,7 @@ from copy import deepcopy from dataclasses import dataclass from datetime import datetime -from typing import Any, Optional, cast +from typing import Any, cast from uuid import UUID from aiohttp.client_exceptions import ServerDisconnectedError @@ -187,10 +187,10 @@ class Bootstrap(ProtectBaseObject): mac_lookup: dict[str, ProtectDeviceRef] = {} id_lookup: dict[str, ProtectDeviceRef] = {} _ws_stats: list[WSStat] = PrivateAttr([]) - _has_doorbell: Optional[bool] = PrivateAttr(None) - _has_smart: Optional[bool] = PrivateAttr(None) - _has_media: Optional[bool] = PrivateAttr(None) - _recording_start: Optional[datetime] = PrivateAttr(None) + _has_doorbell: bool | None = PrivateAttr(None) + _has_smart: bool | None = PrivateAttr(None) + _has_media: bool | None = PrivateAttr(None) + _recording_start: datetime | None = PrivateAttr(None) _refresh_tasks: set[asyncio.Task[None]] = PrivateAttr(set()) @classmethod @@ -221,8 +221,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -346,7 +346,7 @@ def _create_stat( def _get_frame_data( self, packet: WSPacket, - ) -> tuple[dict[str, Any], Optional[dict[str, Any]]]: + ) -> tuple[dict[str, Any], dict[str, Any] | None]: if self.capture_ws_stats: return deepcopy(packet.action_frame.data), deepcopy(packet.data_frame.data) return packet.action_frame.data, packet.data_frame.data @@ -355,7 +355,7 @@ def _process_add_packet( self, packet: WSPacket, data: dict[str, Any], - ) -> Optional[WSSubscriptionMessage]: + ) -> WSSubscriptionMessage | None: obj = create_from_unifi_dict(data, api=self._api) if isinstance(obj, Event): @@ -391,8 +391,8 @@ def _process_add_packet( def _process_remove_packet( self, packet: WSPacket, - data: Optional[dict[str, Any]], - ) -> Optional[WSSubscriptionMessage]: + data: dict[str, Any] | None, + ) -> WSSubscriptionMessage | None: model = packet.action_frame.data.get("modelKey") device_id = packet.action_frame.data.get("id") devices = getattr(self, f"{model}s", None) @@ -419,7 +419,7 @@ def _process_nvr_update( packet: WSPacket, data: dict[str, Any], ignore_stats: bool, - ) -> Optional[WSSubscriptionMessage]: + ) -> WSSubscriptionMessage | None: if ignore_stats: _remove_stats_keys(data) # nothing left to process @@ -457,7 +457,7 @@ def _process_device_update( action: dict[str, Any], data: dict[str, Any], ignore_stats: bool, - ) -> Optional[WSSubscriptionMessage]: + ) -> WSSubscriptionMessage | None: model_type = action["modelKey"] if ignore_stats: _remove_stats_keys(data) @@ -522,9 +522,9 @@ def _process_device_update( def process_ws_packet( self, packet: WSPacket, - models: Optional[set[ModelType]] = None, + models: set[ModelType] | None = None, ignore_stats: bool = False, - ) -> Optional[WSSubscriptionMessage]: + ) -> WSSubscriptionMessage | None: if models is None: models = set() diff --git a/src/uiprotect/data/convert.py b/src/uiprotect/data/convert.py index 944962a0..701a5b58 100644 --- a/src/uiprotect/data/convert.py +++ b/src/uiprotect/data/convert.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any from uiprotect.data.devices import ( Bridge, @@ -61,8 +61,8 @@ def get_klass_from_dict(data: dict[str, Any]) -> type[ProtectModel]: def create_from_unifi_dict( data: dict[str, Any], - api: Optional[ProtectApiClient] = None, - klass: Optional[type[ProtectModel]] = None, + api: ProtectApiClient | None = None, + klass: type[ProtectModel] | None = None, ) -> ProtectModel: """ Helper method to read the `modelKey` from a UFP JSON dict and convert to currect Python class. diff --git a/src/uiprotect/data/devices.py b/src/uiprotect/data/devices.py index 60422a07..5635dc56 100644 --- a/src/uiprotect/data/devices.py +++ b/src/uiprotect/data/devices.py @@ -10,7 +10,7 @@ from functools import cache from ipaddress import IPv4Address from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast +from typing import TYPE_CHECKING, Any, Literal, cast from pydantic.v1.fields import PrivateAttr @@ -132,7 +132,7 @@ class Light(ProtectMotionDeviceModel): light_device_settings: LightDeviceSettings light_on_settings: LightOnSettings light_mode_settings: LightModeSettings - camera_id: Optional[str] + camera_id: str | None is_camera_paired: bool @classmethod @@ -150,14 +150,14 @@ def _get_read_only_fields(cls) -> set[str]: } @property - def camera(self) -> Optional[Camera]: + def camera(self) -> Camera | None: """Paired Camera will always be none if no camera is paired""" if self.camera_id is None: return None return self.api.bootstrap.cameras[self.camera_id] - async def set_paired_camera(self, camera: Optional[Camera]) -> None: + async def set_paired_camera(self, camera: Camera | None) -> None: """Sets the camera paired with the light""" async with self._update_lock: await asyncio.sleep( @@ -186,7 +186,7 @@ def callback() -> None: await self.queue_update(callback) - async def set_light(self, enabled: bool, led_level: Optional[int] = None) -> None: + async def set_light(self, enabled: bool, led_level: int | None = None) -> None: """Force turns on/off the light""" def callback() -> None: @@ -217,9 +217,9 @@ def callback() -> None: async def set_light_settings( self, mode: LightModeType, - enable_at: Optional[LightModeEnableType] = None, - duration: Optional[timedelta] = None, - sensitivity: Optional[int] = None, + enable_at: LightModeEnableType | None = None, + duration: timedelta | None = None, + sensitivity: int | None = None, ) -> None: """ Updates various Light settings. @@ -255,26 +255,26 @@ class CameraChannel(ProtectBaseObject): name: str # read only enabled: bool # read only is_rtsp_enabled: bool - rtsp_alias: Optional[str] # read only + rtsp_alias: str | None # read only width: int height: int fps: int bitrate: int min_bitrate: int # read only max_bitrate: int # read only - min_client_adaptive_bit_rate: Optional[int] # read only - min_motion_adaptive_bit_rate: Optional[int] # read only + min_client_adaptive_bit_rate: int | None # read only + min_motion_adaptive_bit_rate: int | None # read only fps_values: list[int] # read only idr_interval: int # 3.0.22+ - auto_bitrate: Optional[bool] = None - auto_fps: Optional[bool] = None + auto_bitrate: bool | None = None + auto_fps: bool | None = None - _rtsp_url: Optional[str] = PrivateAttr(None) - _rtsps_url: Optional[str] = PrivateAttr(None) + _rtsp_url: str | None = PrivateAttr(None) + _rtsps_url: str | None = PrivateAttr(None) @property - def rtsp_url(self) -> Optional[str]: + def rtsp_url(self) -> str | None: if not self.is_rtsp_enabled or self.rtsp_alias is None: return None @@ -284,7 +284,7 @@ def rtsp_url(self) -> Optional[str]: return self._rtsp_url @property - def rtsps_url(self) -> Optional[str]: + def rtsps_url(self) -> str | None: if not self.is_rtsp_enabled or self.rtsp_alias is None: return None @@ -322,24 +322,24 @@ class ISPSettings(ProtectBaseObject): d_zoom_center_y: int d_zoom_scale: int d_zoom_stream_id: int - focus_mode: Optional[FocusMode] = None + focus_mode: FocusMode | None = None focus_position: int - touch_focus_x: Optional[int] - touch_focus_y: Optional[int] + touch_focus_x: int | None + touch_focus_y: int | None zoom_position: PercentInt - mount_position: Optional[MountPosition] = None + mount_position: MountPosition | None = None # requires 2.8.14+ - is_color_night_vision_enabled: Optional[bool] = None + is_color_night_vision_enabled: bool | None = None # 3.0.22+ - hdr_mode: Optional[HDRMode] = None - icr_custom_value: Optional[ICRCustomValue] = None - icr_switch_mode: Optional[str] = None - spotlight_duration: Optional[int] = None + hdr_mode: HDRMode | None = None + icr_custom_value: ICRCustomValue | None = None + icr_switch_mode: str | None = None + spotlight_duration: int | None = None def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -383,15 +383,15 @@ class RecordingSettings(ProtectBaseObject): mode: RecordingMode geofencing: GeofencingSetting motion_algorithm: MotionAlgorithm - enable_motion_detection: Optional[bool] = None + enable_motion_detection: bool | None = None use_new_motion_algorithm: bool # requires 2.9.20+ - in_schedule_mode: Optional[str] = None - out_schedule_mode: Optional[str] = None + in_schedule_mode: str | None = None + out_schedule_mode: str | None = None # 2.11.13+ - retention_duration: Optional[datetime] = None - smart_detect_post_padding: Optional[timedelta] = None - smart_detect_pre_padding: Optional[timedelta] = None + retention_duration: datetime | None = None + smart_detect_post_padding: timedelta | None = None + smart_detect_pre_padding: timedelta | None = None @classmethod @cache @@ -432,8 +432,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -465,9 +465,9 @@ def unifi_dict( class SmartDetectSettings(ProtectBaseObject): object_types: list[SmartDetectObjectType] - audio_types: Optional[list[SmartDetectAudioType]] = None + audio_types: list[SmartDetectAudioType] | None = None # requires 2.8.22+ - auto_tracking_object_types: Optional[list[SmartDetectObjectType]] = None + auto_tracking_object_types: list[SmartDetectObjectType] | None = None @classmethod def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: @@ -486,7 +486,7 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: class LCDMessage(ProtectBaseObject): type: DoorbellMessageType text: str - reset_at: Optional[datetime] = None + reset_at: datetime | None = None @classmethod def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: @@ -502,7 +502,7 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: return super().unifi_dict_to_dict(data) @classmethod - def _fix_text(cls, text: str, text_type: Optional[str]) -> str: + def _fix_text(cls, text: str, text_type: str | None) -> str: if text_type is None: text_type = cls.type.value @@ -513,8 +513,8 @@ def _fix_text(cls, text: str, text_type: Optional[str]) -> str: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -536,8 +536,8 @@ class TalkbackSettings(ProtectBaseObject): type_in: str bind_addr: IPv4Address bind_port: int - filter_addr: Optional[str] # can be used to restrict sender address - filter_port: Optional[int] # can be used to restrict sender port + filter_addr: str | None # can be used to restrict sender address + filter_port: int | None # can be used to restrict sender port channels: int # 1 or 2 sampling_rate: int # 8000, 11025, 22050, 44100, 48000 bits_per_sample: int @@ -545,22 +545,22 @@ class TalkbackSettings(ProtectBaseObject): class WifiStats(ProtectBaseObject): - channel: Optional[int] - frequency: Optional[int] - link_speed_mbps: Optional[str] + channel: int | None + frequency: int | None + link_speed_mbps: str | None signal_quality: PercentInt signal_strength: int class VideoStats(ProtectBaseObject): - recording_start: Optional[datetime] - recording_end: Optional[datetime] - recording_start_lq: Optional[datetime] - recording_end_lq: Optional[datetime] - timelapse_start: Optional[datetime] - timelapse_end: Optional[datetime] - timelapse_start_lq: Optional[datetime] - timelapse_end_lq: Optional[datetime] + recording_start: datetime | None + recording_end: datetime | None + recording_start_lq: datetime | None + recording_end_lq: datetime | None + timelapse_start: datetime | None + timelapse_end: datetime | None + timelapse_start_lq: datetime | None + timelapse_end_lq: datetime | None @classmethod @cache @@ -596,11 +596,11 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: class StorageStats(ProtectBaseObject): - used: Optional[int] # bytes - rate: Optional[float] # bytes / millisecond + used: int | None # bytes + rate: float | None # bytes / millisecond @property - def rate_per_second(self) -> Optional[float]: + def rate_per_second(self) -> float | None: if self.rate is None: return None @@ -615,8 +615,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -631,7 +631,7 @@ class CameraStats(ProtectBaseObject): tx_bytes: int wifi: WifiStats video: VideoStats - storage: Optional[StorageStats] + storage: StorageStats | None wifi_quality: PercentInt wifi_strength: int @@ -644,8 +644,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -671,8 +671,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -707,17 +707,17 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: class PrivacyMaskCapability(ProtectBaseObject): - max_masks: Optional[int] + max_masks: int | None rectangle_only: bool class HotplugExtender(ProtectBaseObject): - has_flash: Optional[bool] = None - has_ir: Optional[bool] = None - has_radar: Optional[bool] = None - is_attached: Optional[bool] = None + has_flash: bool | None = None + has_ir: bool | None = None + has_radar: bool | None = None + is_attached: bool | None = None # 3.0.22+ - flash_range: Optional[Any] = None + flash_range: Any | None = None @classmethod @cache @@ -726,17 +726,17 @@ def _get_unifi_remaps(cls) -> dict[str, str]: class Hotplug(ProtectBaseObject): - audio: Optional[bool] = None - video: Optional[bool] = None - extender: Optional[HotplugExtender] = None + audio: bool | None = None + video: bool | None = None + extender: HotplugExtender | None = None # 2.8.35+ - standalone_adoption: Optional[bool] = None + standalone_adoption: bool | None = None class PTZRangeSingle(ProtectBaseObject): - max: Optional[float] - min: Optional[float] - step: Optional[float] + max: float | None + min: float | None + step: float | None class PTZRange(ProtectBaseObject): @@ -826,27 +826,27 @@ class CameraFeatureFlags(ProtectBaseObject): audio: list[str] = [] audio_codecs: list[AudioCodecs] = [] mount_positions: list[MountPosition] = [] - has_infrared: Optional[bool] = None - lens_type: Optional[LensType] = None - hotplug: Optional[Hotplug] = None - smart_detect_audio_types: Optional[list[SmartDetectAudioType]] = None + has_infrared: bool | None = None + lens_type: LensType | None = None + hotplug: Hotplug | None = None + smart_detect_audio_types: list[SmartDetectAudioType] | None = None # 2.7.18+ is_doorbell: bool # 2.8.22+ - lens_model: Optional[str] = None + lens_model: str | None = None # 2.9.20+ - has_color_lcd_screen: Optional[bool] = None - has_line_crossing: Optional[bool] = None - has_line_crossing_counting: Optional[bool] = None - has_liveview_tracking: Optional[bool] = None + has_color_lcd_screen: bool | None = None + has_line_crossing: bool | None = None + has_line_crossing_counting: bool | None = None + has_liveview_tracking: bool | None = None # 2.10.10+ - has_flash: Optional[bool] = None - is_ptz: Optional[bool] = None + has_flash: bool | None = None + is_ptz: bool | None = None # 2.11.13+ - audio_style: Optional[list[AudioStyle]] = None - has_vertical_flip: Optional[bool] = None + audio_style: list[AudioStyle] | None = None + has_vertical_flip: bool | None = None # 3.0.22+ - flash_range: Optional[Any] = None + flash_range: Any | None = None focus: PTZRange pan: PTZRange @@ -908,13 +908,13 @@ class Camera(ProtectMotionDeviceModel): is_recording: bool is_motion_detected: bool is_smart_detected: bool - phy_rate: Optional[float] + phy_rate: float | None hdr_mode: bool # Recording Quality -> High Frame video_mode: VideoMode is_probing_for_wifi: bool chime_duration: timedelta - last_ring: Optional[datetime] + last_ring: datetime | None is_live_heatmap_enabled: bool video_reconfiguration_in_progress: bool channels: list[CameraChannel] @@ -930,7 +930,7 @@ class Camera(ProtectMotionDeviceModel): smart_detect_zones: list[SmartMotionZone] stats: CameraStats feature_flags: CameraFeatureFlags - lcd_message: Optional[LCDMessage] + lcd_message: LCDMessage | None lenses: list[CameraLenses] platform: str has_speaker: bool @@ -938,29 +938,29 @@ class Camera(ProtectMotionDeviceModel): audio_bitrate: int can_manage: bool is_managed: bool - voltage: Optional[float] + voltage: float | None # requires 1.21+ - is_poor_network: Optional[bool] - is_wireless_uplink_enabled: Optional[bool] + is_poor_network: bool | None + is_wireless_uplink_enabled: bool | None # requires 2.6.13+ - homekit_settings: Optional[CameraHomekitSettings] = None + homekit_settings: CameraHomekitSettings | None = None # requires 2.6.17+ - ap_mgmt_ip: Optional[IPv4Address] = None + ap_mgmt_ip: IPv4Address | None = None # requires 2.7.5+ - is_waterproof_case_attached: Optional[bool] = None - last_disconnect: Optional[datetime] = None + is_waterproof_case_attached: bool | None = None + last_disconnect: datetime | None = None # requires 2.8.14+ - is_2k: Optional[bool] = None - is_4k: Optional[bool] = None - use_global: Optional[bool] = None + is_2k: bool | None = None + is_4k: bool | None = None + use_global: bool | None = None # requires 2.8.22+ - user_configured_ap: Optional[bool] = None + user_configured_ap: bool | None = None # requires 2.9.20+ - has_recordings: Optional[bool] = None + has_recordings: bool | None = None # requires 2.10.10+ - is_ptz: Optional[bool] = None + is_ptz: bool | None = None # requires 2.11.13+ - audio_settings: Optional[CameraAudioSettings] = None + audio_settings: CameraAudioSettings | None = None # TODO: used for adopting # apMac read only @@ -976,17 +976,17 @@ class Camera(ProtectMotionDeviceModel): # recordingSchedulesV2 # not directly from UniFi - last_ring_event_id: Optional[str] = None - last_smart_detect: Optional[datetime] = None - last_smart_audio_detect: Optional[datetime] = None - last_smart_detect_event_id: Optional[str] = None - last_smart_audio_detect_event_id: Optional[str] = None + last_ring_event_id: str | None = None + last_smart_detect: datetime | None = None + last_smart_audio_detect: datetime | None = None + last_smart_detect_event_id: str | None = None + last_smart_audio_detect_event_id: str | None = None last_smart_detects: dict[SmartDetectObjectType, datetime] = {} last_smart_audio_detects: dict[SmartDetectAudioType, datetime] = {} last_smart_detect_event_ids: dict[SmartDetectObjectType, str] = {} last_smart_audio_detect_event_ids: dict[SmartDetectAudioType, str] = {} - talkback_stream: Optional[TalkbackStream] = None - _last_ring_timeout: Optional[datetime] = PrivateAttr(None) + talkback_stream: TalkbackStream | None = None + _last_ring_timeout: datetime | None = PrivateAttr(None) @classmethod @cache @@ -1040,8 +1040,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: if data is not None: if "motion_zones" in data: @@ -1117,14 +1117,14 @@ def update_from_dict(self, data: dict[str, Any]) -> Camera: return super().update_from_dict(data) @property - def last_ring_event(self) -> Optional[Event]: + def last_ring_event(self) -> Event | None: if self.last_ring_event_id is None: return None return self.api.bootstrap.events.get(self.last_ring_event_id) @property - def last_smart_detect_event(self) -> Optional[Event]: + def last_smart_detect_event(self) -> Event | None: """Get the last smart detect event id.""" if self.last_smart_detect_event_id is None: return None @@ -1151,7 +1151,7 @@ def icr_lux_display(self) -> int | None: def get_last_smart_detect_event( self, smart_type: SmartDetectObjectType, - ) -> Optional[Event]: + ) -> Event | None: """Get the last smart detect event for given type.""" event_id = self.last_smart_detect_event_ids.get(smart_type) if event_id is None: @@ -1160,7 +1160,7 @@ def get_last_smart_detect_event( return self.api.bootstrap.events.get(event_id) @property - def last_smart_audio_detect_event(self) -> Optional[Event]: + def last_smart_audio_detect_event(self) -> Event | None: """Get the last smart audio detect event id.""" if self.last_smart_audio_detect_event_id is None: return None @@ -1170,7 +1170,7 @@ def last_smart_audio_detect_event(self) -> Optional[Event]: def get_last_smart_audio_detect_event( self, smart_type: SmartDetectAudioType, - ) -> Optional[Event]: + ) -> Event | None: """Get the last smart audio detect event for given type.""" event_id = self.last_smart_audio_detect_event_ids.get(smart_type) if event_id is None: @@ -1362,12 +1362,12 @@ def is_person_detection_on(self) -> bool: return self._is_smart_enabled(SmartDetectObjectType.PERSON) @property - def last_person_detect_event(self) -> Optional[Event]: + def last_person_detect_event(self) -> Event | None: """Get the last person smart detection event.""" return self.get_last_smart_detect_event(SmartDetectObjectType.PERSON) @property - def last_person_detect(self) -> Optional[datetime]: + def last_person_detect(self) -> datetime | None: """Get the last person smart detection event.""" return self.last_smart_detects.get(SmartDetectObjectType.PERSON) @@ -1405,12 +1405,12 @@ def is_vehicle_detection_on(self) -> bool: return self._is_smart_enabled(SmartDetectObjectType.VEHICLE) @property - def last_vehicle_detect_event(self) -> Optional[Event]: + def last_vehicle_detect_event(self) -> Event | None: """Get the last vehicle smart detection event.""" return self.get_last_smart_detect_event(SmartDetectObjectType.VEHICLE) @property - def last_vehicle_detect(self) -> Optional[datetime]: + def last_vehicle_detect(self) -> datetime | None: """Get the last vehicle smart detection event.""" return self.last_smart_detects.get(SmartDetectObjectType.VEHICLE) @@ -1444,12 +1444,12 @@ def is_license_plate_detection_on(self) -> bool: ) @property - def last_license_plate_detect_event(self) -> Optional[Event]: + def last_license_plate_detect_event(self) -> Event | None: """Get the last license plate smart detection event.""" return self.get_last_smart_detect_event(SmartDetectObjectType.LICENSE_PLATE) @property - def last_license_plate_detect(self) -> Optional[datetime]: + def last_license_plate_detect(self) -> datetime | None: """Get the last license plate smart detection event.""" return self.last_smart_detects.get(SmartDetectObjectType.LICENSE_PLATE) @@ -1481,12 +1481,12 @@ def is_package_detection_on(self) -> bool: return self._is_smart_enabled(SmartDetectObjectType.PACKAGE) @property - def last_package_detect_event(self) -> Optional[Event]: + def last_package_detect_event(self) -> Event | None: """Get the last package smart detection event.""" return self.get_last_smart_detect_event(SmartDetectObjectType.PACKAGE) @property - def last_package_detect(self) -> Optional[datetime]: + def last_package_detect(self) -> datetime | None: """Get the last package smart detection event.""" return self.last_smart_detects.get(SmartDetectObjectType.PACKAGE) @@ -1515,12 +1515,12 @@ def is_animal_detection_on(self) -> bool: return self._is_smart_enabled(SmartDetectObjectType.ANIMAL) @property - def last_animal_detect_event(self) -> Optional[Event]: + def last_animal_detect_event(self) -> Event | None: """Get the last animal smart detection event.""" return self.get_last_smart_detect_event(SmartDetectObjectType.ANIMAL) @property - def last_animal_detect(self) -> Optional[datetime]: + def last_animal_detect(self) -> datetime | None: """Get the last animal smart detection event.""" return self.last_smart_detects.get(SmartDetectObjectType.ANIMAL) @@ -1592,12 +1592,12 @@ def is_smoke_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.SMOKE) @property - def last_smoke_detect_event(self) -> Optional[Event]: + def last_smoke_detect_event(self) -> Event | None: """Get the last person smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.SMOKE) @property - def last_smoke_detect(self) -> Optional[datetime]: + def last_smoke_detect(self) -> datetime | None: """Get the last smoke smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.SMOKE) @@ -1626,12 +1626,12 @@ def is_co_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.CMONX) @property - def last_cmonx_detect_event(self) -> Optional[Event]: + def last_cmonx_detect_event(self) -> Event | None: """Get the last CO alarm smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.CMONX) @property - def last_cmonx_detect(self) -> Optional[datetime]: + def last_cmonx_detect(self) -> datetime | None: """Get the last CO alarm smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.CMONX) @@ -1660,12 +1660,12 @@ def is_siren_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.SIREN) @property - def last_siren_detect_event(self) -> Optional[Event]: + def last_siren_detect_event(self) -> Event | None: """Get the last Siren smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.SIREN) @property - def last_siren_detect(self) -> Optional[datetime]: + def last_siren_detect(self) -> datetime | None: """Get the last Siren smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.SIREN) @@ -1694,12 +1694,12 @@ def is_baby_cry_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.BABY_CRY) @property - def last_baby_cry_detect_event(self) -> Optional[Event]: + def last_baby_cry_detect_event(self) -> Event | None: """Get the last Baby Cry smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.BABY_CRY) @property - def last_baby_cry_detect(self) -> Optional[datetime]: + def last_baby_cry_detect(self) -> datetime | None: """Get the last Baby Cry smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.BABY_CRY) @@ -1728,12 +1728,12 @@ def is_speaking_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.SPEAK) @property - def last_speaking_detect_event(self) -> Optional[Event]: + def last_speaking_detect_event(self) -> Event | None: """Get the last Speaking smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.SPEAK) @property - def last_speaking_detect(self) -> Optional[datetime]: + def last_speaking_detect(self) -> datetime | None: """Get the last Speaking smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.SPEAK) @@ -1762,12 +1762,12 @@ def is_bark_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.BARK) @property - def last_bark_detect_event(self) -> Optional[Event]: + def last_bark_detect_event(self) -> Event | None: """Get the last Bark smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.BARK) @property - def last_bark_detect(self) -> Optional[datetime]: + def last_bark_detect(self) -> datetime | None: """Get the last Bark smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.BARK) @@ -1797,12 +1797,12 @@ def is_car_alarm_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.BURGLAR) @property - def last_car_alarm_detect_event(self) -> Optional[Event]: + def last_car_alarm_detect_event(self) -> Event | None: """Get the last Car Alarm smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.BURGLAR) @property - def last_car_alarm_detect(self) -> Optional[datetime]: + def last_car_alarm_detect(self) -> datetime | None: """Get the last Car Alarm smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.BURGLAR) @@ -1831,12 +1831,12 @@ def is_car_horn_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.CAR_HORN) @property - def last_car_horn_detect_event(self) -> Optional[Event]: + def last_car_horn_detect_event(self) -> Event | None: """Get the last Car Horn smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.CAR_HORN) @property - def last_car_horn_detect(self) -> Optional[datetime]: + def last_car_horn_detect(self) -> datetime | None: """Get the last Car Horn smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.CAR_HORN) @@ -1865,12 +1865,12 @@ def is_glass_break_detection_on(self) -> bool: return self._is_audio_enabled(SmartDetectObjectType.GLASS_BREAK) @property - def last_glass_break_detect_event(self) -> Optional[Event]: + def last_glass_break_detect_event(self) -> Event | None: """Get the last Glass Break smart detection event.""" return self.get_last_smart_audio_detect_event(SmartDetectAudioType.GLASS_BREAK) @property - def last_glass_break_detect(self) -> Optional[datetime]: + def last_glass_break_detect(self) -> datetime | None: """Get the last Glass Break smart detection event.""" return self.last_smart_audio_detects.get(SmartDetectAudioType.GLASS_BREAK) @@ -1905,25 +1905,25 @@ def is_digital_chime(self) -> bool: return self.chime_type is ChimeType.DIGITAL @property - def high_camera_channel(self) -> Optional[CameraChannel]: + def high_camera_channel(self) -> CameraChannel | None: if len(self.channels) >= 3: return self.channels[0] return None @property - def medium_camera_channel(self) -> Optional[CameraChannel]: + def medium_camera_channel(self) -> CameraChannel | None: if len(self.channels) >= 3: return self.channels[1] return None @property - def low_camera_channel(self) -> Optional[CameraChannel]: + def low_camera_channel(self) -> CameraChannel | None: if len(self.channels) >= 3: return self.channels[2] return None @property - def default_camera_channel(self) -> Optional[CameraChannel]: + def default_camera_channel(self) -> CameraChannel | None: for channel in [ self.high_camera_channel, self.medium_camera_channel, @@ -1934,7 +1934,7 @@ def default_camera_channel(self) -> Optional[CameraChannel]: return self.high_camera_channel @property - def package_camera_channel(self) -> Optional[CameraChannel]: + def package_camera_channel(self) -> CameraChannel | None: if self.feature_flags.has_package_camera and len(self.channels) == 4: return self.channels[3] return None @@ -1983,7 +1983,7 @@ def set_ring_timeout(self) -> None: self._last_ring_timeout = utc_now() + EVENT_PING_INTERVAL self._event_callback_ping() - def get_privacy_zone(self) -> tuple[Optional[int], Optional[CameraZone]]: + def get_privacy_zone(self) -> tuple[int | None, CameraZone | None]: for index, zone in enumerate(self.privacy_zones): if zone.name == PRIVACY_ZONE_NAME: return index, zone @@ -2006,10 +2006,10 @@ def remove_privacy_zone(self) -> None: async def get_snapshot( self, - width: Optional[int] = None, - height: Optional[int] = None, - dt: Optional[datetime] = None, - ) -> Optional[bytes]: + width: int | None = None, + height: int | None = None, + dt: datetime | None = None, + ) -> bytes | None: """ Gets snapshot for camera. @@ -2031,10 +2031,10 @@ async def get_snapshot( async def get_package_snapshot( self, - width: Optional[int] = None, - height: Optional[int] = None, - dt: Optional[datetime] = None, - ) -> Optional[bytes]: + width: int | None = None, + height: int | None = None, + dt: datetime | None = None, + ) -> bytes | None: """ Gets snapshot from the package camera. @@ -2062,12 +2062,12 @@ async def get_video( start: datetime, end: datetime, channel_index: int = 0, - output_file: Optional[Path] = None, - iterator_callback: Optional[IteratorCallback] = None, - progress_callback: Optional[ProgressCallback] = None, + output_file: Path | None = None, + iterator_callback: IteratorCallback | None = None, + progress_callback: ProgressCallback | None = None, chunk_size: int = 65536, - fps: Optional[int] = None, - ) -> Optional[bytes]: + fps: int | None = None, + ) -> bytes | None: """ Exports MP4 video from a given camera at a specific time. @@ -2407,9 +2407,9 @@ def callback() -> None: async def set_lcd_text( self, - text_type: Optional[DoorbellMessageType], - text: Optional[str] = None, - reset_at: Union[None, datetime, DEFAULT_TYPE] = None, + text_type: DoorbellMessageType | None, + text: str | None = None, + reset_at: None | datetime | DEFAULT_TYPE = None, ) -> None: """Sets doorbell LCD text. Requires camera to be doorbell""" if not self.feature_flags.has_lcd_screen: @@ -2450,8 +2450,8 @@ def callback() -> None: async def set_privacy( self, enabled: bool, - mic_level: Optional[int] = None, - recording_mode: Optional[RecordingMode] = None, + mic_level: int | None = None, + recording_mode: RecordingMode | None = None, reenable_global: bool = False, ) -> None: """Adds/removes a privacy zone that blacks out the whole camera.""" @@ -2494,7 +2494,7 @@ def callback() -> None: def create_talkback_stream( self, content_url: str, - ffmpeg_path: Optional[Path] = None, + ffmpeg_path: Path | None = None, ) -> TalkbackStream: """ Creates a subprocess to play audio to a camera through its speaker. @@ -2521,7 +2521,7 @@ def create_talkback_stream( async def play_audio( self, content_url: str, - ffmpeg_path: Optional[Path] = None, + ffmpeg_path: Path | None = None, blocking: bool = True, ) -> None: """ @@ -2726,7 +2726,7 @@ def _get_read_only_fields(cls) -> set[str]: return super()._get_read_only_fields() | {"softwareVersion"} @property - def liveview(self) -> Optional[Liveview]: + def liveview(self) -> Liveview | None: # user may not have permission to see the liveview return self.api.bootstrap.liveviews.get(self.liveview_id) @@ -2764,8 +2764,8 @@ class SensorThresholdSettings(SensorSettingsBase): margin: float # read only # "safe" thresholds for alerting # anything below/above will trigger alert - low_threshold: Optional[float] - high_threshold: Optional[float] + low_threshold: float | None + high_threshold: float | None class SensorSensitivitySettings(SensorSettingsBase): @@ -2773,12 +2773,12 @@ class SensorSensitivitySettings(SensorSettingsBase): class SensorBatteryStatus(ProtectBaseObject): - percentage: Optional[PercentInt] + percentage: PercentInt | None is_low: bool class SensorStat(ProtectBaseObject): - value: Optional[float] + value: float | None status: SensorStatusType @@ -2790,31 +2790,31 @@ class SensorStats(ProtectBaseObject): class Sensor(ProtectAdoptableDeviceModel): alarm_settings: SensorSettingsBase - alarm_triggered_at: Optional[datetime] + alarm_triggered_at: datetime | None battery_status: SensorBatteryStatus - camera_id: Optional[str] + camera_id: str | None humidity_settings: SensorThresholdSettings is_motion_detected: bool is_opened: bool - leak_detected_at: Optional[datetime] + leak_detected_at: datetime | None led_settings: SensorSettingsBase light_settings: SensorThresholdSettings - motion_detected_at: Optional[datetime] + motion_detected_at: datetime | None motion_settings: SensorSensitivitySettings - open_status_changed_at: Optional[datetime] + open_status_changed_at: datetime | None stats: SensorStats - tampering_detected_at: Optional[datetime] + tampering_detected_at: datetime | None temperature_settings: SensorThresholdSettings mount_type: MountType # not directly from UniFi - last_motion_event_id: Optional[str] = None - last_contact_event_id: Optional[str] = None - last_value_event_id: Optional[str] = None - last_alarm_event_id: Optional[str] = None - extreme_value_detected_at: Optional[datetime] = None - _tamper_timeout: Optional[datetime] = PrivateAttr(None) - _alarm_timeout: Optional[datetime] = PrivateAttr(None) + last_motion_event_id: str | None = None + last_contact_event_id: str | None = None + last_value_event_id: str | None = None + last_alarm_event_id: str | None = None + extreme_value_detected_at: datetime | None = None + _tamper_timeout: datetime | None = PrivateAttr(None) + _alarm_timeout: datetime | None = PrivateAttr(None) @classmethod @cache @@ -2838,8 +2838,8 @@ def _get_read_only_fields(cls) -> set[str]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -2857,7 +2857,7 @@ def unifi_dict( return data @property - def camera(self) -> Optional[Camera]: + def camera(self) -> Camera | None: """Paired Camera will always be none if no camera is paired""" if self.camera_id is None: return None @@ -2909,28 +2909,28 @@ def set_alarm_timeout(self) -> None: self._event_callback_ping() @property - def last_motion_event(self) -> Optional[Event]: + def last_motion_event(self) -> Event | None: if self.last_motion_event_id is None: return None return self.api.bootstrap.events.get(self.last_motion_event_id) @property - def last_contact_event(self) -> Optional[Event]: + def last_contact_event(self) -> Event | None: if self.last_contact_event_id is None: return None return self.api.bootstrap.events.get(self.last_contact_event_id) @property - def last_value_event(self) -> Optional[Event]: + def last_value_event(self) -> Event | None: if self.last_value_event_id is None: return None return self.api.bootstrap.events.get(self.last_value_event_id) @property - def last_alarm_event(self) -> Optional[Event]: + def last_alarm_event(self) -> Event | None: if self.last_alarm_event_id is None: return None @@ -3076,7 +3076,7 @@ def callback() -> None: await self.queue_update(callback) - async def set_paired_camera(self, camera: Optional[Camera]) -> None: + async def set_paired_camera(self, camera: Camera | None) -> None: """Sets the camera paired with the sensor""" def callback() -> None: @@ -3101,13 +3101,13 @@ async def clear_tamper(self) -> None: class Doorlock(ProtectAdoptableDeviceModel): - credentials: Optional[str] + credentials: str | None lock_status: LockStatusType enable_homekit: bool auto_close_time: timedelta led_settings: SensorSettingsBase battery_status: SensorBatteryStatus - camera_id: Optional[str] + camera_id: str | None has_homekit: bool private_token: str @@ -3140,14 +3140,14 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: return super().unifi_dict_to_dict(data) @property - def camera(self) -> Optional[Camera]: + def camera(self) -> Camera | None: """Paired Camera will always be none if no camera is paired""" if self.camera_id is None: return None return self.api.bootstrap.cameras[self.camera_id] - async def set_paired_camera(self, camera: Optional[Camera]) -> None: + async def set_paired_camera(self, camera: Camera | None) -> None: """Sets the camera paired with the sensor""" def callback() -> None: @@ -3202,7 +3202,7 @@ async def calibrate(self) -> None: class ChimeFeatureFlags(ProtectBaseObject): has_wifi: bool # 2.9.20+ - has_https_client_ota: Optional[bool] = None + has_https_client_ota: bool | None = None @classmethod @cache @@ -3222,7 +3222,7 @@ def _get_unifi_remaps(cls) -> dict[str, str]: return {**super()._get_unifi_remaps(), "camera": "cameraId"} @property - def camera(self) -> Optional[Camera]: + def camera(self) -> Camera | None: """Paired Camera will always be none if no camera is paired""" if self.camera_id is None: return None # type: ignore[unreachable] @@ -3247,20 +3247,20 @@ def _get_unifi_remaps(cls) -> dict[str, str]: class Chime(ProtectAdoptableDeviceModel): volume: PercentInt is_probing_for_wifi: bool - last_ring: Optional[datetime] + last_ring: datetime | None is_wireless_uplink_enabled: bool camera_ids: list[str] # requires 2.6.17+ - ap_mgmt_ip: Optional[IPv4Address] = None + ap_mgmt_ip: IPv4Address | None = None # requires 2.7.15+ - feature_flags: Optional[ChimeFeatureFlags] = None + feature_flags: ChimeFeatureFlags | None = None # requires 2.8.22+ - user_configured_ap: Optional[bool] = None + user_configured_ap: bool | None = None # requires 3.0.22+ - has_https_client_ota: Optional[bool] = None - platform: Optional[str] = None - repeat_times: Optional[RepeatTimes] = None - track_no: Optional[int] = None + has_https_client_ota: bool | None = None + platform: str | None = None + repeat_times: RepeatTimes | None = None + track_no: int | None = None ring_settings: list[RingSetting] = [] speaker_track_list: list[ChimeTrack] = [] diff --git a/src/uiprotect/data/nvr.py b/src/uiprotect/data/nvr.py index 0ffba647..9683e0ca 100644 --- a/src/uiprotect/data/nvr.py +++ b/src/uiprotect/data/nvr.py @@ -8,7 +8,7 @@ from functools import cache from ipaddress import IPv4Address, IPv6Address from pathlib import Path -from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional, Union +from typing import TYPE_CHECKING, Any, ClassVar, Literal from uuid import UUID import aiofiles @@ -72,7 +72,7 @@ class NVRLocation(UserLocation): is_geofencing_enabled: bool radius: int - model: Optional[ModelType] = None + model: ModelType | None = None class SmartDetectItem(ProtectBaseObject): @@ -120,7 +120,7 @@ def camera(self) -> Camera: return self.api.bootstrap.cameras[self.camera_id] @property - def event(self) -> Optional[Event]: + def event(self) -> Event | None: return self.api.bootstrap.events.get(self.event_id) @@ -135,13 +135,13 @@ class EventThumbnailAttribute(ProtectBaseObject): class EventThumbnailAttributes(ProtectBaseObject): - color: Optional[EventThumbnailAttribute] = None - vehicle_type: Optional[EventThumbnailAttribute] = None + color: EventThumbnailAttribute | None = None + vehicle_type: EventThumbnailAttribute | None = None def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -153,11 +153,11 @@ def unifi_dict( class EventDetectedThumbnail(ProtectBaseObject): - clock_best_wall: Optional[datetime] = None + clock_best_wall: datetime | None = None type: str cropped_id: str - attributes: Optional[EventThumbnailAttributes] = None - name: Optional[str] + attributes: EventThumbnailAttributes | None = None + name: str | None @classmethod def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: @@ -171,8 +171,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -183,28 +183,28 @@ def unifi_dict( class EventMetadata(ProtectBaseObject): - client_platform: Optional[str] - reason: Optional[str] - app_update: Optional[str] - light_id: Optional[str] - light_name: Optional[str] - type: Optional[str] - sensor_id: Optional[str] - sensor_name: Optional[str] - sensor_type: Optional[SensorType] - doorlock_id: Optional[str] - doorlock_name: Optional[str] - from_value: Optional[str] - to_value: Optional[str] - mount_type: Optional[MountType] - status: Optional[SensorStatusType] - alarm_type: Optional[str] - device_id: Optional[str] - mac: Optional[str] + client_platform: str | None + reason: str | None + app_update: str | None + light_id: str | None + light_name: str | None + type: str | None + sensor_id: str | None + sensor_name: str | None + sensor_type: SensorType | None + doorlock_id: str | None + doorlock_name: str | None + from_value: str | None + to_value: str | None + mount_type: MountType | None + status: SensorStatusType | None + alarm_type: str | None + device_id: str | None + mac: str | None # require 2.7.5+ - license_plate: Optional[LicensePlateMetadata] = None + license_plate: LicensePlateMetadata | None = None # requires 2.11.13+ - detected_thumbnails: Optional[list[EventDetectedThumbnail]] = None + detected_thumbnails: list[EventDetectedThumbnail] | None = None _collapse_keys: ClassVar[SetStr] = { "lightId", @@ -241,8 +241,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -262,30 +262,30 @@ def unifi_dict( class Event(ProtectModelWithId): type: EventType start: datetime - end: Optional[datetime] + end: datetime | None score: int - heatmap_id: Optional[str] - camera_id: Optional[str] + heatmap_id: str | None + camera_id: str | None smart_detect_types: list[SmartDetectObjectType] smart_detect_event_ids: list[str] - thumbnail_id: Optional[str] - user_id: Optional[str] - timestamp: Optional[datetime] - metadata: Optional[EventMetadata] + thumbnail_id: str | None + user_id: str | None + timestamp: datetime | None + metadata: EventMetadata | None # requires 2.7.5+ - deleted_at: Optional[datetime] = None - deletion_type: Optional[Literal["manual", "automatic"]] = None + deleted_at: datetime | None = None + deletion_type: Literal["manual", "automatic"] | None = None # only appears if `get_events` is called with category - category: Optional[EventCategories] = None - sub_category: Optional[str] = None + category: EventCategories | None = None + sub_category: str | None = None # TODO: # partition # description - _smart_detect_events: Optional[list[Event]] = PrivateAttr(None) - _smart_detect_track: Optional[SmartDetectTrack] = PrivateAttr(None) - _smart_detect_zones: Optional[dict[int, CameraZone]] = PrivateAttr(None) + _smart_detect_events: list[Event] | None = PrivateAttr(None) + _smart_detect_track: SmartDetectTrack | None = PrivateAttr(None) + _smart_detect_zones: dict[int, CameraZone] | None = PrivateAttr(None) @classmethod @cache @@ -308,8 +308,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -320,28 +320,28 @@ def unifi_dict( return data @property - def camera(self) -> Optional[Camera]: + def camera(self) -> Camera | None: if self.camera_id is None: return None return self.api.bootstrap.cameras.get(self.camera_id) @property - def light(self) -> Optional[Light]: + def light(self) -> Light | None: if self.metadata is None or self.metadata.light_id is None: return None return self.api.bootstrap.lights.get(self.metadata.light_id) @property - def sensor(self) -> Optional[Sensor]: + def sensor(self) -> Sensor | None: if self.metadata is None or self.metadata.sensor_id is None: return None return self.api.bootstrap.sensors.get(self.metadata.sensor_id) @property - def user(self) -> Optional[User]: + def user(self) -> User | None: if self.user_id is None: return None @@ -361,9 +361,9 @@ def smart_detect_events(self) -> list[Event]: async def get_thumbnail( self, - width: Optional[int] = None, - height: Optional[int] = None, - ) -> Optional[bytes]: + width: int | None = None, + height: int | None = None, + ) -> bytes | None: """Gets thumbnail for event""" if self.thumbnail_id is None: return None @@ -379,11 +379,11 @@ async def get_thumbnail( async def get_animated_thumbnail( self, - width: Optional[int] = None, - height: Optional[int] = None, + width: int | None = None, + height: int | None = None, *, speedup: int = 10, - ) -> Optional[bytes]: + ) -> bytes | None: """Gets animated thumbnail for event""" if self.thumbnail_id is None: return None @@ -402,7 +402,7 @@ async def get_animated_thumbnail( speedup=speedup, ) - async def get_heatmap(self) -> Optional[bytes]: + async def get_heatmap(self) -> bytes | None: """Gets heatmap for event""" if self.heatmap_id is None: return None @@ -419,11 +419,11 @@ async def get_heatmap(self) -> Optional[bytes]: async def get_video( self, channel_index: int = 0, - output_file: Optional[Path] = None, - iterator_callback: Optional[IteratorCallback] = None, - progress_callback: Optional[ProgressCallback] = None, + output_file: Path | None = None, + iterator_callback: IteratorCallback | None = None, + progress_callback: ProgressCallback | None = None, chunk_size: int = 65536, - ) -> Optional[bytes]: + ) -> bytes | None: """ Get the MP4 video clip for this given event @@ -512,11 +512,11 @@ class PortConfig(ProtectBaseObject): tcp_bridge: int ucore: int discovery_client: int - piongw: Optional[int] = None - ems_json_cli: Optional[int] = None - stacking: Optional[int] = None + piongw: int | None = None + ems_json_cli: int | None = None + stacking: int | None = None # 3.0.22+ - ai_feature_console: Optional[int] = None + ai_feature_console: int | None = None @classmethod @cache @@ -535,15 +535,15 @@ class CPUInfo(ProtectBaseObject): class MemoryInfo(ProtectBaseObject): - available: Optional[int] - free: Optional[int] - total: Optional[int] + available: int | None + free: int | None + total: int | None class StorageDevice(ProtectBaseObject): model: str size: int - healthy: Union[bool, str] + healthy: bool | str class StorageInfo(ProtectBaseObject): @@ -554,7 +554,7 @@ class StorageInfo(ProtectBaseObject): used: int devices: list[StorageDevice] # requires 2.8.14+ - capability: Optional[str] = None + capability: str | None = None @classmethod def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: @@ -586,25 +586,25 @@ class UOSDisk(ProtectBaseObject): slot: int state: str - type: Optional[Literal["SSD", "HDD"]] = None - model: Optional[str] = None - serial: Optional[str] = None - firmware: Optional[str] = None - rpm: Optional[int] = None - ata: Optional[str] = None - sata: Optional[str] = None - action: Optional[str] = None - healthy: Optional[str] = None - reason: Optional[list[Any]] = None - temperature: Optional[int] = None - power_on_hours: Optional[int] = None - life_span: Optional[PercentFloat] = None - bad_sector: Optional[int] = None - threshold: Optional[int] = None - progress: Optional[PercentFloat] = None - estimate: Optional[timedelta] = None + type: Literal["SSD", "HDD"] | None = None + model: str | None = None + serial: str | None = None + firmware: str | None = None + rpm: int | None = None + ata: str | None = None + sata: str | None = None + action: str | None = None + healthy: str | None = None + reason: list[Any] | None = None + temperature: int | None = None + power_on_hours: int | None = None + life_span: PercentFloat | None = None + bad_sector: int | None = None + threshold: int | None = None + progress: PercentFloat | None = None + estimate: timedelta | None = None # 2.10.10+ - size: Optional[int] = None + size: int | None = None @classmethod @cache @@ -625,8 +625,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -680,12 +680,12 @@ class UOSSpace(ProtectBaseObject): total_bytes: int used_bytes: int action: str - progress: Optional[PercentFloat] = None - estimate: Optional[timedelta] = None + progress: PercentFloat | None = None + estimate: timedelta | None = None # requires 2.8.14+ - health: Optional[str] = None + health: str | None = None # requires 2.8.22+ - space_type: Optional[str] = None + space_type: str | None = None # TODO: # reasons @@ -709,8 +709,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -734,12 +734,12 @@ class SystemInfo(ProtectBaseObject): memory: MemoryInfo storage: StorageInfo tmpfs: TMPFSInfo - ustorage: Optional[UOSStorage] = None + ustorage: UOSStorage | None = None def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -794,10 +794,10 @@ class StorageDistribution(ProtectBaseObject): recording_type_distributions: list[RecordingTypeDistribution] resolution_distributions: list[ResolutionDistribution] - _recording_type_dict: Optional[dict[RecordingType, RecordingTypeDistribution]] = ( + _recording_type_dict: dict[RecordingType, RecordingTypeDistribution] | None = ( PrivateAttr(None) ) - _resolution_dict: Optional[dict[ResolutionStorageType, ResolutionDistribution]] = ( + _resolution_dict: dict[ResolutionStorageType, ResolutionDistribution] | None = ( PrivateAttr(None) ) @@ -824,27 +824,27 @@ def _get_resolution_dict( return self._resolution_dict @property - def timelapse_recordings(self) -> Optional[RecordingTypeDistribution]: + def timelapse_recordings(self) -> RecordingTypeDistribution | None: return self._get_recording_type_dict().get(RecordingType.TIMELAPSE) @property - def continuous_recordings(self) -> Optional[RecordingTypeDistribution]: + def continuous_recordings(self) -> RecordingTypeDistribution | None: return self._get_recording_type_dict().get(RecordingType.CONTINUOUS) @property - def detections_recordings(self) -> Optional[RecordingTypeDistribution]: + def detections_recordings(self) -> RecordingTypeDistribution | None: return self._get_recording_type_dict().get(RecordingType.DETECTIONS) @property - def uhd_usage(self) -> Optional[ResolutionDistribution]: + def uhd_usage(self) -> ResolutionDistribution | None: return self._get_resolution_dict().get(ResolutionStorageType.UHD) @property - def hd_usage(self) -> Optional[ResolutionDistribution]: + def hd_usage(self) -> ResolutionDistribution | None: return self._get_resolution_dict().get(ResolutionStorageType.HD) @property - def free(self) -> Optional[ResolutionDistribution]: + def free(self) -> ResolutionDistribution | None: return self._get_resolution_dict().get(ResolutionStorageType.FREE) def update_from_dict(self, data: dict[str, Any]) -> StorageDistribution: @@ -857,8 +857,8 @@ def update_from_dict(self, data: dict[str, Any]) -> StorageDistribution: class StorageStats(ProtectBaseObject): utilization: float - capacity: Optional[timedelta] - remaining_capacity: Optional[timedelta] + capacity: timedelta | None + remaining_capacity: timedelta | None recording_space: StorageSpace storage_distribution: StorageDistribution @@ -878,11 +878,11 @@ class NVRFeatureFlags(ProtectBaseObject): beta: bool dev: bool notifications_v2: bool - homekit_paired: Optional[bool] = None - ulp_role_management: Optional[bool] = None + homekit_paired: bool | None = None + ulp_role_management: bool | None = None # 2.9.20+ - detection_labels: Optional[bool] = None - has_two_way_audio_media_streams: Optional[bool] = None + detection_labels: bool | None = None + has_two_way_audio_media_streams: bool | None = None class NVRSmartDetection(ProtectBaseObject): @@ -908,25 +908,25 @@ class NVR(ProtectDeviceModel): ucore_version: str hardware_platform: str ports: PortConfig - last_update_at: Optional[datetime] + last_update_at: datetime | None is_station: bool enable_automatic_backups: bool enable_stats_reporting: bool release_channel: FirmwareReleaseChannel - hosts: list[Union[IPv4Address, IPv6Address, str]] + hosts: list[IPv4Address | IPv6Address | str] enable_bridge_auto_adoption: bool hardware_id: UUID host_type: int host_shortname: str is_hardware: bool - is_wireless_uplink_enabled: Optional[bool] + is_wireless_uplink_enabled: bool | None time_format: Literal["12h", "24h"] temperature_unit: Literal["C", "F"] - recording_retention_duration: Optional[timedelta] + recording_retention_duration: timedelta | None enable_crash_reporting: bool disable_audio: bool analytics_data: AnalyticsOption - anonymous_device_id: Optional[UUID] + anonymous_device_id: UUID | None camera_utilization: int is_recycling: bool disable_auto_link: bool @@ -940,38 +940,38 @@ class NVR(ProtectDeviceModel): is_setup: bool network: str max_camera_capacity: dict[Literal["4K", "2K", "HD"], int] - market_name: Optional[str] = None - stream_sharing_available: Optional[bool] = None - is_db_available: Optional[bool] = None - is_insights_enabled: Optional[bool] = None - is_recording_disabled: Optional[bool] = None - is_recording_motion_only: Optional[bool] = None - ui_version: Optional[str] = None - sso_channel: Optional[FirmwareReleaseChannel] = None - is_stacked: Optional[bool] = None - is_primary: Optional[bool] = None - last_drive_slow_event: Optional[datetime] = None - is_u_core_setup: Optional[bool] = None + market_name: str | None = None + stream_sharing_available: bool | None = None + is_db_available: bool | None = None + is_insights_enabled: bool | None = None + is_recording_disabled: bool | None = None + is_recording_motion_only: bool | None = None + ui_version: str | None = None + sso_channel: FirmwareReleaseChannel | None = None + is_stacked: bool | None = None + is_primary: bool | None = None + last_drive_slow_event: datetime | None = None + is_u_core_setup: bool | None = None vault_camera_ids: list[str] = [] # requires 2.8.14+ - corruption_state: Optional[str] = None - country_code: Optional[str] = None - has_gateway: Optional[bool] = None - is_vault_registered: Optional[bool] = None - public_ip: Optional[IPv4Address] = None - ulp_version: Optional[str] = None - wan_ip: Optional[Union[IPv4Address, IPv6Address]] = None + corruption_state: str | None = None + country_code: str | None = None + has_gateway: bool | None = None + is_vault_registered: bool | None = None + public_ip: IPv4Address | None = None + ulp_version: str | None = None + wan_ip: IPv4Address | IPv6Address | None = None # requires 2.9.20+ - hard_drive_state: Optional[str] = None - is_network_installed: Optional[bool] = None - is_protect_updatable: Optional[bool] = None - is_ucore_updatable: Optional[bool] = None + hard_drive_state: str | None = None + is_network_installed: bool | None = None + is_protect_updatable: bool | None = None + is_ucore_updatable: bool | None = None # requires 2.11.13+ - last_device_fw_updates_checked_at: Optional[datetime] = None + last_device_fw_updates_checked_at: datetime | None = None # requires 3.0.22+ - smart_detection: Optional[NVRSmartDetection] = None - is_ucore_stacked: Optional[bool] = None - global_camera_settings: Optional[GlobalRecordingSettings] = None + smart_detection: NVRSmartDetection | None = None + is_ucore_stacked: bool | None = None + global_camera_settings: GlobalRecordingSettings | None = None # TODO: # errorCode read only @@ -1467,7 +1467,7 @@ class LiveviewSlot(ProtectBaseObject): cycle_mode: str cycle_interval: int - _cameras: Optional[list[Camera]] = PrivateAttr(None) + _cameras: list[Camera] | None = PrivateAttr(None) @classmethod @cache @@ -1507,7 +1507,7 @@ def _get_read_only_fields(cls) -> set[str]: return super()._get_read_only_fields() | {"isDefault", "owner"} @property - def owner(self) -> Optional[User]: + def owner(self) -> User | None: """ Owner of liveview. diff --git a/src/uiprotect/data/types.py b/src/uiprotect/data/types.py index 643bd627..11b46668 100644 --- a/src/uiprotect/data/types.py +++ b/src/uiprotect/data/types.py @@ -45,8 +45,8 @@ def __setitem__(self, key: KT, value: VT) -> None: class ValuesEnumMixin: - _values: Optional[list[str]] = None - _values_normalized: Optional[dict[str, str]] = None + _values: list[str] | None = None + _values_normalized: dict[str, str] | None = None @classmethod def values(cls) -> list[str]: @@ -55,7 +55,7 @@ def values(cls) -> list[str]: return cls._values @classmethod - def _missing_(cls, value: Any) -> Optional[Any]: + def _missing_(cls, value: Any) -> Any | None: if cls._values_normalized is None: cls._values_normalized = {e.value.lower(): e for e in cls} # type: ignore[attr-defined] @@ -67,7 +67,7 @@ def _missing_(cls, value: Any) -> Optional[Any]: class UnknownValuesEnumMixin(ValuesEnumMixin): @classmethod - def _missing_(cls, value: Any) -> Optional[Any]: + def _missing_(cls, value: Any) -> Any | None: # value always set in superclass _missing return super()._missing_(value) or cls._values_normalized.get("unknown") # type: ignore[union-attr] @@ -245,7 +245,7 @@ class SmartDetectObjectType(str, ValuesEnumMixin, enum.Enum): PET = "pet" @property - def audio_type(self) -> Optional[SmartDetectAudioType]: + def audio_type(self) -> SmartDetectAudioType | None: return OBJECT_TO_AUDIO_MAP.get(self) diff --git a/src/uiprotect/data/user.py b/src/uiprotect/data/user.py index 82edc066..b01937f8 100644 --- a/src/uiprotect/data/user.py +++ b/src/uiprotect/data/user.py @@ -4,7 +4,7 @@ from datetime import datetime from functools import cache -from typing import Any, Optional +from typing import Any from pydantic.v1.fields import PrivateAttr @@ -16,7 +16,7 @@ class Permission(ProtectBaseObject): raw_permission: str model: ModelType nodes: set[PermissionNode] - obj_ids: Optional[set[str]] + obj_ids: set[str] | None @classmethod def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: @@ -41,13 +41,13 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: def unifi_dict( # type: ignore[override] self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> str: return self.raw_permission @property - def objs(self) -> Optional[list[ProtectModelWithId]]: + def objs(self) -> list[ProtectModelWithId] | None: if self.obj_ids == {"self"} or self.obj_ids is None: return None @@ -72,8 +72,8 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: class UserLocation(ProtectModel): is_away: bool - latitude: Optional[float] - longitude: Optional[float] + latitude: float | None + longitude: float | None class CloudAccount(ProtectModelWithId): @@ -82,8 +82,8 @@ class CloudAccount(ProtectModelWithId): email: str user_id: str name: str - location: Optional[UserLocation] - profile_img: Optional[str] = None + location: UserLocation | None + profile_img: str | None = None @classmethod @cache @@ -92,8 +92,8 @@ def _get_unifi_remaps(cls) -> dict[str, str]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -116,21 +116,21 @@ class UserFeatureFlags(ProtectBaseObject): class User(ProtectModelWithId): permissions: list[Permission] - last_login_ip: Optional[str] - last_login_time: Optional[datetime] + last_login_ip: str | None + last_login_time: datetime | None is_owner: bool enable_notifications: bool has_accepted_invite: bool all_permissions: list[Permission] - scopes: Optional[list[str]] = None - location: Optional[UserLocation] + scopes: list[str] | None = None + location: UserLocation | None name: str first_name: str last_name: str - email: Optional[str] + email: str | None local_username: str group_ids: list[str] - cloud_account: Optional[CloudAccount] + cloud_account: CloudAccount | None feature_flags: UserFeatureFlags # TODO: @@ -140,7 +140,7 @@ class User(ProtectModelWithId): # notifications # cloudProviders - _groups: Optional[list[Group]] = PrivateAttr(None) + _groups: list[Group] | None = PrivateAttr(None) _perm_cache: dict[str, bool] = PrivateAttr({}) def __init__(self, **data: Any) -> None: @@ -175,8 +175,8 @@ def _get_unifi_remaps(cls) -> dict[str, str]: def unifi_dict( self, - data: Optional[dict[str, Any]] = None, - exclude: Optional[set[str]] = None, + data: dict[str, Any] | None = None, + exclude: set[str] | None = None, ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) @@ -206,7 +206,7 @@ def can( self, model: ModelType, node: PermissionNode, - obj: Optional[ProtectModelWithId] = None, + obj: ProtectModelWithId | None = None, ) -> bool: """Checks if a user can do a specific action""" check_self = False diff --git a/src/uiprotect/data/websocket.py b/src/uiprotect/data/websocket.py index 17b0dd93..d9b5922c 100644 --- a/src/uiprotect/data/websocket.py +++ b/src/uiprotect/data/websocket.py @@ -7,7 +7,7 @@ import struct import zlib from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any from uuid import UUID import orjson @@ -42,14 +42,14 @@ class WSSubscriptionMessage: action: WSAction new_update_id: UUID changed_data: dict[str, Any] - new_obj: Optional[ProtectModelWithId] = None - old_obj: Optional[ProtectModelWithId] = None + new_obj: ProtectModelWithId | None = None + old_obj: ProtectModelWithId | None = None class BaseWSPacketFrame: data: Any position: int = 0 - header: Optional[WSPacketFrameHeader] = None + header: WSPacketFrameHeader | None = None payload_format: ProtectWSPayloadFormat = ProtectWSPayloadFormat.NodeBuffer is_deflated: bool = False length: int = 0 @@ -75,7 +75,7 @@ def klass_from_format(format_raw: int) -> type[BaseWSPacketFrame]: def from_binary( data: bytes, position: int = 0, - klass: Optional[type[WSRawPacketFrame]] = None, + klass: type[WSRawPacketFrame] | None = None, ) -> BaseWSPacketFrame: """ Decode a unifi updates websocket frame. @@ -176,10 +176,10 @@ def json(self) -> bytes: class WSPacket: _raw: bytes - _raw_encoded: Optional[str] = None + _raw_encoded: str | None = None - _action_frame: Optional[BaseWSPacketFrame] = None - _data_frame: Optional[BaseWSPacketFrame] = None + _action_frame: BaseWSPacketFrame | None = None + _data_frame: BaseWSPacketFrame | None = None def __init__(self, data: bytes): self._raw = data diff --git a/src/uiprotect/stream.py b/src/uiprotect/stream.py index a86129a9..695fc337 100644 --- a/src/uiprotect/stream.py +++ b/src/uiprotect/stream.py @@ -6,7 +6,7 @@ from asyncio.subprocess import PIPE, Process, create_subprocess_exec from pathlib import Path from shlex import split -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING from urllib.parse import urlparse from aioshutil import which @@ -20,14 +20,14 @@ class FfmpegCommand: - ffmpeg_path: Optional[Path] + ffmpeg_path: Path | None args: list[str] - process: Optional[Process] = None + process: Process | None = None stdout: list[str] = [] stderr: list[str] = [] - def __init__(self, cmd: str, ffmpeg_path: Optional[Path] = None) -> None: + def __init__(self, cmd: str, ffmpeg_path: Path | None = None) -> None: self.args = split(cmd) if "ffmpeg" in self.args[0] and ffmpeg_path is None: @@ -85,7 +85,7 @@ async def stop(self) -> None: self.process.kill() await self.process.wait() - async def _read_stream(self, stream: Optional[StreamReader], attr: str) -> None: + async def _read_stream(self, stream: StreamReader | None, attr: str) -> None: if stream is None: return @@ -120,7 +120,7 @@ def __init__( self, camera: Camera, content_url: str, - ffmpeg_path: Optional[Path] = None, + ffmpeg_path: Path | None = None, ): if not camera.feature_flags.has_speaker: raise BadRequest("Camera does not have a speaker for talkback") diff --git a/src/uiprotect/test_util/__init__.py b/src/uiprotect/test_util/__init__.py index 75f2c6ea..c9465113 100644 --- a/src/uiprotect/test_util/__init__.py +++ b/src/uiprotect/test_util/__init__.py @@ -30,7 +30,7 @@ def placeholder_image( output_path: Path, width: int, - height: Optional[int] = None, + height: int | None = None, ) -> None: if height is None: height = width @@ -51,9 +51,9 @@ class SampleDataGenerator: _record_ws_start_time: float = time.monotonic() _record_listen_for_events: bool = False _record_ws_messages: dict[str, dict[str, Any]] = {} - _log: Optional[LOG_CALLABLE] = None - _log_warning: Optional[LOG_CALLABLE] = None - _ws_progress: Optional[PROGRESS_CALLABLE] = None + _log: LOG_CALLABLE | None = None + _log_warning: LOG_CALLABLE | None = None + _ws_progress: PROGRESS_CALLABLE | None = None constants: dict[str, Any] = {} client: ProtectApiClient @@ -68,9 +68,9 @@ def __init__( output: Path, anonymize: bool, wait_time: int, - log: Optional[LOG_CALLABLE] = None, - log_warning: Optional[LOG_CALLABLE] = None, - ws_progress: Optional[PROGRESS_CALLABLE] = None, + log: LOG_CALLABLE | None = None, + log_warning: LOG_CALLABLE | None = None, + ws_progress: PROGRESS_CALLABLE | None = None, do_zip: bool = False, ) -> None: self.client = client @@ -179,7 +179,7 @@ async def write_json_file( self, name: str, data: list[Any], - anonymize: Optional[bool] = None, + anonymize: bool | None = None, ) -> list[Any]: ... @overload @@ -187,15 +187,15 @@ async def write_json_file( self, name: str, data: dict[str, Any], - anonymize: Optional[bool] = None, + anonymize: bool | None = None, ) -> dict[str, Any]: ... async def write_json_file( self, name: str, - data: Union[list[Any], dict[str, Any]], - anonymize: Optional[bool] = None, - ) -> Union[list[Any], dict[str, Any]]: + data: list[Any] | dict[str, Any], + anonymize: bool | None = None, + ) -> list[Any] | dict[str, Any]: if anonymize is None: anonymize = self.anonymize @@ -211,7 +211,7 @@ async def write_binary_file( self, name: str, ext: str, - raw: Optional[bytes], + raw: bytes | None, ) -> None: def write() -> None: if raw is None: @@ -224,19 +224,19 @@ def write() -> None: loop = asyncio.get_running_loop() await loop.run_in_executor(None, write) - async def write_image_file(self, name: str, raw: Optional[bytes]) -> None: + async def write_image_file(self, name: str, raw: bytes | None) -> None: await self.write_binary_file(name, "png", raw) async def generate_event_data( self, - ) -> tuple[Optional[dict[str, Any]], Optional[dict[str, Any]]]: + ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: data = await self.client.get_events_raw() self.constants["time"] = datetime.now(tz=timezone.utc).isoformat() self.constants["event_count"] = len(data) - motion_event: Optional[dict[str, Any]] = None - smart_detection: Optional[dict[str, Any]] = None + motion_event: dict[str, Any] | None = None + smart_detection: dict[str, Any] | None = None for event_dict in reversed(data): if ( motion_event is None @@ -267,8 +267,8 @@ async def generate_event_data( async def generate_device_data( self, - motion_event: Optional[dict[str, Any]], - smart_detection: Optional[dict[str, Any]], + motion_event: dict[str, Any] | None, + smart_detection: dict[str, Any] | None, ) -> None: await asyncio.gather( self.generate_camera_data(), @@ -285,7 +285,7 @@ async def generate_device_data( async def generate_camera_data(self) -> None: objs = await self.client.api_request_list("cameras") - device_id: Optional[str] = None + device_id: str | None = None camera_is_online = False for obj_dict in objs: device_id = obj_dict["id"] @@ -320,7 +320,7 @@ async def generate_camera_data(self) -> None: async def generate_motion_data( self, - motion_event: Optional[dict[str, Any]], + motion_event: dict[str, Any] | None, ) -> None: if motion_event is None: self.log("No motion event, skipping thumbnail and heatmap generation...") @@ -375,7 +375,7 @@ async def generate_motion_data( async def generate_smart_detection_data( self, - smart_detection: Optional[dict[str, Any]], + smart_detection: dict[str, Any] | None, ) -> None: if smart_detection is None: self.log("No smart detection event, skipping smart detection data...") @@ -392,7 +392,7 @@ async def generate_smart_detection_data( async def generate_light_data(self) -> None: objs = await self.client.api_request_list("lights") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -407,7 +407,7 @@ async def generate_light_data(self) -> None: async def generate_viewport_data(self) -> None: objs = await self.client.api_request_list("viewers") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -422,7 +422,7 @@ async def generate_viewport_data(self) -> None: async def generate_sensor_data(self) -> None: objs = await self.client.api_request_list("sensors") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -437,7 +437,7 @@ async def generate_sensor_data(self) -> None: async def generate_lock_data(self) -> None: objs = await self.client.api_request_list("doorlocks") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -452,7 +452,7 @@ async def generate_lock_data(self) -> None: async def generate_chime_data(self) -> None: objs = await self.client.api_request_list("chimes") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -467,7 +467,7 @@ async def generate_chime_data(self) -> None: async def generate_bridge_data(self) -> None: objs = await self.client.api_request_list("bridges") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] if is_online(obj_dict): @@ -482,7 +482,7 @@ async def generate_bridge_data(self) -> None: async def generate_liveview_data(self) -> None: objs = await self.client.api_request_list("liveviews") - device_id: Optional[str] = None + device_id: str | None = None for obj_dict in objs: device_id = obj_dict["id"] break diff --git a/src/uiprotect/test_util/anonymize.py b/src/uiprotect/test_util/anonymize.py index d2e7b2f5..d1c321ca 100644 --- a/src/uiprotect/test_util/anonymize.py +++ b/src/uiprotect/test_util/anonymize.py @@ -3,7 +3,7 @@ import secrets import string import uuid -from typing import Any, Optional +from typing import Any from urllib.parse import urlparse import typer @@ -13,7 +13,7 @@ object_id_mapping: dict[str, str] = {} -def anonymize_data(value: Any, name: Optional[str] = None) -> Any: +def anonymize_data(value: Any, name: str | None = None) -> Any: if isinstance(value, list): value = anonymize_list(value, name=name) elif isinstance(value, dict): @@ -72,7 +72,7 @@ def anonymize_user(user_dict: dict[str, Any]) -> dict[str, Any]: return user_dict -def anonymize_value(value: Any, name: Optional[str] = None) -> Any: +def anonymize_value(value: Any, name: str | None = None) -> Any: if isinstance(value, str): if name == "accessKey": value = f"{random_number(13)}:{random_hex(24)}:{random_hex(128)}" @@ -105,7 +105,7 @@ def anonymize_value(value: Any, name: Optional[str] = None) -> Any: return value -def anonymize_dict(obj: dict[str, Any], name: Optional[str] = None) -> dict[str, Any]: +def anonymize_dict(obj: dict[str, Any], name: str | None = None) -> dict[str, Any]: obj_type = None if "modelKey" in obj: if obj["modelKey"] in [m.value for m in ModelType]: @@ -142,7 +142,7 @@ def anonymize_dict(obj: dict[str, Any], name: Optional[str] = None) -> dict[str, return obj -def anonymize_list(items: list[Any], name: Optional[str] = None) -> list[Any]: +def anonymize_list(items: list[Any], name: str | None = None) -> list[Any]: for index, value in enumerate(items): handled = False diff --git a/src/uiprotect/utils.py b/src/uiprotect/utils.py index 05b4374c..9ed70d47 100644 --- a/src/uiprotect/utils.py +++ b/src/uiprotect/utils.py @@ -22,7 +22,7 @@ from inspect import isclass from ipaddress import IPv4Address, IPv6Address, ip_address from pathlib import Path -from typing import TYPE_CHECKING, Any, Optional, TypeVar, Union, overload +from typing import TYPE_CHECKING, Any, TypeVar, Union, overload from uuid import UUID import jwt @@ -128,7 +128,7 @@ def to_js_time(dt: datetime | int) -> int: ... def to_js_time(dt: None) -> None: ... -def to_js_time(dt: datetime | int | None) -> Optional[int]: +def to_js_time(dt: datetime | int | None) -> int | None: """Converts Python datetime to Javascript timestamp""" if dt is None: return None @@ -142,7 +142,7 @@ def to_js_time(dt: datetime | int | None) -> Optional[int]: return int(dt.astimezone(timezone.utc).timestamp() * 1000) -def to_ms(duration: Optional[timedelta]) -> Optional[int]: +def to_ms(duration: timedelta | None) -> int | None: """Converts python timedelta to Milliseconds""" if duration is None: return None @@ -154,7 +154,7 @@ def utc_now() -> datetime: return datetime.now(tz=timezone.utc) -def from_js_time(num: Union[float, str, datetime]) -> datetime: +def from_js_time(num: float | str | datetime) -> datetime: """Converts Javascript timestamp to Python datetime""" if isinstance(num, datetime): return num @@ -162,15 +162,15 @@ def from_js_time(num: Union[float, str, datetime]) -> datetime: return datetime.fromtimestamp(int(num) / 1000, tz=timezone.utc) -def process_datetime(data: dict[str, Any], key: str) -> Optional[datetime]: +def process_datetime(data: dict[str, Any], key: str) -> datetime | None: """Extracts datetime object from Protect dictionary""" return None if data.get(key) is None else from_js_time(data[key]) def format_datetime( - dt: Optional[datetime], - default: Optional[str] = None, -) -> Optional[str]: + dt: datetime | None, + default: str | None = None, +) -> str | None: """Formats a datetime object in a consisent format""" return default if dt is None else dt.strftime(DATETIME_FORMAT) @@ -271,7 +271,7 @@ def serialize_dict(data: dict[str, Any], levels: int = -1) -> dict[str, Any]: return data -def serialize_coord(coord: CoordType) -> Union[int, float]: +def serialize_coord(coord: CoordType) -> int | float: """Serializes UFP zone coordinate""" from uiprotect.data import Percent @@ -283,7 +283,7 @@ def serialize_coord(coord: CoordType) -> Union[int, float]: return coord -def serialize_point(point: tuple[CoordType, CoordType]) -> list[Union[int, float]]: +def serialize_point(point: tuple[CoordType, CoordType]) -> list[int | float]: """Serializes UFP zone coordinate point""" return [ serialize_coord(point[0]), @@ -338,7 +338,7 @@ def ip_from_host(host: str) -> IPv4Address | IPv6Address: return ip_address(socket.gethostbyname(host)) -def dict_diff(orig: Optional[dict[str, Any]], new: dict[str, Any]) -> dict[str, Any]: +def dict_diff(orig: dict[str, Any] | None, new: dict[str, Any]) -> dict[str, Any]: changed: dict[str, Any] = {} if orig is None: @@ -375,7 +375,7 @@ def ws_stat_summmary( return unfiltered, percent, keys, models, actions -async def write_json(output_path: Path, data: Union[list[Any], dict[str, Any]]) -> None: +async def write_json(output_path: Path, data: list[Any] | dict[str, Any]) -> None: def write() -> None: with open(output_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=4) @@ -387,7 +387,7 @@ def write() -> None: def print_ws_stat_summary( stats: list[WSStat], - output: Optional[Callable[[Any], Any]] = None, + output: Callable[[Any], Any] | None = None, ) -> None: # typer<0.4.1 is incompatible with click>=8.1.0 # allows only the CLI interface to break if both are installed @@ -426,10 +426,10 @@ def print_ws_stat_summary( async def profile_ws( protect: ProtectApiClient, duration: int, - output_path: Optional[Path] = None, - ws_progress: Optional[PROGRESS_CALLABLE] = None, + output_path: Path | None = None, + ws_progress: PROGRESS_CALLABLE | None = None, do_print: bool = True, - print_output: Optional[Callable[[Any], Any]] = None, + print_output: Callable[[Any], Any] | None = None, ) -> None: if protect.bootstrap.capture_ws_stats: raise NvrError("Profile already in progress") diff --git a/src/uiprotect/websocket.py b/src/uiprotect/websocket.py index 18e63d03..4c5e0a7f 100644 --- a/src/uiprotect/websocket.py +++ b/src/uiprotect/websocket.py @@ -36,10 +36,10 @@ class Websocket: _ws_subscriptions: list[Callable[[WSMessage], None]] _connect_lock: asyncio.Lock - _headers: Optional[dict[str, str]] = None - _websocket_loop_task: Optional[asyncio.Task[None]] = None - _timer_task: Optional[asyncio.Task[None]] = None - _ws_connection: Optional[ClientWebSocketResponse] = None + _headers: dict[str, str] | None = None + _websocket_loop_task: asyncio.Task[None] | None = None + _timer_task: asyncio.Task[None] | None = None + _ws_connection: ClientWebSocketResponse | None = None _last_connect: float = -1000 _recent_failures: int = 0