From e73c6ae5f6c50af4087a04dcf38de47d4e563bfc Mon Sep 17 00:00:00 2001 From: Nathan Spencer Date: Fri, 11 Oct 2024 17:19:29 -0600 Subject: [PATCH 1/2] Add user data to system --- vivintpy/const.py | 14 +++++++ vivintpy/devices/door_lock.py | 6 +++ vivintpy/system.py | 25 ++++++++++- vivintpy/user.py | 78 +++++++++++++++++++++++++++++++++++ 4 files changed, 121 insertions(+), 2 deletions(-) create mode 100644 vivintpy/user.py diff --git a/vivintpy/const.py b/vivintpy/const.py index 328e27e..23e4949 100644 --- a/vivintpy/const.py +++ b/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.""" diff --git a/vivintpy/devices/door_lock.py b/vivintpy/devices/door_lock.py index a1efcfc..238a250 100644 --- a/vivintpy/devices/door_lock.py +++ b/vivintpy/devices/door_lock.py @@ -2,6 +2,7 @@ from __future__ import annotations +from ..const import LockAttribute from ..const import ZWaveDeviceAttribute as Attribute from ..utils import send_deprecation_warning from . import BypassTamperDevice @@ -26,6 +27,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 self.data.get(LockAttribute.USER_CODE_LIST, []) + async def set_state(self, locked: bool) -> None: """Set door lock's state.""" assert self.alarm_panel diff --git a/vivintpy/system.py b/vivintpy/system.py index fe287e5..5f943c2 100644 --- a/vivintpy/system.py +++ b/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, d=d: 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 + ) diff --git a/vivintpy/user.py b/vivintpy/user.py new file mode 100644 index 0000000..d7c7a6c --- /dev/null +++ b/vivintpy/user.py @@ -0,0 +1,78 @@ +"""Module that implements the User class.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +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 self.data[Attribute.HAS_LOCK_PIN] + + @property + def has_panel_pin(self) -> bool: + """Return True if the user has pins.""" + return self.data[Attribute.HAS_PANEL_PIN] + + @property + def has_pins(self) -> bool: + """Return True if the user has pins.""" + return self.data[Attribute.HAS_PINS] + + @property + def has_remote_access(self) -> bool: + """Return True if the user has remote access.""" + return 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 self.data[Attribute.ADMIN] + + @property + def is_registered(self) -> bool: + """Return True if the user is registered.""" + return self.data[Attribute.REGISTERED] + + @property + def lock_ids(self) -> list[int]: + """User's lock ids.""" + return self.data.get(Attribute.LOCK_IDS, []) + + @property + def name(self) -> str: + """User's name.""" + return 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) From 3fea32e48460745d12113da1f578a8be6a1b69a3 Mon Sep 17 00:00:00 2001 From: Nathan Spencer Date: Fri, 11 Oct 2024 17:25:56 -0600 Subject: [PATCH 2/2] Fix mypy errors --- vivintpy/devices/door_lock.py | 4 +++- vivintpy/system.py | 2 +- vivintpy/user.py | 18 +++++++++--------- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/vivintpy/devices/door_lock.py b/vivintpy/devices/door_lock.py index 238a250..223ee7b 100644 --- a/vivintpy/devices/door_lock.py +++ b/vivintpy/devices/door_lock.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import cast + from ..const import LockAttribute from ..const import ZWaveDeviceAttribute as Attribute from ..utils import send_deprecation_warning @@ -30,7 +32,7 @@ def node_online(self) -> bool: @property def user_code_list(self) -> list[int]: """Return the user code list.""" - return self.data.get(LockAttribute.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.""" diff --git a/vivintpy/system.py b/vivintpy/system.py index 5f943c2..d3a98ec 100644 --- a/vivintpy/system.py +++ b/vivintpy/system.py @@ -78,7 +78,7 @@ async def refresh(self) -> None: def update_user_data(self, data: list[dict]) -> None: """Update user data.""" for d in data: - user = first_or_none(self.users, lambda user, d=d: user.id == d["_id"]) + 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 diff --git a/vivintpy/user.py b/vivintpy/user.py index d7c7a6c..7589ff4 100644 --- a/vivintpy/user.py +++ b/vivintpy/user.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from .const import UserAttribute as Attribute from .entity import Entity @@ -28,22 +28,22 @@ def __repr__(self) -> str: @property def has_lock_pin(self) -> bool: """Return True if the user has pins.""" - return self.data[Attribute.HAS_LOCK_PIN] + return bool(self.data[Attribute.HAS_LOCK_PIN]) @property def has_panel_pin(self) -> bool: """Return True if the user has pins.""" - return self.data[Attribute.HAS_PANEL_PIN] + return bool(self.data[Attribute.HAS_PANEL_PIN]) @property def has_pins(self) -> bool: """Return True if the user has pins.""" - return self.data[Attribute.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 self.data[Attribute.REMOTE_ACCESS] + return bool(self.data[Attribute.REMOTE_ACCESS]) @property def id(self) -> int: # pylint: disable=invalid-name @@ -53,22 +53,22 @@ def id(self) -> int: # pylint: disable=invalid-name @property def is_admin(self) -> bool: """Return True if the user is an admin.""" - return self.data[Attribute.ADMIN] + return bool(self.data[Attribute.ADMIN]) @property def is_registered(self) -> bool: """Return True if the user is registered.""" - return self.data[Attribute.REGISTERED] + return bool(self.data[Attribute.REGISTERED]) @property def lock_ids(self) -> list[int]: """User's lock ids.""" - return self.data.get(Attribute.LOCK_IDS, []) + return cast(list[int], self.data.get(Attribute.LOCK_IDS, [])) @property def name(self) -> str: """User's name.""" - return self.data[Attribute.NAME] + return str(self.data[Attribute.NAME]) def handle_pubnub_message(self, message: dict) -> None: """Handle a pubnub message addressed to this user."""