Add zwave_js MultilevelSwitch Notification (#70470)

Co-authored-by: Raman Gupta <7243222+raman325@users.noreply.github.com>
pull/70885/head
Reuben Bijl 2022-04-27 19:12:23 +12:00 committed by GitHub
parent 21badfc40f
commit 3901b5c8e8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 35 additions and 1 deletions

View File

@ -12,6 +12,7 @@ from zwave_js_server.exceptions import BaseZwaveJSServerError, InvalidServerVers
from zwave_js_server.model.node import Node as ZwaveNode from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.notification import ( from zwave_js_server.model.notification import (
EntryControlNotification, EntryControlNotification,
MultilevelSwitchNotification,
NotificationNotification, NotificationNotification,
PowerLevelNotification, PowerLevelNotification,
) )
@ -47,6 +48,7 @@ from .const import (
ATTR_COMMAND_CLASS, ATTR_COMMAND_CLASS,
ATTR_COMMAND_CLASS_NAME, ATTR_COMMAND_CLASS_NAME,
ATTR_DATA_TYPE, ATTR_DATA_TYPE,
ATTR_DIRECTION,
ATTR_ENDPOINT, ATTR_ENDPOINT,
ATTR_EVENT, ATTR_EVENT,
ATTR_EVENT_DATA, ATTR_EVENT_DATA,
@ -403,7 +405,7 @@ async def async_setup_entry( # noqa: C901
if "notification" not in event: if "notification" not in event:
LOGGER.info("Unknown notification: %s", event) LOGGER.info("Unknown notification: %s", event)
return return
notification: EntryControlNotification | NotificationNotification | PowerLevelNotification = event[ notification: EntryControlNotification | NotificationNotification | PowerLevelNotification | MultilevelSwitchNotification = event[
"notification" "notification"
] ]
device = dev_reg.async_get_device({get_device_id(client, notification.node)}) device = dev_reg.async_get_device({get_device_id(client, notification.node)})
@ -446,6 +448,14 @@ async def async_setup_entry( # noqa: C901
ATTR_ACKNOWLEDGED_FRAMES: notification.acknowledged_frames, ATTR_ACKNOWLEDGED_FRAMES: notification.acknowledged_frames,
} }
) )
elif isinstance(notification, MultilevelSwitchNotification):
event_data.update(
{
ATTR_COMMAND_CLASS_NAME: "Multilevel Switch",
ATTR_EVENT_TYPE: notification.event_type,
ATTR_DIRECTION: notification.direction,
}
)
else: else:
raise TypeError(f"Unhandled notification type: {notification}") raise TypeError(f"Unhandled notification type: {notification}")

View File

@ -45,6 +45,7 @@ ATTR_PROPERTY_KEY_NAME = "property_key_name"
ATTR_PROPERTY = "property" ATTR_PROPERTY = "property"
ATTR_PROPERTY_KEY = "property_key" ATTR_PROPERTY_KEY = "property_key"
ATTR_PARAMETERS = "parameters" ATTR_PARAMETERS = "parameters"
ATTR_DIRECTION = "direction"
ATTR_EVENT = "event" ATTR_EVENT = "event"
ATTR_EVENT_LABEL = "event_label" ATTR_EVENT_LABEL = "event_label"
ATTR_EVENT_TYPE = "event_type" ATTR_EVENT_TYPE = "event_type"

View File

@ -198,6 +198,29 @@ async def test_notifications(hass, hank_binary_switch, integration, client):
assert events[1].data["command_class"] == CommandClass.ENTRY_CONTROL assert events[1].data["command_class"] == CommandClass.ENTRY_CONTROL
assert events[1].data["command_class_name"] == "Entry Control" assert events[1].data["command_class_name"] == "Entry Control"
# Publish fake Multilevel Switch CC notification
event = Event(
type="notification",
data={
"source": "node",
"event": "notification",
"nodeId": 32,
"ccId": 38,
"args": {"eventType": 4, "direction": "up"},
},
)
node.receive_event(event)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 3
assert events[2].data["home_id"] == client.driver.controller.home_id
assert events[2].data["node_id"] == 32
assert events[2].data["event_type"] == 4
assert events[2].data["direction"] == "up"
assert events[2].data["command_class"] == CommandClass.SWITCH_MULTILEVEL
assert events[2].data["command_class_name"] == "Multilevel Switch"
async def test_value_updated(hass, vision_security_zl7432, integration, client): async def test_value_updated(hass, vision_security_zl7432, integration, client):
"""Test value updated events.""" """Test value updated events."""