Replace zwave_js_notification event listener with alarm_type/alarm_level state subscriptions

Reworks last_action/last_user derivation to subscribe to HA state changes on
sensor.{lock_object_id}_alarm_type instead of the raw zwave_js_notification bus event.
Replaces EVENT_MAP with a typed ALARM_MAP (alarm_type int → AlarmMapping) and a
UserSource enum (SLOT/MANUAL/RADIO). Covers alarm types 18/19 (keypad), 21/22 (manual),
24/25 (radio). For SLOT entries, alarm_level is read from the sibling sensor at event time.

custom_components/zlock_slots/const.py

@@ -1,3 +1,8 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
 DOMAIN = "zlock_slots"
 
 STORAGE_KEY = "zlock_slots_names"
@@ -5,19 +10,26 @@ STORAGE_VERSION = 1
 
 CONF_LOCK_ENTITY_IDS = "lock_entity_ids"
 
-ZWAVE_JS_NOTIFICATION_EVENT = "zwave_js_notification"
-
-LOCK_OP_TYPE = 6
-
-# Maps Z-Wave lock operation event number → (last_action, last_user_template)
-# last_user_template is "keypad" (resolve slot), "<Manual>", or "<Radio>"
-EVENT_MAP: dict[int, tuple[str, str]] = {
-  1: ("unlocked", "<Radio>"),
-  2: ("locked",   "<Radio>"),
-  3: ("unlocked", "keypad"),
-  4: ("locked",   "keypad"),
-  5: ("unlocked", "<Manual>"),
-  6: ("locked",   "<Manual>"),
-  7: ("unlocked", "<Radio>"),
-  8: ("locked",   "<Radio>"),
+
+class UserSource(Enum):
+  MANUAL = "manual"  # last_user = "<Manual>"; physical operation, alarm_level ignored
+  RADIO  = "radio"   # last_user = "<Radio>";  HA/Z-Wave command, alarm_level ignored
+  SLOT   = "slot"    # last_user = slot name or number from alarm_level
+
+
+@dataclass(frozen=True)
+class AlarmMapping:
+  action: str        # "locked" | "unlocked"
+  user_source: UserSource
+
+
+# Maps alarm_type value → lock action and how to derive last_user.
+# alarm_level meaning varies by alarm_type; only SLOT entries use it.
+ALARM_MAP: dict[int, AlarmMapping] = {
+  18: AlarmMapping("locked",   UserSource.SLOT),
+  19: AlarmMapping("unlocked", UserSource.SLOT),
+  21: AlarmMapping("locked",   UserSource.MANUAL),
+  22: AlarmMapping("unlocked", UserSource.MANUAL),
+  24: AlarmMapping("locked",   UserSource.RADIO),
+  25: AlarmMapping("unlocked", UserSource.RADIO),
 }

custom_components/zlock_slots/coordinator.py

@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
 from homeassistant.core import HomeAssistant, Event, callback
-from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.event import async_track_state_change_event
 from homeassistant.helpers.storage import Store
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
@@ -14,9 +14,8 @@ from .const import (
   STORAGE_KEY,
   STORAGE_VERSION,
   CONF_LOCK_ENTITY_IDS,
-  ZWAVE_JS_NOTIFICATION_EVENT,
-  LOCK_OP_TYPE,
-  EVENT_MAP,
+  ALARM_MAP,
+  UserSource,
 )
 
 if TYPE_CHECKING:
@@ -38,6 +37,15 @@ class LockState:
   last_user: str | None = None
 
 
+def _alarm_entity(lock_entity_id: str, suffix: str) -> str:
+  """Derive alarm sensor entity ID from lock entity ID.
+
+  lock.living_room_front_lock -> sensor.living_room_front_lock_alarm_<suffix>
+  """
+  object_id = lock_entity_id.split(".")[-1]
+  return f"sensor.{object_id}_alarm_{suffix}"
+
+
 class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
   """Manages slot data, name storage, and event listening for all configured locks."""
 
@@ -47,7 +55,10 @@ class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
     self._lock_entity_ids: list[str] = entry.data[CONF_LOCK_ENTITY_IDS]
     self._store: Store = Store(hass, STORAGE_VERSION, STORAGE_KEY)
     self._names: dict[str, str] = {}  # "{entity_id}:{slot_num}" → name
-    self._device_to_entity: dict[str, str] = {}  # device_id → lock_entity_id
+    # alarm_type entity_id → lock_entity_id
+    self._alarm_to_lock: dict[str, str] = {
+      _alarm_entity(eid, "type"): eid for eid in self._lock_entity_ids
+    }
     self._unsub_event: callback | None = None
     self.data: dict[str, LockState] = {eid: LockState() for eid in self._lock_entity_ids}
 
@@ -56,11 +67,12 @@ class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
     if stored:
       self._names = stored.get("names", {})
 
-    self._build_device_map()
     await self._async_fetch_all_slots()
 
-    self._unsub_event = self.hass.bus.async_listen(
-      ZWAVE_JS_NOTIFICATION_EVENT, self._handle_zwave_event
+    self._unsub_event = async_track_state_change_event(
+      self.hass,
+      list(self._alarm_to_lock),
+      self._handle_alarm_state_change,
     )
 
   def async_teardown(self) -> None:
@@ -68,13 +80,6 @@ class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
       self._unsub_event()
       self._unsub_event = None
 
-  def _build_device_map(self) -> None:
-    er_obj = er.async_get(self.hass)
-    for lock_entity_id in self._lock_entity_ids:
-      entry = er_obj.async_get(lock_entity_id)
-      if entry and entry.device_id:
-        self._device_to_entity[entry.device_id] = lock_entity_id
-
   async def _async_fetch_all_slots(self) -> None:
     for lock_entity_id in self._lock_entity_ids:
       slots = await self._fetch_slots(lock_entity_id)
@@ -104,33 +109,40 @@ class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
     return slots
 
   @callback
-  def _handle_zwave_event(self, event: Event) -> None:
-    data = event.data
-    if data.get("type") != LOCK_OP_TYPE:
+  def _handle_alarm_state_change(self, event: Event) -> None:
+    entity_id: str = event.data["entity_id"]
+    lock_entity_id = self._alarm_to_lock.get(entity_id)
+    if not lock_entity_id:
       return
 
-    device_id = data.get("device_id")
-    lock_entity_id = self._device_to_entity.get(device_id)
-    if not lock_entity_id:
+    new_state = event.data.get("new_state")
+    if new_state is None:
+      return
+
+    try:
+      alarm_type = int(new_state.state)
+    except (ValueError, TypeError):
       return
 
-    event_num = data.get("event")
-    mapping = EVENT_MAP.get(event_num)
+    mapping = ALARM_MAP.get(alarm_type)
     if not mapping:
       return
 
-    action, user_template = mapping
-    if user_template == "keypad":
-      params = data.get("parameters", {})
-      user_id = params.get("userId")
-      if user_id is not None:
-        user_str = self.get_name(lock_entity_id, int(user_id)) or str(user_id)
-      else:
-        user_str = "<Keypad>"
+    if mapping.user_source == UserSource.SLOT:
+      alarm_level_state = self.hass.states.get(_alarm_entity(lock_entity_id, "level"))
+      try:
+        slot_num = int(alarm_level_state.state)
+        user_str = self.get_name(lock_entity_id, slot_num) or str(slot_num)
+      except (AttributeError, ValueError, TypeError):
+        user_str = "<Unknown>"
+    elif mapping.user_source == UserSource.MANUAL:
+      user_str = "<Manual>"
+    elif mapping.user_source == UserSource.RADIO:
+      user_str = "<Radio>"
     else:
-      user_str = user_template
+      user_str = "<Unknown>"
 
-    self.data[lock_entity_id].last_action = action
+    self.data[lock_entity_id].last_action = mapping.action
     self.data[lock_entity_id].last_user = user_str
     self.async_update_listeners()
 
@@ -145,7 +157,6 @@ class ZLockCoordinator(DataUpdateCoordinator[dict[str, LockState]]):
       self._names.pop(key, None)
     await self._store.async_save({"names": self._names})
 
-    # Refresh the name on the cached slot if it exists
     state = self.data.get(lock_entity_id)
     if state and slot_num in state.slots:
       state.slots[slot_num].name = name or None

custom_components/zlock_slots/sensor.py

@@ -80,6 +80,10 @@ class ZLockLastUserSensor(_ZLockBaseSensor):
     return state.last_user if state else None
 
 
+# TODO: ZLockSlotsSensor exists only to carry the slots list as extra_state_attributes
+# for the Lovelace card. It should be moved to a more appropriate entity once one exists
+# (e.g. a lock-state or config sensor). Keeping it as a standalone sensor causes HA to
+# track slot-list diffs in history/stats, which we don't want.
 class ZLockSlotsSensor(_ZLockBaseSensor):
   _attr_translation_key = "slots"