disable entities for events without notifications (#186)

* disable entities for events without notifications

* skip ISAPI requests for disabled entities

* Update disable state in entity registry according to Notify Surveillance Center flag

* update readme

README.md

@@ -28,7 +28,7 @@ The Home Assistant integration for Hikvision NVRs and IP cameras. Receives and s
 - NVR Input Triggers
 
 **NOTE**
-Events must be set to alert the surveillance center in Linkage Action for HA to be notified. An attribute on the event switch shows if this is set.
+Events must be set to alert the surveillance center in Linkage Action for Home Assistant to be notified. Otherwise related binary sensors and switches will appear as disabled entities.
 
 ## Preview
 

custom_components/hikvision_next/__init__.py

@@ -13,6 +13,9 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import entity_registry as er
+from homeassistant.components.switch import ENTITY_ID_FORMAT as SWITCH_ENTITY_ID_FORMAT
+from homeassistant.components.binary_sensor import ENTITY_ID_FORMAT as BINARY_SENSOR_ENTITY_ID_FORMAT
 
 from .const import (
     ALARM_SERVER_PATH,
@@ -87,6 +90,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     if get_first_instance_unique_id(hass) == entry.unique_id:
         hass.http.register_view(EventNotificationsView(hass))
 
+    refresh_disabled_entities_in_registry(hass, isapi)
+
     return True
 
 
@@ -154,3 +159,24 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):
         )
 
     return True
+
+
+def refresh_disabled_entities_in_registry(hass: HomeAssistant, isapi: ISAPI):
+    """Set disable state according to Notify Surveillance Center flag."""
+
+    def update_entity(event, ENTITY_ID_FORMAT):
+        entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
+        entity = entity_registry.async_get(entity_id)
+        if entity.disabled != event.disabled:
+            disabled_by = er.RegistryEntryDisabler.INTEGRATION if event.disabled else None
+            entity_registry.async_update_entity(entity_id, disabled_by=disabled_by)
+
+    entity_registry = er.async_get(hass)
+    for camera in isapi.cameras:
+        for event in camera.events_info:
+            update_entity(event, SWITCH_ENTITY_ID_FORMAT)
+            update_entity(event, BINARY_SENSOR_ENTITY_ID_FORMAT)
+
+    for event in isapi.device_info.events_info:
+        update_entity(event, SWITCH_ENTITY_ID_FORMAT)
+        update_entity(event, BINARY_SENSOR_ENTITY_ID_FORMAT)

custom_components/hikvision_next/binary_sensor.py

@@ -47,3 +47,4 @@ def __init__(self, isapi, device_id: int, event: EventInfo) -> None:
             self._attr_translation_placeholders = {"io_port_id": event.io_port_id}
         self._attr_device_class = EVENTS[event.id]["device_class"]
         self._attr_device_info = isapi.hass_device_info(device_id)
+        self._attr_entity_registry_enabled_default = not event.disabled

custom_components/hikvision_next/coordinator.py

@@ -42,6 +42,8 @@ async def _async_update_data(self):
             # Get camera event status
             for camera in self.isapi.cameras:
                 for event in camera.events_info:
+                    if event.disabled:
+                        continue
                     try:
                         entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
                         data[entity_id] = await self.isapi.get_event_enabled_state(event)
@@ -50,6 +52,8 @@ async def _async_update_data(self):
 
             # Get NVR event status
             for event in self.isapi.device_info.events_info:
+                if event.disabled:
+                    continue
                 try:
                     entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
                     data[entity_id] = await self.isapi.get_event_enabled_state(event)

custom_components/hikvision_next/diagnostics.py

@@ -88,6 +88,7 @@ async def _async_get_diagnostics(
         "ContentMgmt/Storage",
         "Security/adminAccesses",
         "Event/triggers",
+        "Event/triggers/scenechangedetection-1",
         "Event/notification/httpHosts",
     ]
 

custom_components/hikvision_next/isapi.py

@@ -82,7 +82,7 @@ class EventInfo:
     io_port_id: int
     unique_id: str
     url: str
-    notifiers: list[str] = field(default_factory=list)
+    disabled: bool = False
 
 
 @dataclass
@@ -361,7 +361,7 @@ async def get_device_event_capabilities(
         device_id: int,
         connection_type: str = CONNECTION_TYPE_DIRECT,
     ) -> list[EventInfo]:
-        """Get events info supported by integration (device id:  NVR = 0, camera > 0)."""
+        """Get events info handled by integration (device id:  NVR = 0, camera > 0)."""
         events = []
 
         if device_id == 0:  # NVR
@@ -386,31 +386,23 @@ async def get_device_event_capabilities(
                     io_port_id=event.io_port_id,
                     unique_id=unique_id,
                     url=self.get_event_url(event, connection_type),
-                    notifiers=event.notifications,
+                    disabled=("center" not in event.notifications), # Disable if not set Notify Surveillance Center
                 )
                 events.append(event_info)
         return events
 
     async def get_supported_events(self, system_capabilities: dict) -> list[SupportedEventsInfo]:
         """Get list of all supported events available."""
-        events = []
-        event_triggers = await self.request(GET, "Event/triggers")
-        event_notification = event_triggers.get("EventNotification")
-        if event_notification:
-            supported_events = deep_get(event_notification, "EventTriggerList.EventTrigger", [])
-        else:
-            supported_events = deep_get(event_triggers, "EventTriggerList.EventTrigger", [])
 
