Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 36 additions & 5 deletions test/model/test_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,18 @@
SupervisionStatus,
Weekday,
)
from zwave_js_server.const.command_class.battery import BatteryReplacementStatus
from zwave_js_server.const.command_class.entry_control import (
EntryControlDataType,
EntryControlEventType,
)
from zwave_js_server.const.command_class.multilevel_switch import (
MultilevelSwitchCommand,
)
from zwave_js_server.const.command_class.notification import (
AccessControlNotificationEvent,
NotificationType,
)
from zwave_js_server.const.command_class.power_level import PowerLevelTestStatus
from zwave_js_server.event import Event
from zwave_js_server.exceptions import (
Expand Down Expand Up @@ -1039,8 +1044,11 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
assert event.data["notification"].command_class == CommandClass.NOTIFICATION
assert event.data["notification"].node_id == 23
assert event.data["notification"].endpoint_idx == 0
assert event.data["notification"].type_ == 6
assert event.data["notification"].event == 5
assert event.data["notification"].type_ == NotificationType.ACCESS_CONTROL
assert (
event.data["notification"].event
== AccessControlNotificationEvent.KEYPAD_LOCK_OPERATION
)
assert event.data["notification"].label == "Access Control"
assert event.data["notification"].event_label == "Keypad lock operation"
assert event.data["notification"].parameters == {"userId": 1}
Expand All @@ -1053,7 +1061,7 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
"event": "notification",
"nodeId": 23,
"endpointIndex": 0,
"ccId": CommandClass.POWERLEVEL.value,
"ccId": 115,
"args": {"testNodeId": 1, "status": 0, "acknowledgedFrames": 2},
},
)
Expand All @@ -1074,7 +1082,7 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
"event": "notification",
"nodeId": 23,
"endpointIndex": 0,
"ccId": CommandClass.SWITCH_MULTILEVEL.value,
"ccId": 38,
"args": {"direction": "up", "eventType": 4, "eventTypeLabel": "c"},
},
)
Expand All @@ -1098,7 +1106,7 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
"event": "notification",
"nodeId": 23,
"endpointIndex": 0,
"ccId": CommandClass.SWITCH_MULTILEVEL.value,
"ccId": 38,
"args": {"eventType": 4, "eventTypeLabel": "c"},
},
)
Expand All @@ -1114,6 +1122,29 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
)
assert event.data["notification"].event_type_label == "c"

# Validate that Battery CC notification event is received as expected
event = Event(
type="notification",
data={
"source": "node",
"event": "notification",
"nodeId": 23,
"ccId": 128,
"endpointIndex": 0,
"args": {
"eventType": "battery low",
"urgency": 1,
},
},
)

node.handle_notification(event)
assert event.data["notification"].command_class == CommandClass.BATTERY
assert event.data["notification"].node_id == 23
assert event.data["notification"].endpoint_idx == 0
assert event.data["notification"].event_type == "battery low"
assert event.data["notification"].urgency == BatteryReplacementStatus.SOON


async def test_notification_unknown(lock_schlage_be469: node_pkg.Node, caplog):
"""Test unrecognized command class notification events."""
Expand Down
14 changes: 14 additions & 0 deletions zwave_js_server/const/command_class/battery.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
"""Constants for the Battery CC."""

from __future__ import annotations

from enum import IntEnum


class BatteryReplacementStatus(IntEnum):
"""Enum with all (known/used) Z-Wave Battery Replacement Statuses."""

# https://github.com/zwave-js/node-zwave-js/blob/master/packages/cc/src/lib/_Types.ts#L328
NO = 0
SOON = 1
NOW = 2
20 changes: 13 additions & 7 deletions zwave_js_server/model/node/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
from ..device_config import DeviceConfig
from ..endpoint import Endpoint, EndpointDataType
from ..notification import (
BatteryNotification,
BatteryNotificationDataType,
EntryControlNotification,
EntryControlNotificationDataType,
MultilevelSwitchNotification,
Expand Down Expand Up @@ -1105,22 +1107,26 @@ def handle_metadata_updated(self, event: Event) -> None:
def handle_notification(self, event: Event) -> None:
"""Process a node notification event."""
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.BATTERY:
event.data["notification"] = BatteryNotification(
self, cast(BatteryNotificationDataType, event.data)
)
case CommandClass.ENTRY_CONTROL:
event.data["notification"] = EntryControlNotification(
self, cast(EntryControlNotificationDataType, event.data)
)
case CommandClass.NOTIFICATION:
event.data["notification"] = NotificationNotification(
self, cast(NotificationNotificationDataType, event.data)
)
case CommandClass.POWERLEVEL:
event.data["notification"] = PowerLevelNotification(
self, cast(PowerLevelNotificationDataType, event.data)
)
case CommandClass.SWITCH_MULTILEVEL:
event.data["notification"] = MultilevelSwitchNotification(
self, cast(MultilevelSwitchNotificationDataType, event.data)
)
case _:
_LOGGER.info(
"Unhandled notification command class: %s", command_class.name
Expand Down
29 changes: 29 additions & 0 deletions zwave_js_server/model/notification.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal, TypedDict

from ..const.command_class.battery import BatteryReplacementStatus
from ..const.command_class.multilevel_switch import MultilevelSwitchCommand
from ..const.command_class.power_level import PowerLevelTestStatus
from ..util.helpers import parse_buffer
Expand Down Expand Up @@ -44,6 +45,34 @@ def __post_init__(self) -> None:
self.command_class = self.data["ccId"]


class BatteryNotificationArgsDataType(TypedDict):
"""Represent args for a Battery CC notification event data dict type."""

eventType: Literal["battery low"] # required
urgency: int # required


class BatteryNotificationDataType(BaseNotificationDataType):
"""Represent a Battery CC notification event data dict type."""

args: BatteryNotificationArgsDataType # required


@dataclass
class BatteryNotification(BaseNotification):
"""Model for a Zwave Node's Battery CC notification event."""

data: BatteryNotificationDataType = field(repr=False)
event_type: str = field(init=False)
urgency: BatteryReplacementStatus = field(init=False)

def __post_init__(self) -> None:
"""Post initialize."""
super().__post_init__()
self.event_type = self.data["args"]["eventType"]
self.urgency = BatteryReplacementStatus(self.data["args"]["urgency"])


class EntryControlNotificationArgsDataType(TypedDict, total=False):
"""Represent args for a Entry Control CC notification event data dict type."""

Expand Down