Improve typing of deCONZ events (#69478)

homeassistant/components/deconz/deconz_event.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+from typing import Any
+
+from pydeconz.models.event import EventType
 from pydeconz.sensor import (
     ANCILLARY_CONTROL_EMERGENCY,
     ANCILLARY_CONTROL_FIRE,
@@ -20,7 +23,6 @@ from homeassistant.const import (
 )
 from homeassistant.core import callback
 from homeassistant.helpers import device_registry as dr
-from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.util import slugify
 
 from .const import CONF_ANGLE, CONF_GESTURE, LOGGER
@@ -42,43 +44,42 @@ async def async_setup_events(gateway: DeconzGateway) -> None:
     """Set up the deCONZ events."""
 
     @callback
-    def async_add_sensor(
+    def async_add_sensor(_, sensor_id: str) -> None:
-        sensors: AncillaryControl | Switch | None = None,
-    ) -> None:
         """Create DeconzEvent."""
-        new_events = []
+        new_event: DeconzAlarmEvent | DeconzEvent
-        known_events = {event.unique_id for event in gateway.events}
+        sensor = gateway.api.sensors[sensor_id]
 
-        if sensors is None:
+        if not gateway.option_allow_clip_sensor and sensor.type.startswith("CLIP"):
-            sensors = gateway.api.sensors.values()
+            return None
 
-        for sensor in sensors:
+        if isinstance(sensor, Switch):
+            new_event = DeconzEvent(sensor, gateway)
 
-            if not gateway.option_allow_clip_sensor and sensor.type.startswith("CLIP"):
+        elif isinstance(sensor, AncillaryControl):
-                continue
+            new_event = DeconzAlarmEvent(sensor, gateway)
 
-            if sensor.unique_id in known_events:
+        else:
-                continue
+            return None
 
-            if isinstance(sensor, Switch):
+        gateway.hass.async_create_task(new_event.async_update_device_registry())
-                new_events.append(DeconzEvent(sensor, gateway))
+        gateway.events.append(new_event)
-
-            elif isinstance(sensor, AncillaryControl):
-                new_events.append(DeconzAlarmEvent(sensor, gateway))
-
-        for new_event in new_events:
-            gateway.hass.async_create_task(new_event.async_update_device_registry())
-            gateway.events.append(new_event)
 
     gateway.config_entry.async_on_unload(
-        async_dispatcher_connect(
+        gateway.api.sensors.ancillary_control.subscribe(
-            gateway.hass,
-            gateway.signal_new_sensor,
             async_add_sensor,
+            EventType.ADDED,
         )
     )
 
-    async_add_sensor()
+    gateway.config_entry.async_on_unload(
+        gateway.api.sensors.switch.subscribe(
+            async_add_sensor,
+            EventType.ADDED,
+        )
+    )
+
+    for sensor_id in gateway.api.sensors:
+        async_add_sensor(EventType.ADDED, sensor_id)
 
 
 @callback
@@ -90,7 +91,7 @@ def async_unload_events(gateway: DeconzGateway) -> None:
     gateway.events.clear()
 
 
-class DeconzEvent(DeconzBase):
+class DeconzEventBase(DeconzBase):
     """When you want signals instead of entities.
 
     Stateless sensors such as remotes are expected to generate an event
@@ -105,21 +106,44 @@ class DeconzEvent(DeconzBase):
         """Register callback that will be used for signals."""
         super().__init__(device, gateway)
 
-        self._device.register_callback(self.async_update_callback)
+        self._unsubscribe = device.subscribe(self.async_update_callback)
 
+        self.device = device
         self.device_id: str | None = None
         self.event_id = slugify(self._device.name)
         LOGGER.debug("deCONZ event created: %s", self.event_id)
 
-    @property
-    def device(self) -> AncillaryControl | Switch:
-        """Return Event device."""
-        return self._device
-
     @callback
     def async_will_remove_from_hass(self) -> None:
         """Disconnect event object when removed."""
-        self._device.remove_callback(self.async_update_callback)
+        self._unsubscribe()
+
+    @callback
+    def async_update_callback(self) -> None:
+        """Fire the event if reason is that state is updated."""
+        raise NotImplementedError
+
+    async def async_update_device_registry(self) -> None:
+        """Update device registry."""
+        if not self.device_info:
+            return
+
+        device_registry = dr.async_get(self.gateway.hass)
+
+        entry = device_registry.async_get_or_create(
+            config_entry_id=self.gateway.config_entry.entry_id, **self.device_info
+        )
+        self.device_id = entry.id
+
+
+class DeconzEvent(DeconzEventBase):
+    """When you want signals instead of entities.
+
+    Stateless sensors such as remotes are expected to generate an event
+    instead of a sensor entity in hass.
+    """
+
+    _device: Switch
 
     @callback
     def async_update_callback(self) -> None:
@@ -130,7 +154,7 @@ class DeconzEvent(DeconzBase):
         ):
             return
 
-        data = {
+        data: dict[str, Any] = {
             CONF_ID: self.event_id,
             CONF_UNIQUE_ID: self.serial,
             CONF_EVENT: self._device.button_event,
@@ -150,20 +174,8 @@ class DeconzEvent(DeconzBase):
 
         self.gateway.hass.bus.async_fire(CONF_DECONZ_EVENT, data)
 
-    async def async_update_device_registry(self) -> None:
-        """Update device registry."""
-        if not self.device_info:
-            return
 
-        device_registry = dr.async_get(self.gateway.hass)
+class DeconzAlarmEvent(DeconzEventBase):
-
-        entry = device_registry.async_get_or_create(
-            config_entry_id=self.gateway.config_entry.entry_id, **self.device_info
-        )
-        self.device_id = entry.id
-
-
-class DeconzAlarmEvent(DeconzEvent):
     """Alarm control panel companion event when user interacts with a keypad."""
 
     _device: AncillaryControl