Fix event filter (#131)

custom_components/lock_code_manager/backports.py

@@ -0,0 +1,20 @@
+"""Backports for lock_code_manager."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from homeassistant.const import MAJOR_VERSION, MINOR_VERSION
+from homeassistant.core import Event
+
+_LOGGER = logging.getLogger(__name__)
+
+EVENT_DATA_PASSED_IN = MAJOR_VERSION > 2024 or (
+    MAJOR_VERSION == 2024 and MINOR_VERSION >= 4
+)
+
+
+def get_event_data_for_filter(evt: Event | dict[str, Any]) -> dict[str, Any]:
+    """Get event data for filter."""
+    return evt if EVENT_DATA_PASSED_IN else evt.data  # type: ignore[union-attr]

custom_components/lock_code_manager/entity.py

@@ -24,6 +24,7 @@
 from homeassistant.helpers.entity import DeviceInfo, Entity, EntityCategory
 from homeassistant.helpers.event import async_track_state_change
 
+from .backports import get_event_data_for_filter
 from .const import (
     ATTR_CODE_SLOT,
     ATTR_ENTITIES_ADDED_TRACKER,
@@ -234,14 +235,13 @@ def _listen_for_initial_state(
             self._unsub_initial_state = None
 
     @callback
-    def _event_filter(self, event: Event) -> bool:
+    def _event_filter(self, event: Event | dict[str, Any]) -> bool:
         """Filter events."""
+        data = get_event_data_for_filter(event)
         return (
-            any(
-                event.data[ATTR_ENTITY_ID] == lock.lock.entity_id for lock in self.locks
-            )
-            and event.data[ATTR_CODE_SLOT] == int(self.slot_num)
-            and event.data[ATTR_TO] == STATE_UNLOCKED
+            any(data[ATTR_ENTITY_ID] == lock.lock.entity_id for lock in self.locks)
+            and data[ATTR_CODE_SLOT] == int(self.slot_num)
+            and data[ATTR_TO] == STATE_UNLOCKED
         )
 
     async def async_will_remove_from_hass(self) -> None:

custom_components/lock_code_manager/providers/zwave_js.py

@@ -36,21 +36,16 @@
     ATTR_ENTITY_ID,
     CONF_ENABLED,
     CONF_PIN,
-    MAJOR_VERSION,
-    MINOR_VERSION,
     STATE_ON,
 )
 from homeassistant.core import Event, callback
 
+from ..backports import get_event_data_for_filter
 from ..const import CONF_LOCKS, CONF_SLOTS, DOMAIN
 from ..data import get_entry_data
 from ..exceptions import LockDisconnected
 from ._base import BaseLock
 
-EVENT_DATA_PASSED_IN = MAJOR_VERSION > 2024 or (
-    MAJOR_VERSION == 2024 and MINOR_VERSION >= 4
-)
-
 _LOGGER = logging.getLogger(__name__)
 
 # All known Access Control Notification CC events that indicate the lock is locked
@@ -88,11 +83,11 @@ def node(self) -> Node:
         )
 
     @callback
-    def _zwave_js_event_filter(self, evt: Event | dict[str, Any]) -> bool:
+    def _zwave_js_event_filter(self, event: Event | dict[str, Any]) -> bool:
         """Filter out events."""
         # Try to find the lock that we are getting an event for, skipping
         # ones that don't match
-        data: dict[str, Any] = evt if EVENT_DATA_PASSED_IN else evt.data  # type: ignore[union-attr]
+        data = get_event_data_for_filter(event)
         assert self.node.client.driver
         return (
             data[ATTR_HOME_ID] == self.node.client.driver.controller.home_id