From 9ed50d8b0c487dc6693a408b86a3d522ac8171a0 Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Wed, 31 Jan 2024 01:17:43 -0500 Subject: [PATCH] Add last seen sensor for zwave_js devices (#107345) --- homeassistant/components/zwave_js/sensor.py | 110 ++++++++++++-------- tests/components/zwave_js/test_sensor.py | 2 + 2 files changed, 69 insertions(+), 43 deletions(-) diff --git a/homeassistant/components/zwave_js/sensor.py b/homeassistant/components/zwave_js/sensor.py index f89498af72b..9fed7158d4a 100644 --- a/homeassistant/components/zwave_js/sensor.py +++ b/homeassistant/components/zwave_js/sensor.py @@ -1,8 +1,10 @@ """Representation of Z-Wave sensors.""" from __future__ import annotations -from collections.abc import Mapping -from typing import cast +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from datetime import datetime +from typing import Any, cast import voluptuous as vol from zwave_js_server.client import Client as ZwaveClient @@ -326,131 +328,166 @@ ENTITY_DESCRIPTION_KEY_MAP = { } +def convert_dict_of_dicts( + statistics: ControllerStatisticsDataType | NodeStatisticsDataType, key: str +) -> Any: + """Convert a dictionary of dictionaries to a value.""" + keys = key.split(".") + return statistics.get(keys[0], {}).get(keys[1], {}).get(keys[2]) # type: ignore[attr-defined] + + +@dataclass(frozen=True, kw_only=True) +class ZWaveJSStatisticsSensorEntityDescription(SensorEntityDescription): + """Class to represent a Z-Wave JS statistics sensor entity description.""" + + convert: Callable[ + [ControllerStatisticsDataType | NodeStatisticsDataType, str], Any + ] = lambda statistics, key: statistics.get(key) + + # Controller statistics descriptions ENTITY_DESCRIPTION_CONTROLLER_STATISTICS_LIST = [ - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="messagesTX", name="Successful messages (TX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="messagesRX", name="Successful messages (RX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="messagesDroppedTX", name="Messages dropped (TX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="messagesDroppedRX", name="Messages dropped (RX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="NAK", name="Messages not accepted", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="CAN", name="Collisions", state_class=SensorStateClass.TOTAL ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="timeoutACK", name="Missing ACKs", state_class=SensorStateClass.TOTAL ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="timeoutResponse", name="Timed out responses", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="timeoutCallback", name="Timed out callbacks", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel0.average", name="Average background RSSI (channel 0)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, + convert=convert_dict_of_dicts, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel0.current", name="Current background RSSI (channel 0)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, state_class=SensorStateClass.MEASUREMENT, + convert=convert_dict_of_dicts, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel1.average", name="Average background RSSI (channel 1)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, + convert=convert_dict_of_dicts, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel1.current", name="Current background RSSI (channel 1)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, state_class=SensorStateClass.MEASUREMENT, + convert=convert_dict_of_dicts, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel2.average", name="Average background RSSI (channel 2)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, + convert=convert_dict_of_dicts, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="backgroundRSSI.channel2.current", name="Current background RSSI (channel 2)", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, state_class=SensorStateClass.MEASUREMENT, + convert=convert_dict_of_dicts, ), ] # Node statistics descriptions ENTITY_DESCRIPTION_NODE_STATISTICS_LIST = [ - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="commandsRX", name="Successful commands (RX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="commandsTX", name="Successful commands (TX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="commandsDroppedRX", name="Commands dropped (RX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="commandsDroppedTX", name="Commands dropped (TX)", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="timeoutResponse", name="Timed out responses", state_class=SensorStateClass.TOTAL, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="rtt", name="Round Trip Time", native_unit_of_measurement=UnitOfTime.MILLISECONDS, device_class=SensorDeviceClass.DURATION, state_class=SensorStateClass.MEASUREMENT, ), - SensorEntityDescription( + ZWaveJSStatisticsSensorEntityDescription( key="rssi", name="RSSI", native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, device_class=SensorDeviceClass.SIGNAL_STRENGTH, state_class=SensorStateClass.MEASUREMENT, ), + ZWaveJSStatisticsSensorEntityDescription( + key="lastSeen", + name="Last Seen", + device_class=SensorDeviceClass.TIMESTAMP, + convert=( + lambda statistics, key: ( + datetime.fromisoformat(dt) # type: ignore[arg-type] + if (dt := statistics.get(key)) + else None + ) + ), + ), ] @@ -890,6 +927,7 @@ class ZWaveControllerStatusSensor(SensorEntity): class ZWaveStatisticsSensor(SensorEntity): """Representation of a node/controller statistics sensor.""" + entity_description: ZWaveJSStatisticsSensorEntityDescription _attr_should_poll = False _attr_entity_category = EntityCategory.DIAGNOSTIC _attr_entity_registry_enabled_default = False @@ -900,7 +938,7 @@ class ZWaveStatisticsSensor(SensorEntity): config_entry: ConfigEntry, driver: Driver, statistics_src: ZwaveNode | Controller, - description: SensorEntityDescription, + description: ZWaveJSStatisticsSensorEntityDescription, ) -> None: """Initialize a Z-Wave statistics entity.""" self.entity_description = description @@ -929,25 +967,11 @@ class ZWaveStatisticsSensor(SensorEntity): " service won't work for it" ) - def _get_data_from_statistics( - self, statistics: ControllerStatisticsDataType | NodeStatisticsDataType - ) -> int | None: - """Get the data from the statistics dict.""" - if "." not in self.entity_description.key: - return cast(int | None, statistics.get(self.entity_description.key)) - - # If key contains dots, we need to traverse the dict to get to the right value - for key in self.entity_description.key.split("."): - if key not in statistics: - return None - statistics = statistics[key] # type: ignore[literal-required] - return cast(int, statistics) - @callback def statistics_updated(self, event_data: dict) -> None: """Call when statistics updated event is received.""" - self._attr_native_value = self._get_data_from_statistics( - event_data["statistics"] + self._attr_native_value = self.entity_description.convert( + event_data["statistics"], self.entity_description.key ) self.async_write_ha_state() @@ -972,6 +996,6 @@ class ZWaveStatisticsSensor(SensorEntity): ) # Set initial state - self._attr_native_value = self._get_data_from_statistics( - self.statistics_src.statistics.data + self._attr_native_value = self.entity_description.convert( + self.statistics_src.statistics.data, self.entity_description.key ) diff --git a/tests/components/zwave_js/test_sensor.py b/tests/components/zwave_js/test_sensor.py index 96ad8bfa82c..4e88b2b50cc 100644 --- a/tests/components/zwave_js/test_sensor.py +++ b/tests/components/zwave_js/test_sensor.py @@ -731,6 +731,7 @@ NODE_STATISTICS_SUFFIXES = { NODE_STATISTICS_SUFFIXES_UNKNOWN = { "round_trip_time": 6, "rssi": 7, + "last_seen": "2024-01-01T00:00:00+00:00", } @@ -843,6 +844,7 @@ async def test_statistics_sensors( "repeaterRSSI": [], "routeFailedBetween": [], }, + "lastSeen": "2024-01-01T00:00:00+0000", }, }, )