Skip to content

Commit

Permalink
Refactor notification module (#758)
Browse files Browse the repository at this point in the history
  • Loading branch information
@raman325
raman325 authored Sep 20, 2023
1 parent 221a038 commit 7c6e5a6
Show file tree
Hide file tree
Showing 4 changed files with 110 additions and 165 deletions.
15 changes: 12 additions & 3 deletions test/model/test_node.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
},
)

Expand All @@ -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):
Expand Down Expand Up @@ -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):
Expand Down
40 changes: 21 additions & 19 deletions zwave_js_server/model/node/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
3 changes: 3 additions & 0 deletions zwave_js_server/model/node/event_model.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from ...event import BaseEventModel
from ..notification import (
EntryControlNotificationArgsDataType,
MultilevelSwitchNotificationArgsDataType,
NotificationNotificationArgsDataType,
PowerLevelNotificationArgsDataType,
)
Expand Down Expand Up @@ -106,11 +107,13 @@ class NotificationEventModel(BaseNodeEventModel):
"""Model for `notification` event data."""

event: Literal["notification"]
nodeId: int
ccId: CommandClass
args: (
NotificationNotificationArgsDataType
| EntryControlNotificationArgsDataType
| PowerLevelNotificationArgsDataType
| MultilevelSwitchNotificationArgsDataType
)


Expand Down
217 changes: 74 additions & 143 deletions zwave_js_server/model/notification.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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."""

Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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")

0 comments on commit 7c6e5a6

Please sign in to comment.