Add Z-Wave battery low event entity

homeassistant/components/zwave_js/__init__.py

@@ -17,6 +17,7 @@
 from zwave_js_server.model.driver import Driver
 from zwave_js_server.model.node import Node as ZwaveNode
 from zwave_js_server.model.notification import (
+    BatteryNotification,
     EntryControlNotification,
     MultilevelSwitchNotification,
     NotificationNotification,
@@ -79,6 +80,7 @@
     ATTR_STATUS,
     ATTR_TEST_NODE_ID,
     ATTR_TYPE,
+    ATTR_URGENCY,
     ATTR_VALUE,
     ATTR_VALUE_RAW,
     CONF_ADDON_DEVICE,
@@ -826,6 +828,17 @@ async def async_on_node_ready(self, node: ZwaveNode) -> None:
                 node,
             )
 
+        # Create a battery low event entity for each non-controller node that
+        # supports the Battery CC.
+        if not node.is_controller_node and any(
+            cc.id == CommandClass.BATTERY.value for cc in node.command_classes
+        ):
+            async_dispatcher_send(
+                self.hass,
+                f"{DOMAIN}_{self.config_entry.entry_id}_add_battery_low_event_entity",
+                node,
+            )
+
         # After ensuring the node is set up in HA, we should check if the node's
         # device config has changed, and if so, issue a repair registry entry for a
         # possible reinterview
@@ -965,7 +978,8 @@ def async_on_notification(self, event: dict[str, Any]) -> None:
 
         driver = self.controller_events.driver_events.driver
         notification: (
-            EntryControlNotification
+            BatteryNotification
+            | EntryControlNotification
             | NotificationNotification
             | PowerLevelNotification
             | MultilevelSwitchNotification
@@ -985,7 +999,15 @@ def async_on_notification(self, event: dict[str, Any]) -> None:
             ATTR_COMMAND_CLASS: notification.command_class,
         }
 
