From 7c6e5a65a32e923d918c71b55ae1cde9cf096361 Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Wed, 20 Sep 2023 10:48:18 -0400 Subject: [PATCH] Refactor notification module (#758) --- test/model/test_node.py | 15 +- zwave_js_server/model/node/__init__.py | 40 ++-- zwave_js_server/model/node/event_model.py | 3 + zwave_js_server/model/notification.py | 217 ++++++++-------------- 4 files changed, 110 insertions(+), 165 deletions(-) diff --git a/test/model/test_node.py b/test/model/test_node.py index 1ea0ead0f..b694409b8 100644 --- a/test/model/test_node.py +++ b/test/model/test_node.py @@ -963,7 +963,7 @@ async def test_notification(lock_schlage_be469: node_pkg.Node): "event": "notification", "nodeId": 23, "ccId": CommandClass.SWITCH_MULTILEVEL.value, - "args": {"eventType": 4}, + "args": {"eventType": 4, "eventTypeLabel": "c"}, }, ) @@ -975,6 +975,7 @@ async def test_notification(lock_schlage_be469: node_pkg.Node): event.data["notification"].event_type == MultilevelSwitchCommand.START_LEVEL_CHANGE ) + assert event.data["notification"].event_type_label == "c" async def test_notification_unknown(lock_schlage_be469: node_pkg.Node, caplog): @@ -1008,15 +1009,23 @@ async def test_entry_control_notification(ring_keypad): "event": "notification", "nodeId": 10, "ccId": 111, - "args": {"eventType": 5, "dataType": 2, "eventData": "555"}, + "args": { + "eventType": 5, + "eventTypeLabel": "foo", + "dataType": 2, + "dataTypeLabel": "bar", + "eventData": "cat", + }, }, ) node.handle_notification(event) assert event.data["notification"].command_class == CommandClass.ENTRY_CONTROL assert event.data["notification"].node_id == 10 assert event.data["notification"].event_type == EntryControlEventType.ARM_AWAY + assert event.data["notification"].event_type_label == "foo" assert event.data["notification"].data_type == EntryControlDataType.ASCII - assert event.data["notification"].event_data == "555" + assert event.data["notification"].data_type_label == "bar" + assert event.data["notification"].event_data == "cat" async def test_interview_events(multisensor_6): diff --git a/zwave_js_server/model/node/__init__.py b/zwave_js_server/model/node/__init__.py index 5452b1794..0980e3002 100644 --- a/zwave_js_server/model/node/__init__.py +++ b/zwave_js_server/model/node/__init__.py @@ -1058,25 +1058,27 @@ def handle_metadata_updated(self, event: Event) -> None: def handle_notification(self, event: Event) -> None: """Process a node notification event.""" - command_class = CommandClass(event.data["ccId"]) - if command_class == CommandClass.NOTIFICATION: - event.data["notification"] = NotificationNotification( - self, cast(NotificationNotificationDataType, event.data) - ) - elif command_class == CommandClass.SWITCH_MULTILEVEL: - event.data["notification"] = MultilevelSwitchNotification( - self, cast(MultilevelSwitchNotificationDataType, event.data) - ) - elif command_class == CommandClass.ENTRY_CONTROL: - event.data["notification"] = EntryControlNotification( - self, cast(EntryControlNotificationDataType, event.data) - ) - elif command_class == CommandClass.POWERLEVEL: - event.data["notification"] = PowerLevelNotification( - self, cast(PowerLevelNotificationDataType, event.data) - ) - else: - _LOGGER.info("Unhandled notification command class: %s", command_class.name) + match command_class := CommandClass(event.data["ccId"]): + case CommandClass.NOTIFICATION: + event.data["notification"] = NotificationNotification( + self, cast(NotificationNotificationDataType, event.data) + ) + case CommandClass.SWITCH_MULTILEVEL: + event.data["notification"] = MultilevelSwitchNotification( + self, cast(MultilevelSwitchNotificationDataType, event.data) + ) + case CommandClass.ENTRY_CONTROL: + event.data["notification"] = EntryControlNotification( + self, cast(EntryControlNotificationDataType, event.data) + ) + case CommandClass.POWERLEVEL: + event.data["notification"] = PowerLevelNotification( + self, cast(PowerLevelNotificationDataType, event.data) + ) + case _: + _LOGGER.info( + "Unhandled notification command class: %s", command_class.name + ) def handle_firmware_update_progress(self, event: Event) -> None: """Process a node firmware update progress event.""" diff --git a/zwave_js_server/model/node/event_model.py b/zwave_js_server/model/node/event_model.py index 82b3f00a1..a6459033c 100644 --- a/zwave_js_server/model/node/event_model.py +++ b/zwave_js_server/model/node/event_model.py @@ -7,6 +7,7 @@ from ...event import BaseEventModel from ..notification import ( EntryControlNotificationArgsDataType, + MultilevelSwitchNotificationArgsDataType, NotificationNotificationArgsDataType, PowerLevelNotificationArgsDataType, ) @@ -106,11 +107,13 @@ class NotificationEventModel(BaseNodeEventModel): """Model for `notification` event data.""" event: Literal["notification"] + nodeId: int ccId: CommandClass args: ( NotificationNotificationArgsDataType | EntryControlNotificationArgsDataType | PowerLevelNotificationArgsDataType + | MultilevelSwitchNotificationArgsDataType ) diff --git a/zwave_js_server/model/notification.py b/zwave_js_server/model/notification.py index 85031df02..282a06536 100644 --- a/zwave_js_server/model/notification.py +++ b/zwave_js_server/model/notification.py @@ -5,6 +5,7 @@ """ from __future__ import annotations +from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Literal, TypedDict from ..const.command_class.multilevel_switch import MultilevelSwitchCommand @@ -24,6 +25,21 @@ class BaseNotificationDataType(TypedDict): ccId: int # required +@dataclass +class BaseNotification: + """Model for a Zwave Node's notification event.""" + + node: Node + data: BaseNotificationDataType + node_id: int = field(init=False) + command_class: int = field(init=False) + + def __post_init__(self) -> None: + """Post initialization.""" + self.node_id = self.data["nodeId"] + self.command_class = self.data["ccId"] + + class EntryControlNotificationArgsDataType(TypedDict, total=False): """Represent args for a Entry Control CC notification event data dict type.""" @@ -40,50 +56,26 @@ class EntryControlNotificationDataType(BaseNotificationDataType): args: EntryControlNotificationArgsDataType # required -class EntryControlNotification: +@dataclass +class EntryControlNotification(BaseNotification): """Model for a Zwave Node's Entry Control CC notification event.""" - def __init__(self, node: "Node", data: EntryControlNotificationDataType) -> None: - """Initialize.""" - self.node = node - self.data = data - - @property - def node_id(self) -> int: - """Return node ID property.""" - return self.data["nodeId"] - - @property - def command_class(self) -> int: - """Return command class.""" - return self.data["ccId"] - - @property - def event_type(self) -> int: - """Return event type property.""" - return self.data["args"]["eventType"] - - @property - def event_type_label(self) -> str: - """Return event type label property.""" - return self.data["args"]["eventTypeLabel"] - - @property - def data_type(self) -> int: - """Return data type property.""" - return self.data["args"]["dataType"] - - @property - def data_type_label(self) -> str: - """Return data type label property.""" - return self.data["args"]["dataTypeLabel"] - - @property - def event_data(self) -> str | None: - """Return event data property.""" + data: EntryControlNotificationDataType + event_type: int = field(init=False) + event_type_label: str = field(init=False) + data_type: int = field(init=False) + data_type_label: str = field(init=False) + event_data: str | dict[str, Any] | None = field(init=False, default=None) + + def __post_init__(self) -> None: + """Post initialize.""" + super().__post_init__() + self.event_type = self.data["args"]["eventType"] + self.event_type_label = self.data["args"]["eventTypeLabel"] + self.data_type = self.data["args"]["dataType"] + self.data_type_label = self.data["args"]["dataTypeLabel"] if event_data := self.data["args"].get("eventData"): - return parse_buffer(event_data) - return None + self.event_data = parse_buffer(event_data) class NotificationNotificationArgsDataType(TypedDict, total=False): @@ -102,48 +94,25 @@ class NotificationNotificationDataType(BaseNotificationDataType): args: NotificationNotificationArgsDataType # required -class NotificationNotification: +@dataclass +class NotificationNotification(BaseNotification): """Model for a Zwave Node's Notification CC notification event.""" - def __init__(self, node: "Node", data: NotificationNotificationDataType) -> None: - """Initialize.""" - self.node = node - self.data = data - - @property - def node_id(self) -> int: - """Return node ID property.""" - return self.data["nodeId"] - - @property - def command_class(self) -> int: - """Return command class.""" - return self.data["ccId"] - - @property - def type_(self) -> int: - """Return type property.""" - return self.data["args"]["type"] + data: NotificationNotificationDataType + type_: int = field(init=False) + label: str = field(init=False) + event: int = field(init=False) + event_label: str = field(init=False) + parameters: dict[str, Any] = field(init=False) - @property - def label(self) -> str: - """Return label property.""" - return self.data["args"]["label"] - - @property - def event(self) -> int: - """Return event property.""" - return self.data["args"]["event"] - - @property - def event_label(self) -> str: - """Return notification label property.""" - return self.data["args"]["eventLabel"] - - @property - def parameters(self) -> dict[str, Any]: - """Return installer icon property.""" - return self.data["args"].get("parameters", {}) + def __post_init__(self) -> None: + """Post initialize.""" + super().__post_init__() + self.type_ = self.data["args"]["type"] + self.label = self.data["args"]["label"] + self.event = self.data["args"]["event"] + self.event_label = self.data["args"]["eventLabel"] + self.parameters = self.data["args"].get("parameters", {}) class PowerLevelNotificationArgsDataType(TypedDict): @@ -160,38 +129,21 @@ class PowerLevelNotificationDataType(BaseNotificationDataType): args: PowerLevelNotificationArgsDataType # required -class PowerLevelNotification: +@dataclass +class PowerLevelNotification(BaseNotification): """Model for a Zwave Node's Power Level CC notification event.""" - def __init__(self, node: "Node", data: PowerLevelNotificationDataType) -> None: - """Initialize.""" - self.node = node - self.data = data - - @property - def node_id(self) -> int: - """Return node ID property.""" - return self.data["nodeId"] - - @property - def command_class(self) -> int: - """Return command class.""" - return self.data["ccId"] - - @property - def test_node_id(self) -> int: - """Return test node ID property.""" - return self.data["args"]["testNodeId"] - - @property - def status(self) -> PowerLevelTestStatus: - """Return status.""" - return PowerLevelTestStatus(self.data["args"]["status"]) + data: PowerLevelNotificationDataType + test_node_id: int = field(init=False) + status: PowerLevelTestStatus = field(init=False) + acknowledged_frames: int = field(init=False) - @property - def acknowledged_frames(self) -> int: - """Return acknowledged frames property.""" - return self.data["args"]["acknowledgedFrames"] + def __post_init__(self) -> None: + """Post initialize.""" + super().__post_init__() + self.test_node_id = self.data["args"]["testNodeId"] + self.status = PowerLevelTestStatus(self.data["args"]["status"]) + self.acknowledged_frames = self.data["args"]["acknowledgedFrames"] class MultilevelSwitchNotificationArgsDataType(TypedDict, total=False): @@ -208,39 +160,18 @@ class MultilevelSwitchNotificationDataType(BaseNotificationDataType): args: MultilevelSwitchNotificationArgsDataType # required -class MultilevelSwitchNotification: +@dataclass +class MultilevelSwitchNotification(BaseNotification): """Model for a Zwave Node's Multi Level CC notification event.""" - def __init__( - self, node: "Node", data: MultilevelSwitchNotificationDataType - ) -> None: - """Initialize.""" - self.node = node - self.data = data - - @property - def node_id(self) -> int: - """Return node ID property.""" - return self.data["nodeId"] - - @property - def command_class(self) -> int: - """Return command class.""" - return self.data["ccId"] - - @property - def event_type(self) -> MultilevelSwitchCommand: - """Return event type property.""" - return MultilevelSwitchCommand(self.data["args"]["eventType"]) - - @property - def event_type_label(self) -> str: - """Return event type label property.""" - return self.data["args"]["eventTypeLabel"] - - @property - def direction(self) -> str | None: - """Return direction property.""" - if direction := self.data["args"].get("direction"): - return direction - return None + data: MultilevelSwitchNotificationDataType + event_type: MultilevelSwitchCommand = field(init=False) + event_type_label: str = field(init=False) + direction: str | None = field(init=False) + + def __post_init__(self) -> None: + """Post initialize.""" + super().__post_init__() + self.event_type = MultilevelSwitchCommand(self.data["args"]["eventType"]) + self.event_type_label = self.data["args"]["eventTypeLabel"] + self.direction = self.data["args"].get("direction")