diff --git a/test/model/test_node.py b/test/model/test_node.py index 1e6502f96..699530dd7 100644 --- a/test/model/test_node.py +++ b/test/model/test_node.py @@ -17,8 +17,14 @@ EntryControlDataType, EntryControlEventType, ) +from zwave_js_server.const.command_class.power_level import PowerLevelTestStatus from zwave_js_server.event import Event -from zwave_js_server.exceptions import FailedCommand, NotFoundError, UnwriteableValue +from zwave_js_server.exceptions import ( + FailedCommand, + NotFoundError, + NotificationHasUnsupportedCommandClass, + UnwriteableValue, +) from zwave_js_server.model import node as node_pkg from zwave_js_server.model.firmware import FirmwareUpdateStatus from zwave_js_server.model.node_health_check import ( @@ -712,6 +718,29 @@ async def test_notification(lock_schlage_be469: node_pkg.Node): """Test notification CC notification events.""" node = lock_schlage_be469 + # Validate that Entry Control CC notification event is received as expected + event = Event( + type="notification", + data={ + "source": "node", + "event": "notification", + "nodeId": 23, + "ccId": 111, + "args": { + "eventType": 0, + "dataType": 0, + "eventData": "test", + }, + }, + ) + + node.handle_notification(event) + assert event.data["notification"].command_class == CommandClass.ENTRY_CONTROL + assert event.data["notification"].node_id == 23 + assert event.data["notification"].event_type == EntryControlEventType.CACHING + assert event.data["notification"].data_type == EntryControlDataType.NONE + assert event.data["notification"].event_data == "test" + # Validate that Notification CC notification event is received as expected event = Event( type="notification", @@ -739,6 +768,39 @@ async def test_notification(lock_schlage_be469: node_pkg.Node): assert event.data["notification"].event_label == "Keypad lock operation" assert event.data["notification"].parameters == {"userId": 1} + # Validate that Power Level CC notification event is received as expected + event = Event( + type="notification", + data={ + "source": "node", + "event": "notification", + "nodeId": 23, + "ccId": CommandClass.POWERLEVEL.value, + "args": {"testNodeId": 1, "status": 0, "acknowledgedFrames": 2}, + }, + ) + + node.handle_notification(event) + assert event.data["notification"].command_class == CommandClass.POWERLEVEL + assert event.data["notification"].node_id == 23 + assert event.data["notification"].test_node_id == 1 + assert event.data["notification"].status == PowerLevelTestStatus.FAILED + assert event.data["notification"].acknowledged_frames == 2 + + # Validate that an unrecognized CC notification event raises Exception + event = Event( + type="notification", + data={ + "source": "node", + "event": "notification", + "nodeId": 23, + "ccId": 0, + }, + ) + + with pytest.raises(NotificationHasUnsupportedCommandClass): + node.handle_notification(event) + async def test_entry_control_notification(ring_keypad): """Test entry control CC notification events.""" diff --git a/zwave_js_server/const/command_class/power_level.py b/zwave_js_server/const/command_class/power_level.py new file mode 100644 index 000000000..e947eb4a6 --- /dev/null +++ b/zwave_js_server/const/command_class/power_level.py @@ -0,0 +1,11 @@ +"""Constants for the Power Level Command Class.""" +from enum import IntEnum + + +class PowerLevelTestStatus(IntEnum): + """Enum with all known power level test statuses.""" + + # https://github.com/zwave-js/node-zwave-js/blob/master/packages/zwave-js/src/lib/commandclass/PowerlevelCC.ts#L52 + FAILED = 0 + SUCCESS = 1 + IN_PROGRESS = 2 diff --git a/zwave_js_server/exceptions.py b/zwave_js_server/exceptions.py index 367855e7f..b6ee18d37 100644 --- a/zwave_js_server/exceptions.py +++ b/zwave_js_server/exceptions.py @@ -4,6 +4,7 @@ from .const import CommandClass if TYPE_CHECKING: + from .event import Event from .model.value import Value @@ -156,3 +157,16 @@ def __init__(self, value: "Value", path: str) -> None: "upstream issue with the driver or missing support for this data in the " "library" ) + + +class NotificationHasUnsupportedCommandClass(BaseZwaveJSServerError): + """Exception raised when notification is received for an unsupported CC.""" + + def __init__(self, event: "Event", command_class: CommandClass) -> None: + """Initialize an invalid Command Class error.""" + self.event_data = event.data + self.command_class = command_class + super().__init__( + "Notification received with unsupported command class " + f"{command_class.name}: {event.data}" + ) diff --git a/zwave_js_server/model/node.py b/zwave_js_server/model/node.py index 92e491193..1373cd2d1 100644 --- a/zwave_js_server/model/node.py +++ b/zwave_js_server/model/node.py @@ -12,6 +12,7 @@ from ..exceptions import ( FailedCommand, NotFoundError, + NotificationHasUnsupportedCommandClass, UnparseableValue, UnwriteableValue, ) @@ -37,6 +38,8 @@ EntryControlNotificationDataType, NotificationNotification, NotificationNotificationDataType, + PowerLevelNotification, + PowerLevelNotificationDataType, ) from .value import ( ConfigurationValue, @@ -811,14 +814,21 @@ def handle_metadata_updated(self, event: Event) -> None: def handle_notification(self, event: Event) -> None: """Process a node notification event.""" - if event.data["ccId"] == CommandClass.NOTIFICATION.value: + command_class = CommandClass(event.data["ccId"]) + if command_class == CommandClass.NOTIFICATION: event.data["notification"] = NotificationNotification( self, cast(NotificationNotificationDataType, event.data) ) - else: + 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: + raise NotificationHasUnsupportedCommandClass(event, command_class) def handle_firmware_update_progress(self, event: Event) -> None: """Process a node firmware update progress event.""" diff --git a/zwave_js_server/model/notification.py b/zwave_js_server/model/notification.py index b1c2c100b..df4893400 100644 --- a/zwave_js_server/model/notification.py +++ b/zwave_js_server/model/notification.py @@ -5,6 +5,7 @@ """ from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, TypedDict, Union +from zwave_js_server.const.command_class.power_level import PowerLevelTestStatus from zwave_js_server.util.helpers import parse_buffer @@ -12,8 +13,8 @@ from .node import Node -class NotificationDataType(TypedDict): - """Represent a generic notification event data dict type.""" +class BaseNotificationDataType(TypedDict): + """Represent a base notification event data dict type.""" source: Literal["node"] # required event: Literal["notification"] # required @@ -29,12 +30,48 @@ class EntryControlNotificationArgsDataType(TypedDict, total=False): eventData: Union[str, Dict[str, Any]] -class EntryControlNotificationDataType(NotificationDataType): +class EntryControlNotificationDataType(BaseNotificationDataType): """Represent an Entry Control CC notification event data dict type.""" args: EntryControlNotificationArgsDataType # required +class EntryControlNotification: + """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 data_type(self) -> int: + """Return data type property.""" + return self.data["args"]["dataType"] + + @property + def event_data(self) -> Optional[str]: + """Return event data property.""" + if event_data := self.data["args"].get("eventData"): + return parse_buffer(event_data) + return None + + class NotificationNotificationArgsDataType(TypedDict, total=False): """Represent args for a Notification CC notification event data dict type.""" @@ -45,7 +82,7 @@ class NotificationNotificationArgsDataType(TypedDict, total=False): parameters: Dict[str, Any] -class NotificationNotificationDataType(NotificationDataType): +class NotificationNotificationDataType(BaseNotificationDataType): """Represent a Notification CC notification event data dict type.""" args: NotificationNotificationArgsDataType # required @@ -95,10 +132,24 @@ def parameters(self) -> Dict[str, Any]: return self.data["args"].get("parameters", {}) -class EntryControlNotification: - """Model for a Zwave Node's Entry Control CC notification event.""" +class PowerLevelNotificationArgsDataType(TypedDict): + """Represent args for a Power Level CC notification event data dict type.""" - def __init__(self, node: "Node", data: EntryControlNotificationDataType) -> None: + testNodeId: int + status: int + acknowledgedFrames: int + + +class PowerLevelNotificationDataType(BaseNotificationDataType): + """Represent a Power Level CC notification event data dict type.""" + + args: PowerLevelNotificationArgsDataType # required + + +class PowerLevelNotification: + """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 @@ -114,18 +165,16 @@ def command_class(self) -> int: return self.data["ccId"] @property - def event_type(self) -> int: - """Return event type property.""" - return self.data["args"]["eventType"] + def test_node_id(self) -> int: + """Return test node ID property.""" + return self.data["args"]["testNodeId"] @property - def data_type(self) -> int: - """Return data type property.""" - return self.data["args"]["dataType"] + def status(self) -> PowerLevelTestStatus: + """Return status.""" + return PowerLevelTestStatus(self.data["args"]["status"]) @property - def event_data(self) -> Optional[str]: - """Return event data property.""" - if event_data := self.data["args"].get("eventData"): - return parse_buffer(event_data) - return None + def acknowledged_frames(self) -> int: + """Return acknowledged frames property.""" + return self.data["args"]["acknowledgedFrames"]