-        if isinstance(notification, EntryControlNotification):
+        if isinstance(notification, BatteryNotification):
+            event_data.update(
+                {
+                    ATTR_COMMAND_CLASS_NAME: "Battery",
+                    ATTR_EVENT_TYPE: notification.event_type,
+                    ATTR_URGENCY: notification.urgency,
+                }
+            )
+        elif isinstance(notification, EntryControlNotification):
             event_data.update(
                 {
                     ATTR_COMMAND_CLASS_NAME: "Entry Control",

homeassistant/components/zwave_js/const.py

@@ -85,6 +85,7 @@
 ATTR_DATA_TYPE_LABEL = "data_type_label"
 ATTR_NOTIFICATION_TYPE = "notification_type"
 ATTR_NOTIFICATION_EVENT = "notification_event"
+ATTR_URGENCY = "urgency"
 
 ATTR_NODE = "node"
 ATTR_ZWAVE_VALUE = "zwave_value"

homeassistant/components/zwave_js/event.py

@@ -1,22 +1,27 @@
 """Support for Z-Wave controls using the event platform."""
 
 from dataclasses import dataclass
+from typing import Any
 
+from zwave_js_server.const.command_class.battery import BatteryReplacementStatus
 from zwave_js_server.model.driver import Driver
+from zwave_js_server.model.node import Node as ZwaveNode
+from zwave_js_server.model.notification import BatteryNotification
 from zwave_js_server.model.value import Value, ValueNotification
 
 from homeassistant.components.event import (
     DOMAIN as EVENT_DOMAIN,
     EventEntity,
     EventEntityDescription,
 )
-from homeassistant.const import Platform
+from homeassistant.const import EntityCategory, Platform
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from .const import ATTR_VALUE, DOMAIN
+from .const import ATTR_URGENCY, ATTR_VALUE, DOMAIN
 from .entity import NewZwaveDiscoveryInfo, ZWaveBaseEntity
+from .helpers import get_device_info, get_valueless_base_unique_id
 from .models import (
     NewZWaveDiscoverySchema,
     ZwaveJSConfigEntry,
@@ -49,6 +54,13 @@ def async_add_event(info: NewZwaveDiscoveryInfo) -> None:
         ]
         async_add_entities(entities)
 
+    @callback
+    def async_add_battery_low_event_entity(node: ZwaveNode) -> None:
+        """Add a battery low event entity for the given node."""
+        driver = client.driver
+        assert driver is not None  # Driver is ready before platforms are loaded.
+        async_add_entities([ZWaveBatteryLowEventEntity(driver, node)])
+
     config_entry.async_on_unload(
         async_dispatcher_connect(
             hass,
@@ -57,6 +69,14 @@ def async_add_event(info: NewZwaveDiscoveryInfo) -> None:
         )
     )
 
+    config_entry.async_on_unload(
+        async_dispatcher_connect(
+            hass,
+            f"{DOMAIN}_{config_entry.entry_id}_add_battery_low_event_entity",
+            async_add_battery_low_event_entity,
+        )
+    )
+
 
 def _cc_and_label(value: Value) -> str:
     """Return a string with the command class and label."""
@@ -116,6 +136,58 @@ async def async_added_to_hass(self) -> None:
         )
 
 
+class ZWaveBatteryLowEventEntity(EventEntity):
+    """Representation of a Battery CC battery low event entity."""
+
+    _attr_entity_category = EntityCategory.DIAGNOSTIC
+    _attr_event_types = [
+        BatteryReplacementStatus.SOON.name.lower(),
+        BatteryReplacementStatus.NOW.name.lower(),
+    ]
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+    _attr_translation_key = "battery_low"
+
+    def __init__(self, driver: Driver, node: ZwaveNode) -> None:
+        """Initialize a Battery low event entity."""
+        self.node = node
+        self._base_unique_id = get_valueless_base_unique_id(driver, node)
+        self._attr_unique_id = f"{self._base_unique_id}.battery_low"
+        # device may not be precreated in main handler yet
+        self._attr_device_info = get_device_info(driver, node)
+
+    @callback
+    def _async_handle_notification(self, event: dict[str, Any]) -> None:
+        """Handle a node battery low notification."""
+        if not isinstance(
+            notification := event.get("notification"), BatteryNotification
+        ):
+            return
+        urgency = notification.urgency
+        self._trigger_event(urgency.name.lower(), {ATTR_URGENCY: urgency.value})
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self) -> None:
+        """Call when entity is added."""
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{DOMAIN}_{self._base_unique_id}_remove_entity",
+                self.async_remove,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{DOMAIN}_{self._base_unique_id}_remove_entity_on_interview_started",
+                self.async_remove,
+            )
+        )
+        self.async_on_remove(
+            self.node.on("notification", self._async_handle_notification)
+        )
+
+
 DISCOVERY_SCHEMAS: list[NewZWaveDiscoverySchema] = [
     NewZWaveDiscoverySchema(
         platform=Platform.EVENT,

homeassistant/components/zwave_js/strings.json

@@ -228,6 +228,19 @@
         "name": "Ping"
       }
     },
