diff --git a/src/uiprotect/data/base.py b/src/uiprotect/data/base.py index 18d011ca..c176585d 100644 --- a/src/uiprotect/data/base.py +++ b/src/uiprotect/data/base.py @@ -6,7 +6,7 @@ import logging from collections.abc import Callable from datetime import datetime, timedelta -from functools import cache +from functools import cache, cached_property from ipaddress import IPv4Address from typing import TYPE_CHECKING, Any, ClassVar, TypeVar from uuid import UUID @@ -89,6 +89,7 @@ class Config: arbitrary_types_allowed = True validate_assignment = True copy_on_model_validation = "shallow" + keep_untouched = (cached_property,) def __init__(self, api: ProtectApiClient | None = None, **data: Any) -> None: """ diff --git a/src/uiprotect/data/bootstrap.py b/src/uiprotect/data/bootstrap.py index 7ad481c2..ef58e89a 100644 --- a/src/uiprotect/data/bootstrap.py +++ b/src/uiprotect/data/bootstrap.py @@ -8,6 +8,7 @@ from copy import deepcopy from dataclasses import dataclass from datetime import datetime +from functools import cache, cached_property from typing import TYPE_CHECKING, Any, cast from aiohttp.client_exceptions import ServerDisconnectedError @@ -180,10 +181,6 @@ class Bootstrap(ProtectBaseObject): mac_lookup: dict[str, ProtectDeviceRef] = {} id_lookup: dict[str, ProtectDeviceRef] = {} _ws_stats: list[WSStat] = PrivateAttr([]) - _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 @@ -191,8 +188,11 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: api: ProtectApiClient | None = data.get("api") or ( cls._api if isinstance(cls, ProtectBaseObject) else None ) - data["macLookup"] = {} - data["idLookup"] = {} + mac_lookup: dict[str, dict[str, str | ModelType]] = {} + id_lookup: dict[str, dict[str, str | ModelType]] = {} + data["idLookup"] = id_lookup + data["macLookup"] = mac_lookup + for model_type in ModelType.bootstrap_models_types_set: key = model_type.devices_key # type: ignore[attr-defined] items: dict[str, ProtectModel] = {} @@ -204,16 +204,22 @@ def unifi_dict_to_dict(cls, data: dict[str, Any]) -> dict[str, Any]: ): continue - ref = {"model": model_type, "id": item["id"]} - items[item["id"]] = item - data["idLookup"][item["id"]] = ref + id_: str = item["id"] + ref = {"model": model_type, "id": id_} + items[id_] = item + id_lookup[id_] = ref if "mac" in item: cleaned_mac = normalize_mac(item["mac"]) - data["macLookup"][cleaned_mac] = ref + mac_lookup[cleaned_mac] = ref data[key] = items return super().unifi_dict_to_dict(data) + @classmethod + @cache + def _unifi_dict_remove_keys(cls) -> set[str]: + return {"events", "captureWsStats", "macLookup", "idLookup"} + def unifi_dict( self, data: dict[str, Any] | None = None, @@ -221,15 +227,9 @@ def unifi_dict( ) -> dict[str, Any]: data = super().unifi_dict(data=data, exclude=exclude) - if "events" in data: - del data["events"] - if "captureWsStats" in data: - del data["captureWsStats"] - if "macLookup" in data: - del data["macLookup"] - if "idLookup" in data: - del data["idLookup"] - + for key in Bootstrap._unifi_dict_remove_keys(): + if key in data: + del data[key] for model_type in ModelType.bootstrap_models_types_set: attr = model_type.devices_key # type: ignore[attr-defined] if attr in data and isinstance(data[attr], dict): @@ -246,51 +246,35 @@ def clear_ws_stats(self) -> None: @property def auth_user(self) -> User: - user: User = self._api.bootstrap.users[self.auth_user_id] - return user + return self._api.bootstrap.users[self.auth_user_id] - @property + @cached_property def has_doorbell(self) -> bool: - if self._has_doorbell is None: - self._has_doorbell = any( - c.feature_flags.is_doorbell for c in self.cameras.values() - ) + return any(c.feature_flags.is_doorbell for c in self.cameras.values()) - return self._has_doorbell - - @property + @cached_property def recording_start(self) -> datetime | None: - """Get earilest recording date.""" - if self._recording_start is None: - try: - self._recording_start = min( - c.stats.video.recording_start - for c in self.cameras.values() - if c.stats.video.recording_start is not None - ) - except ValueError: - return None - return self._recording_start + """Get earliest recording date.""" + try: + return min( + c.stats.video.recording_start + for c in self.cameras.values() + if c.stats.video.recording_start is not None + ) + except ValueError: + return None - @property + @cached_property def has_smart_detections(self) -> bool: """Check if any camera has smart detections.""" - if self._has_smart is None: - self._has_smart = any( - c.feature_flags.has_smart_detect for c in self.cameras.values() - ) - return self._has_smart + return any(c.feature_flags.has_smart_detect for c in self.cameras.values()) - @property + @cached_property def has_media(self) -> bool: """Checks if user can read media for any camera.""" - if self._has_media is None: - if self.recording_start is None: - return False - self._has_media = any( - c.can_read_media(self.auth_user) for c in self.cameras.values() - ) - return self._has_media + if self.recording_start is None: + return False + return any(c.can_read_media(self.auth_user) for c in self.cameras.values()) def get_device_from_mac(self, mac: str) -> ProtectAdoptableDeviceModel | None: """Retrieve a device from MAC address.""" @@ -419,14 +403,13 @@ def _process_nvr_update( return None # for another NVR in stack - nvr_id = packet.action_frame.data.get("id") + nvr_id: str | None = packet.action_frame.data.get("id") if nvr_id and nvr_id != self.nvr.id: self._create_stat(packet, None, True) return None - data = self.nvr.unifi_dict_to_dict(data) # nothing left to process - if not data: + if not (data := self.nvr.unifi_dict_to_dict(data)): self._create_stat(packet, None, True) return None diff --git a/tests/test_api.py b/tests/test_api.py index 4fb9929c..abe33b90 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -341,6 +341,18 @@ async def test_bootstrap(protect_client: ProtectApiClient): await check_bootstrap(protect_client.bootstrap) +@pytest.mark.asyncio() +async def test_bootstrap_cached_property(protect_client: ProtectApiClient): + """Test cached property works with bootstrap.""" + bootstrap = protect_client.bootstrap + + assert bootstrap.has_doorbell is True + bootstrap.cameras = {} + assert bootstrap.has_doorbell is True + del bootstrap.__dict__["has_doorbell"] + assert bootstrap.has_doorbell is False + + @pytest.mark.asyncio() async def test_bootstrap_construct(protect_client_no_debug: ProtectApiClient): """Verifies lookup of all object via ID"""