Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add logbook entries for zwave_js events (#72508) 

* Add logbook entries for zwave_js events

* Fix test

* Update homeassistant/components/zwave_js/logbook.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/zwave_js/logbook.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/zwave_js/logbook.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/zwave_js/logbook.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* black

* Remove value updated event

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/72824/head
Raman Gupta 2022-05-26 19:20:05 -04:00 committed by Paulus Schoutsen
parent a7fc1a4d62
commit 370e4c53f3
4 changed files with 251 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/zwave_js/__init__.py
View File

@ -297,8 +297,8 @@ async def setup_driver( # noqa: C901
if not disc_info.assumed_state:
return
value_updates_disc_info[disc_info.primary_value.value_id] = disc_info
# If this is the first time we found a value we want to watch for updates,
# return early
# If this is not the first time we found a value we want to watch for updates,
# return early because we only need one listener for all values.
if len(value_updates_disc_info) != 1:
return
# add listener for value updated events
@ -503,7 +503,7 @@ async def setup_driver( # noqa: C901
elif isinstance(notification, PowerLevelNotification):
event_data.update(
{
ATTR_COMMAND_CLASS_NAME: "Power Level",
ATTR_COMMAND_CLASS_NAME: "Powerlevel",
ATTR_TEST_NODE_ID: notification.test_node_id,
ATTR_STATUS: notification.status,
ATTR_ACKNOWLEDGED_FRAMES: notification.acknowledged_frames,

115
homeassistant/components/zwave_js/logbook.py Normal file
View File

@ -0,0 +1,115 @@
"""Describe Z-Wave JS logbook events."""
from __future__ import annotations
from collections.abc import Callable
from zwave_js_server.const import CommandClass
from homeassistant.components.logbook.const import (
LOGBOOK_ENTRY_MESSAGE,
LOGBOOK_ENTRY_NAME,
)
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import Event, HomeAssistant, callback
import homeassistant.helpers.device_registry as dr
from .const import (
ATTR_COMMAND_CLASS,
ATTR_COMMAND_CLASS_NAME,
ATTR_DATA_TYPE,
ATTR_DIRECTION,
ATTR_EVENT_LABEL,
ATTR_EVENT_TYPE,
ATTR_LABEL,
ATTR_VALUE,
DOMAIN,
ZWAVE_JS_NOTIFICATION_EVENT,
ZWAVE_JS_VALUE_NOTIFICATION_EVENT,
)
@callback
def async_describe_events(
hass: HomeAssistant,
async_describe_event: Callable[[str, str, Callable[[Event], dict[str, str]]], None],
) -> None:
"""Describe logbook events."""
dev_reg = dr.async_get(hass)
@callback
def async_describe_zwave_js_notification_event(
event: Event,
) -> dict[str, str]:
"""Describe Z-Wave JS notification event."""
device = dev_reg.devices[event.data[ATTR_DEVICE_ID]]
# Z-Wave JS devices always have a name
device_name = device.name_by_user or device.name
assert device_name
command_class = event.data[ATTR_COMMAND_CLASS]
command_class_name = event.data[ATTR_COMMAND_CLASS_NAME]
data: dict[str, str] = {LOGBOOK_ENTRY_NAME: device_name}
prefix = f"fired {command_class_name} CC 'notification' event"
if command_class == CommandClass.NOTIFICATION:
label = event.data[ATTR_LABEL]
event_label = event.data[ATTR_EVENT_LABEL]
return {
**data,
LOGBOOK_ENTRY_MESSAGE: f"{prefix} '{label}': '{event_label}'",
}
if command_class == CommandClass.ENTRY_CONTROL:
event_type = event.data[ATTR_EVENT_TYPE]
data_type = event.data[ATTR_DATA_TYPE]
return {
**data,
LOGBOOK_ENTRY_MESSAGE: (
f"{prefix} for event type '{event_type}' with data type "
f"'{data_type}'"
),
}
if command_class == CommandClass.SWITCH_MULTILEVEL:
event_type = event.data[ATTR_EVENT_TYPE]
direction = event.data[ATTR_DIRECTION]
return {
**data,
LOGBOOK_ENTRY_MESSAGE: (
f"{prefix} for event type '{event_type}': '{direction}'"
),
}
return {**data, LOGBOOK_ENTRY_MESSAGE: prefix}
@callback
def async_describe_zwave_js_value_notification_event(
event: Event,
) -> dict[str, str]:
"""Describe Z-Wave JS value notification event."""
device = dev_reg.devices[event.data[ATTR_DEVICE_ID]]
# Z-Wave JS devices always have a name
device_name = device.name_by_user or device.name
assert device_name
command_class = event.data[ATTR_COMMAND_CLASS_NAME]
label = event.data[ATTR_LABEL]
value = event.data[ATTR_VALUE]
return {
LOGBOOK_ENTRY_NAME: device_name,
LOGBOOK_ENTRY_MESSAGE: (
f"fired {command_class} CC 'value notification' event for '{label}': "
f"'{value}'"
),
}
async_describe_event(
DOMAIN, ZWAVE_JS_NOTIFICATION_EVENT, async_describe_zwave_js_notification_event
)
async_describe_event(
DOMAIN,
ZWAVE_JS_VALUE_NOTIFICATION_EVENT,
async_describe_zwave_js_value_notification_event,
)

2
tests/components/zwave_js/test_events.py
View File

@ -312,7 +312,7 @@ async def test_power_level_notification(hass, hank_binary_switch, integration, c
node.receive_event(event)
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["command_class_name"] == "Power Level"
assert events[0].data["command_class_name"] == "Powerlevel"
assert events[0].data["command_class"] == 115
assert events[0].data["test_node_id"] == 1
assert events[0].data["status"] == 0

132
tests/components/zwave_js/test_logbook.py Normal file
View File

@ -0,0 +1,132 @@
"""The tests for Z-Wave JS logbook."""
from zwave_js_server.const import CommandClass
from homeassistant.components.zwave_js.const import (
ZWAVE_JS_NOTIFICATION_EVENT,
ZWAVE_JS_VALUE_NOTIFICATION_EVENT,
)
from homeassistant.components.zwave_js.helpers import get_device_id
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from tests.components.logbook.common import MockRow, mock_humanify
async def test_humanifying_zwave_js_notification_event(
hass, client, lock_schlage_be469, integration
):
"""Test humanifying Z-Wave JS notification events."""
dev_reg = dr.async_get(hass)
device = dev_reg.async_get_device(
identifiers={get_device_id(client.driver, lock_schlage_be469)}
)
assert device
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
events = mock_humanify(
hass,
[
MockRow(
ZWAVE_JS_NOTIFICATION_EVENT,
{
"device_id": device.id,
"command_class": CommandClass.NOTIFICATION.value,
"command_class_name": "Notification",
"label": "label",
"event_label": "event_label",
},
),
MockRow(
ZWAVE_JS_NOTIFICATION_EVENT,
{
"device_id": device.id,
"command_class": CommandClass.ENTRY_CONTROL.value,
"command_class_name": "Entry Control",
"event_type": 1,
"data_type": 2,
},
),
MockRow(
ZWAVE_JS_NOTIFICATION_EVENT,
{
"device_id": device.id,
"command_class": CommandClass.SWITCH_MULTILEVEL.value,
"command_class_name": "Multilevel Switch",
"event_type": 1,
"direction": "up",
},
),
MockRow(
ZWAVE_JS_NOTIFICATION_EVENT,
{
"device_id": device.id,
"command_class": CommandClass.POWERLEVEL.value,
"command_class_name": "Powerlevel",
},
),
],
)
assert events[0]["name"] == "Touchscreen Deadbolt"
assert events[0]["domain"] == "zwave_js"
assert (
events[0]["message"]
== "fired Notification CC 'notification' event 'label': 'event_label'"
)
assert events[1]["name"] == "Touchscreen Deadbolt"
assert events[1]["domain"] == "zwave_js"
assert (
events[1]["message"]
== "fired Entry Control CC 'notification' event for event type '1' with data type '2'"
)
assert events[2]["name"] == "Touchscreen Deadbolt"
assert events[2]["domain"] == "zwave_js"
assert (
events[2]["message"]
== "fired Multilevel Switch CC 'notification' event for event type '1': 'up'"
)
assert events[3]["name"] == "Touchscreen Deadbolt"
assert events[3]["domain"] == "zwave_js"
assert events[3]["message"] == "fired Powerlevel CC 'notification' event"
async def test_humanifying_zwave_js_value_notification_event(
hass, client, lock_schlage_be469, integration
):
"""Test humanifying Z-Wave JS value notification events."""
dev_reg = dr.async_get(hass)
device = dev_reg.async_get_device(
identifiers={get_device_id(client.driver, lock_schlage_be469)}
)
assert device
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
events = mock_humanify(
hass,
[
MockRow(
ZWAVE_JS_VALUE_NOTIFICATION_EVENT,
{
"device_id": device.id,
"command_class": CommandClass.SCENE_ACTIVATION.value,
"command_class_name": "Scene Activation",
"label": "Scene ID",
"value": "001",
},
),
],
)
assert events[0]["name"] == "Touchscreen Deadbolt"
assert events[0]["domain"] == "zwave_js"
assert (
events[0]["message"]
== "fired Scene Activation CC 'value notification' event for 'Scene ID': '001'"
)