Merge pull request #202 from natekspencer/dev

Add user data to system

Author: natekspencer

vivintpy/const.py

@@ -32,14 +32,21 @@ class AuthUserAttribute:
 class UserAttribute:
     """User attributes."""
 
+    ADMIN = "ad"
     DOCUMENT_SEQUENCE = "DocumentSequence"
     EMAIL = "e"
     GHOME = "ghome"
     GROUP_IDS = "grpid"
     ID = "_id"
+    HAS_LOCK_PIN = "hasLockPin"
+    HAS_PANEL_PIN = "hasPanelPin"
+    HAS_PINS = "hasPins"
+    LOCK_IDS = "lids"
     MESSAGE_BROADCAST_CHANNEL = "mbc"
     NAME = "n"
     PING_ID = "pngid"
+    REGISTERED = "reg"
+    REMOTE_ACCESS = "ra"
     RESTRICTED_SYSTEM = "rsystem"
     SMART_HOME_SYSTEM = "smarthomesystem"
     SETTINGS = "stg"
@@ -72,6 +79,7 @@ class SystemAttribute:
     PARTITION_ID = "parid"
     SYSTEM = "system"
     SYSTEM_NICKNAME = "sn"
+    USERS = "u"
 
 
 class PubNubMessageAttribute:
@@ -149,6 +157,12 @@ class CameraAttribute(VivintDeviceAttribute):
     WIRELESS_SIGNAL_STRENGTH = "wiss"
 
 
+class LockAttribute(VivintDeviceAttribute):
+    """Lock attributes."""
+
+    USER_CODE_LIST = "ucl"
+
+
 class SwitchAttribute(VivintDeviceAttribute):
     """Switch attributes."""
 

vivintpy/devices/door_lock.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+from typing import cast
+
+from ..const import LockAttribute
 from ..const import ZWaveDeviceAttribute as Attribute
 from ..utils import send_deprecation_warning
 from . import BypassTamperDevice
@@ -26,6 +29,11 @@ def node_online(self) -> bool:
         send_deprecation_warning("node_online", "is_online")
         return self.is_online
 
+    @property
+    def user_code_list(self) -> list[int]:
+        """Return the user code list."""
+        return cast(list[int], self.data.get(LockAttribute.USER_CODE_LIST, []))
+
     async def set_state(self, locked: bool) -> None:
         """Set door lock's state."""
         assert self.alarm_panel

vivintpy/system.py

@@ -8,6 +8,7 @@
 from .const import SystemAttribute as Attribute
 from .devices.alarm_panel import AlarmPanel
 from .entity import Entity
+from .user import User
 from .utils import first_or_none, send_deprecation_warning
 from .vivintskyapi import VivintSkyApi
 
@@ -27,6 +28,10 @@ def __init__(self, data: dict, api: VivintSkyApi, *, name: str, is_admin: bool):
             AlarmPanel(panel_data, self)
             for panel_data in self.data[Attribute.SYSTEM][Attribute.PARTITION]
         ]
+        self.users = [
+            User(user_data, self)
+            for user_data in self.data[Attribute.SYSTEM][Attribute.USERS]
+        ]
 
     @property
     def api(self) -> VivintSkyApi:
@@ -70,17 +75,29 @@ async def refresh(self) -> None:
             else:
                 self.alarm_panels.append(AlarmPanel(panel_data, self))
 
+    def update_user_data(self, data: list[dict]) -> None:
+        """Update user data."""
+        for d in data:
+            user = first_or_none(self.users, lambda user: user.id == d["_id"])
+            if not user:
+                _LOGGER.debug("User not found for system %s: %s", self.id, d)
+                return
+            user.handle_pubnub_message(d)
+
     def handle_pubnub_message(self, message: dict) -> None:
         """Handle a pubnub message."""
-        if message[PubNubMessageAttribute.TYPE] == "account_system":
+        if (message_type := message[PubNubMessageAttribute.TYPE]) == "account_system":
             # this is a system message
             operation = message.get(PubNubMessageAttribute.OPERATION)
             data = message.get(PubNubMessageAttribute.DATA)
 
             if data and operation == "u":
+                if Attribute.USERS in data:
+                    self.update_user_data(data[Attribute.USERS])
+                    del data[Attribute.USERS]
                 self.update_data(data)
 
-        elif message[PubNubMessageAttribute.TYPE] == "account_partition":
+        elif message_type == "account_partition":
             # this is a message for one of the devices attached to this system
             partition_id = message.get(PubNubMessageAttribute.PARTITION_ID)
             if not partition_id:
@@ -106,3 +123,7 @@ def handle_pubnub_message(self, message: dict) -> None:
                 return
 
             alarm_panel.handle_pubnub_message(message)
+        else:
+            _LOGGER.warning(
+                "Unknown message received by system %s: %s", self.id, message
+            )

vivintpy/user.py

@@ -0,0 +1,78 @@
+"""Module that implements the User class."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from .const import UserAttribute as Attribute
+from .entity import Entity
+
+if TYPE_CHECKING:
+    from .system import System
+
+ADD_LOCK = f"{Attribute.LOCK_IDS}.1"
+
+
+class User(Entity):
+    """Describe a Vivint user."""
+
+    def __init__(self, data: dict, system: System):
+        """Initialize a user."""
+        super().__init__(data)
+        self._system = system
+
+    def __repr__(self) -> str:
+        """Return custom __repr__ of user."""
+        return f"<{self.__class__.__name__} {self.id}, {self.name}{' (admin)' if self.is_admin else ''}>"
+
+    @property
+    def has_lock_pin(self) -> bool:
+        """Return True if the user has pins."""
+        return bool(self.data[Attribute.HAS_LOCK_PIN])
+
+    @property
+    def has_panel_pin(self) -> bool:
+        """Return True if the user has pins."""
+        return bool(self.data[Attribute.HAS_PANEL_PIN])
+
+    @property
+    def has_pins(self) -> bool:
+        """Return True if the user has pins."""
+        return bool(self.data[Attribute.HAS_PINS])
+
+    @property
+    def has_remote_access(self) -> bool:
+        """Return True if the user has remote access."""
+        return bool(self.data[Attribute.REMOTE_ACCESS])
+
+    @property
+    def id(self) -> int:  # pylint: disable=invalid-name
+        """User's id."""
+        return int(self.data[Attribute.ID])
+
+    @property
+    def is_admin(self) -> bool:
+        """Return True if the user is an admin."""
+        return bool(self.data[Attribute.ADMIN])
+
+    @property
+    def is_registered(self) -> bool:
+        """Return True if the user is registered."""
+        return bool(self.data[Attribute.REGISTERED])
+
+    @property
+    def lock_ids(self) -> list[int]:
+        """User's lock ids."""
+        return cast(list[int], self.data.get(Attribute.LOCK_IDS, []))
+
+    @property
+    def name(self) -> str:
+        """User's name."""
+        return str(self.data[Attribute.NAME])
+
+    def handle_pubnub_message(self, message: dict) -> None:
+        """Handle a pubnub message addressed to this user."""
+        if ADD_LOCK in message:
+            message[Attribute.LOCK_IDS] = self.lock_ids + [message[ADD_LOCK]]
+            del message[ADD_LOCK]
+        super().handle_pubnub_message(message)