-        for support_event in supported_events:
-            notifications = support_event.get("EventTriggerNotificationList", {})
-            event_type = support_event.get("eventType")
-            # Fix for empty EventTriggerNotificationList in IP camera
-            if not notifications or not event_type:
-                continue
+        def get_event(event_trigger: dict):
+            notification_list = event_trigger.get("EventTriggerNotificationList", {}) or {}
+            event_type = event_trigger.get("eventType")
+            if not event_type:
+                return None
 
-            channel = support_event.get("videoInputChannelID", support_event.get("dynVideoInputChannelID", 0))
-            io_port = support_event.get("inputIOPortID", support_event.get("dynInputIOPortID", 0))
-            notifications = notifications.get("EventTriggerNotification", [])
+            channel = event_trigger.get("videoInputChannelID", event_trigger.get("dynVideoInputChannelID", 0))
+            io_port = event_trigger.get("inputIOPortID", event_trigger.get("dynInputIOPortID", 0))
+            notifications = notification_list.get("EventTriggerNotification", [])
 
             if not isinstance(notifications, list):
                 notifications = [notifications]
@@ -419,28 +411,33 @@ async def get_supported_events(self, system_capabilities: dict) -> list[Supporte
             if event_type.lower() in EVENTS_ALTERNATE_ID:
                 event_type = EVENTS_ALTERNATE_ID[event_type.lower()]
 
-            events.append(
-                SupportedEventsInfo(
-                    channel_id=int(channel),
-                    io_port_id=int(io_port),
-                    event_id=event_type.lower(),
-                    notifications=[notify.get("notificationMethod") for notify in notifications]
-                    if notifications
-                    else [],
-                )
+            return SupportedEventsInfo(
+                channel_id=int(channel),
+                io_port_id=int(io_port),
+                event_id=event_type.lower(),
+                notifications=[notify.get("notificationMethod") for notify in notifications] if notifications else [],
             )
 
-        # some single cameras do not have scenechangedetection in Event/triggers
-        if not self.device_info.is_nvr and not [e for e in events if e.event_id == "scenechangedetection"]:
+        events = []
+        event_triggers = await self.request(GET, "Event/triggers")
+        event_notification = event_triggers.get("EventNotification")
+        if event_notification:
+            supported_events = deep_get(event_notification, "EventTriggerList.EventTrigger", [])
+        else:
+            supported_events = deep_get(event_triggers, "EventTriggerList.EventTrigger", [])
+
+        for event_trigger in supported_events:
+            if event := get_event(event_trigger):
+                events.append(event)
+
+        # some devices do not have scenechangedetection in Event/triggers
+        if not [e for e in events if e.event_id == "scenechangedetection"]:
             is_supported = str_to_bool(deep_get(system_capabilities, "SmartCap.isSupportSceneChangeDetection", False))
             if is_supported:
-                events.append(
-                    SupportedEventsInfo(
-                        channel_id=1,
-                        io_port_id=0,
-                        event_id="scenechangedetection",
-                    )
-                )
+                event_trigger = await self.request(GET, "Event/triggers/scenechangedetection-1")
+                event_trigger = deep_get(event_trigger, "EventTrigger", {})
+                if event := get_event(event_trigger):
+                    events.append(event)
 
         return events
 

custom_components/hikvision_next/switch.py

@@ -66,6 +66,7 @@ def __init__(self, device_id: int, event: EventInfo, coordinator) -> None:
         self._attr_translation_key = event.id
         if event.id == EVENT_IO:
             self._attr_translation_placeholders = {"io_port_id": event.io_port_id}
+        self._attr_entity_registry_enabled_default = not event.disabled
         self.device_id = device_id
         self.event = event
 
@@ -92,14 +93,6 @@ async def async_turn_off(self, **kwargs: Any) -> None:
         finally:
             await self.coordinator.async_request_refresh()
 
-    @property
-    def extra_state_attributes(self):
-        """Return extra attributes."""
-        attrs = {}
-        attrs["notify_HA"] = "center" in self.event.notifiers
-        return attrs
-
-
 class NVROutputSwitch(CoordinatorEntity, SwitchEntity):
     """Detection events switch."""
 
@@ -151,9 +144,7 @@ class HolidaySwitch(CoordinatorEntity, SwitchEntity):
     def __init__(self, coordinator) -> None:
         """Initialize."""
         super().__init__(coordinator)
-        self._attr_unique_id = (
-            f"{slugify(coordinator.isapi.device_info.serial_no.lower())}_{HOLIDAY_MODE}"
-        )
+        self._attr_unique_id = f"{slugify(coordinator.isapi.device_info.serial_no.lower())}_{HOLIDAY_MODE}"
         self.entity_id = ENTITY_ID_FORMAT.format(self.unique_id)
         self._attr_device_info = coordinator.isapi.hass_device_info()
 

tests/fixtures/devices/DS-2CD2386G2-IU.json

@@ -1634,6 +1634,34 @@
             "sensitivityLevel": "50"
           }
         }
+      },
+      "Event/triggers/scenechangedetection-1": {
+        "response": {
+          "EventTrigger": {
+            "@version": "2.0",
+            "@xmlns": "http://www.hikvision.com/ver20/XMLSchema",
+            "id": "scenechangedetection-1",
+            "eventType": "scenechangedetection",
+            "eventDescription": "scenechangedetection Event trigger Information",
+            "videoInputChannelID": "1",
+            "dynVideoInputChannelID": "1",
+            "EventTriggerNotificationList": {
+              "EventTriggerNotification": [
+                {
+                  "id": "record-1",
+                  "notificationMethod": "record",
+                  "notificationRecurrence": "beginning",
+                  "videoInputID": "1"
+                },
+                {
+                  "id": "center",
+                  "notificationMethod": "center",
+                  "notificationRecurrence": "beginning"
+                }
+              ]
+            }
+          }
+        }
       }
     }
   }