+    "event": {
+      "battery_low": {
+        "name": "Battery low",
+        "state_attributes": {
+          "event_type": {
+            "state": {
+              "now": "Battery should be replaced now",
+              "soon": "Battery should be replaced soon"
+            }
+          }
+        }
+      }
+    },
     "sensor": {
       "avg_signal_noise": {
         "name": "Avg. signal noise (channel {channel})"

tests/components/zwave_js/test_event.py

@@ -1,17 +1,30 @@
 """Test the Z-Wave JS event platform."""
 
 from datetime import timedelta
+from unittest.mock import MagicMock
 
 from freezegun import freeze_time
 import pytest
 from zwave_js_server.event import Event
+from zwave_js_server.model.node import Node
 
-from homeassistant.const import STATE_UNKNOWN, Platform
+from homeassistant.components.event import ATTR_EVENT_TYPE
+from homeassistant.components.zwave_js.const import ATTR_URGENCY
+from homeassistant.const import (
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+    EntityCategory,
+    Platform,
+)
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
 from homeassistant.util import dt as dt_util
 
+from tests.common import MockConfigEntry
+
 BASIC_EVENT_VALUE_ENTITY = "event.honeywell_in_wall_smart_fan_control_event_value"
 CENTRAL_SCENE_ENTITY = "event.node_51_scene_002"
+BATTERY_LOW_EVENT_ENTITY = "event.keypad_v2_battery_low"
 
 
 @pytest.fixture
@@ -204,3 +217,131 @@ async def test_central_scene(
         ],
         "value": 1,
     }
+
+
+def _battery_notification_event(node_id: int, urgency: int) -> Event:
+    """Build a Battery CC notification event."""
+    return Event(
+        type="notification",
+        data={
+            "source": "node",
+            "event": "notification",
+            "nodeId": node_id,
+            "endpointIndex": 0,
+            "ccId": 128,  # Battery CC
+            "args": {
+                "eventType": "battery low",
+                "urgency": urgency,
+            },
+        },
+    )
+
+
+async def test_battery_low_event(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    client: MagicMock,
+    ring_keypad: Node,
+    integration: MockConfigEntry,
+) -> None:
+    """Test the Battery CC battery low event entity."""
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes["event_types"] == ["soon", "now"]
+
+    entity_entry = entity_registry.async_get(BATTERY_LOW_EVENT_ENTITY)
+    assert entity_entry
+    assert entity_entry.entity_category is EntityCategory.DIAGNOSTIC
+    assert entity_entry.translation_key == "battery_low"
+    assert entity_entry.unique_id.endswith(".battery_low")
+
+    fut = dt_util.now() + timedelta(minutes=1)
+    with freeze_time(fut):
+        ring_keypad.receive_event(
+            _battery_notification_event(ring_keypad.node_id, urgency=1)
+        )
+        await hass.async_block_till_done()
+
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == dt_util.as_utc(fut).isoformat(timespec="milliseconds")
+    assert state.attributes[ATTR_EVENT_TYPE] == "soon"
+    assert state.attributes[ATTR_URGENCY] == 1
+
+    fut2 = fut + timedelta(hours=2)
+    with freeze_time(fut2):
+        ring_keypad.receive_event(
+            _battery_notification_event(ring_keypad.node_id, urgency=2)
+        )
+        await hass.async_block_till_done()
+
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == dt_util.as_utc(fut2).isoformat(timespec="milliseconds")
+    assert state.attributes[ATTR_EVENT_TYPE] == "now"
+    assert state.attributes[ATTR_URGENCY] == 2
+
+
+async def test_battery_low_event_other_notifications_ignored(
+    hass: HomeAssistant,
+    client: MagicMock,
+    ring_keypad: Node,
+    integration: MockConfigEntry,
+) -> None:
+    """Test that non-Battery CC notifications do not trigger the entity."""
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    # Notification CC (113) should be ignored by the battery low entity
+    other_event = Event(
+        type="notification",
+        data={
+            "source": "node",
+            "event": "notification",
+            "nodeId": ring_keypad.node_id,
+            "endpointIndex": 0,
+            "ccId": 113,
+            "args": {
+                "type": 7,
+                "label": "Home Security",
+                "event": 3,
+                "eventLabel": "Tampering, product cover removed",
+            },
+        },
+    )
+    ring_keypad.receive_event(other_event)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+
+async def test_battery_low_event_removed_on_interview_started(
+    hass: HomeAssistant,
+    client: MagicMock,
+    ring_keypad: Node,
+    integration: MockConfigEntry,
+) -> None:
+    """Test the entity is removed when the node starts a reinterview."""
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    ring_keypad.receive_event(
+        Event(
+            type="interview started",
+            data={
+                "source": "node",
+                "event": "interview started",
+                "nodeId": ring_keypad.node_id,
+            },
+        )
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(BATTERY_LOW_EVENT_ENTITY)
+    assert state
+    assert state.state == STATE_UNAVAILABLE