feat: add keyring and ulp-user

src/uiprotect/api.py

@@ -27,6 +27,9 @@
 from platformdirs import user_cache_dir, user_config_dir
 from yarl import URL
 
+from uiprotect.data.convert import dict_from_unifi_list
+from uiprotect.data.user import Keyring, UlpUser
+
 from ._compat import cached_property
 from .data import (
     NVR,
@@ -90,6 +93,8 @@
 _LOGGER = logging.getLogger(__name__)
 _COOKIE_RE = re.compile(r"^set-cookie: ", re.IGNORECASE)
 
+NFC_FINGERPRINT_SUPPORT_VERSION = Version("5.1.57")
+
 # TODO: Urls to still support
 # Backups
 # * GET /backups - list backends
@@ -823,6 +828,17 @@ async def update(self) -> Bootstrap:
         """
         async with self._update_lock:
             bootstrap = await self.get_bootstrap()
+            if bootstrap.nvr.version >= NFC_FINGERPRINT_SUPPORT_VERSION:
+                bootstrap.keyrings = cast(
+                    dict[str, Keyring],
+                    dict_from_unifi_list(self, await self.api_request_list("keyrings")),
+                )
+                bootstrap.ulp_users = cast(
+                    dict[str, UlpUser],
+                    dict_from_unifi_list(
+                        self, await self.api_request_list("ulp-users")
+                    ),
+                )
             self.__dict__.pop("bootstrap", None)
             self._bootstrap = bootstrap
             return bootstrap

src/uiprotect/data/bootstrap.py

@@ -6,22 +6,22 @@
 import logging
 from dataclasses import dataclass
 from datetime import datetime
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from aiohttp.client_exceptions import ServerDisconnectedError
 from convertertools import pop_dict_set, pop_dict_tuple
 from pydantic.v1 import PrivateAttr, ValidationError
 
 from ..exceptions import ClientError
-from ..utils import normalize_mac, utc_now
+from ..utils import normalize_mac, to_snake_case, utc_now
 from .base import (
     RECENT_EVENT_MAX,
     ProtectBaseObject,
     ProtectDeviceModel,
     ProtectModel,
     ProtectModelWithId,
 )
-from .convert import create_from_unifi_dict
+from .convert import MODEL_TO_CLASS, create_from_unifi_dict
 from .devices import (
     Bridge,
     Camera,
@@ -34,7 +34,7 @@
 )
 from .nvr import NVR, Event, Liveview
 from .types import EventType, FixSizeOrderedDict, ModelType
-from .user import Group, User
+from .user import Group, Keyring, UlpUser, User
 from .websocket import (
     WSAction,
     WSPacket,
@@ -188,6 +188,8 @@
     # agreements
 
     # not directly from UniFi
+    keyrings: dict[str, Keyring] = {}
+    ulp_users: dict[str, UlpUser] = {}
     events: dict[str, Event] = FixSizeOrderedDict()
     capture_ws_stats: bool = False
     mac_lookup: dict[str, ProtectDeviceRef] = {}
@@ -384,6 +386,59 @@
             old_obj=device,
         )
 
+    def _process_ws_keyring_or_ulp_user_message(
+        self,
+        action: dict[str, Any],
+        data: dict[str, Any],
+        model_type: ModelType,
+    ) -> WSSubscriptionMessage | None:
+        action_id = action["id"]
+        dict_from_bootstrap: dict[str, ProtectModelWithId] = getattr(
+            self, to_snake_case(model_type.devices_key)
+        )
+        action_type = action["action"]
+        if action_type == "add":
+            add_obj = create_from_unifi_dict(data, api=self._api, model_type=model_type)
+            if TYPE_CHECKING:
+                model_class = MODEL_TO_CLASS.get(model_type)
+                assert model_class is not None and isinstance(add_obj, model_class)
+            add_obj = cast(ProtectModelWithId, add_obj)
+            dict_from_bootstrap[add_obj.id] = add_obj
+            return WSSubscriptionMessage(
+                action=WSAction.ADD,
+                new_update_id=self.last_update_id,
+                changed_data=add_obj.dict(),
+                new_obj=add_obj,
+            )
+        elif action_type == "remove":
+            removed_obj = dict_from_bootstrap.pop(action_id, None)
+            if removed_obj is None:
+                return None
+            return WSSubscriptionMessage(
+                action=WSAction.REMOVE,
+                new_update_id=self.last_update_id,
+                changed_data={},
+                old_obj=removed_obj,
+            )
+        elif action_type == "update":
+            updated_obj = dict_from_bootstrap.get(action_id)
+            if updated_obj is None:
+                return None
+
+            old_obj = updated_obj.copy()
+            updated_data = {to_snake_case(k): v for k, v in data.items()}
+            updated_obj.update_from_dict(updated_data)
+
+            return WSSubscriptionMessage(
+                action=WSAction.UPDATE,
+                new_update_id=self.last_update_id,
+                changed_data=updated_data,
+                new_obj=updated_obj,
+                old_obj=old_obj,
+            )
+        _LOGGER.debug("Unexpected ws action for %s: %s", model_type, action_type)
+        return None
+
     def _process_nvr_update(
         self,
         action: dict[str, Any],
@@ -540,13 +595,16 @@
             return None
 
         action_action: str = action["action"]
-        if action_action == "remove":
-            return self._process_remove_packet(model_type, action)
-
-        if not data and not is_ping_back:
-            return None
 
         try:
+            if model_type in {ModelType.KEYRING, ModelType.ULP_USER}:
+                return self._process_ws_keyring_or_ulp_user_message(
+                    action, data, model_type
+                )
+            if action_action == "remove":
+                return self._process_remove_packet(model_type, action)
+            if not data and not is_ping_back:
+                return None
             if action_action == "add":
                 return self._process_add_packet(model_type, data)
             if action_action == "update":

src/uiprotect/data/convert.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
+
+from uiprotect.data.base import ProtectModelWithId
 
 from ..exceptions import DataDecodeError
 from .devices import (
@@ -16,7 +18,7 @@
 )
 from .nvr import NVR, Event, Liveview
 from .types import ModelType
-from .user import CloudAccount, Group, User, UserLocation
+from .user import CloudAccount, Group, Keyring, UlpUser, User, UserLocation
 
 if TYPE_CHECKING:
     from ..api import ProtectApiClient
@@ -38,6 +40,8 @@
     ModelType.SENSOR: Sensor,
     ModelType.DOORLOCK: Doorlock,
     ModelType.CHIME: Chime,
+    ModelType.KEYRING: Keyring,
+    ModelType.ULP_USER: UlpUser,
 }
 
 
@@ -79,3 +83,13 @@ def create_from_unifi_dict(
         klass = get_klass_from_dict(data)
 
     return klass.from_unifi_dict(**data, api=api)
+
+
+def dict_from_unifi_list(
+    api: ProtectApiClient, unifi_list: list[dict[str, ProtectModelWithId]]
+) -> dict[str, ProtectModelWithId]:
+    return_dict: dict[str, ProtectModelWithId] = {}
+    for obj_dict in unifi_list:
+        obj = cast(ProtectModelWithId, create_from_unifi_dict(obj_dict, api))
+        return_dict[obj.id] = obj
+    return return_dict

src/uiprotect/data/types.py

@@ -105,6 +105,8 @@ class ModelType(str, UnknownValuesEnumMixin, enum.Enum):
     CHIME = "chime"
     DEVICE_GROUP = "deviceGroup"
     RECORDING_SCHEDULE = "recordingSchedule"
+    ULP_USER = "ulpUser"
+    KEYRING = "keyring"
     UNKNOWN = "unknown"
 
     bootstrap_model_types: tuple[ModelType, ...]

src/uiprotect/data/user.py

@@ -234,3 +234,21 @@ def can(
                 return True
         self._perm_cache[perm_str] = False
         return False
+
+
+class Keyring(ProtectModelWithId):
+    device_type: str
+    device_id: str
+    registry_type: str
+    registry_id: str
+    last_activity: datetime | None = None
+    ulp_user: str
+
+
+class UlpUser(ProtectModelWithId):
+    ulp_id: str
+    first_name: str
+    last_name: str
+    full_name: str
+    avatar: str
+    status: str

tests/data/test_common.py

@@ -311,6 +311,10 @@ def test_bootstrap(bootstrap: dict[str, Any]):
     if "deviceGroups" in bootstrap:
         del bootstrap["deviceGroups"]
 
+    # Remove additional keys from obj_dict
+    obj_dict.pop("keyrings", None)
+    obj_dict.pop("ulpUsers", None)
+
     for model_type in ModelType.bootstrap_models:
         key = model_type + "s"
         expected_data = bootstrap.pop(key)

tests/sample_data/sample_keyrings.json

@@ -0,0 +1,22 @@
+[
+    {
+        "deviceType": "camera",
+        "deviceId": "1c9a2db4df6efda47a3509be",
+        "registryType": "nfc",
+        "registryId": "64B2A621",
+        "lastActivity": 1732904108638,
+        "ulpUser": "73791632-9805-419c-8351-f3afaab8f064",
+        "id": "672b573764f79603e400031d",
+        "modelKey": "keyring"
+    },
+    {
+        "deviceType": "camera",
+        "deviceId": "1c9a2db4df6efda47a3509be",
+        "registryType": "fingerprint",
+        "registryId": "1",
+        "lastActivity": 1732904119477,
+        "ulpUser": "0ef32f28-f654-404d-ab34-30e373e66436",
+        "id": "672b573764f79603e44871d",
+        "modelKey": "keyring"
+    }
+]

tests/sample_data/sample_ulp_users.json

@@ -0,0 +1,32 @@
+[
+    {
+        "ulpId": "73791632-9805-419c-8351-f3afaab8f064",
+        "firstName": "John Doe",
+        "lastName": "",
+        "fullName": "John Doe",
+        "avatar": "",
+        "status": "ACTIVE",
+        "id": "73791632-9805-419c-8351-f3afaab8f064",
+        "modelKey": "ulpUser"
+    },
+    {
+        "ulpId": "ddec43ea-1845-4a50-bdab-83bcd4b3c81d",
+        "firstName": "Jane Doe",
+        "lastName": "",
+        "fullName": "Jane Doe",
+        "avatar": "",
+        "status": "ACTIVE",
+        "id": "ddec43ea-1845-4a50-bdab-83bcd4b3c81d",
+        "modelKey": "ulpUser"
+    },
+    {
+        "ulpId": "0ef32f28-f654-404d-ab34-30e373e66436",
+        "firstName": "You Know Who",
+        "lastName": "",
+        "fullName": "You Know Who",
+        "avatar": "/proxy/users/public/avatar/1732954155_589f14b3-b137-4487-9823-db62bb793c19.jpg",
+        "status": "DEACTIVATED",
+        "id": "0ef32f28-f654-404d-ab34-30e373e66436",
+        "modelKey": "ulpUser"
+    }
+]