IO events v2

custom_components/hikvision_next/binary_sensor.py

@@ -8,7 +8,7 @@
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import DATA_ISAPI, DOMAIN, EVENTS
-from .isapi import ISAPI, AnalogCamera, EventInfo, IPCamera
+from .isapi import EventInfo
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
@@ -22,39 +22,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e
     # Camera Events
     for camera in isapi.cameras:
         for event in camera.supported_events:
-            entities.append(CameraEventBinarySensor(isapi, camera, event))
+            entities.append(EventBinarySensor(isapi, camera.id, event))
 
     # NVR Events
     if isapi.device_info.is_nvr:
-        for io_event in isapi.device_info.supported_events:
-            entities.append(NVREventBinarySensor(isapi, io_event))
+        for event in isapi.device_info.supported_events:
+            entities.append(EventBinarySensor(isapi, 0, event))
 
     async_add_entities(entities)
 
 
-class CameraEventBinarySensor(BinarySensorEntity):
+class EventBinarySensor(BinarySensorEntity):
     """Event detection sensor."""
 
     _attr_has_entity_name = True
     _attr_is_on = False
 
-    def __init__(self, isapi, camera: AnalogCamera | IPCamera, event: EventInfo) -> None:
+    def __init__(self, isapi, device_id: int, event: EventInfo) -> None:
         self.entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
         self._attr_unique_id = self.entity_id
-        self._attr_name = EVENTS[event.id]["label"]
+        self._attr_name = f"{EVENTS[event.id]['label']}{' ' + str(event.io_port_id) if event.io_port_id != 0 else ''}"
         self._attr_device_class = EVENTS[event.id]["device_class"]
-        self._attr_device_info = isapi.get_device_info(camera.id)
-
-
-class NVREventBinarySensor(BinarySensorEntity):
-    """IO Event detection sensor."""
-
-    _attr_has_entity_name = True
-    _attr_is_on = False
-
-    def __init__(self, isapi: ISAPI, event: EventInfo) -> None:
-        self.entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
-        self._attr_unique_id = self.entity_id
-        self._attr_name = f"{EVENTS[event.id]['label']} {event.channel_id}"
-        self._attr_device_class = EVENTS[event.id]["device_class"]
-        self._attr_device_info = isapi.get_device_info(0)
+        self._attr_device_info = isapi.get_device_info(device_id)

custom_components/hikvision_next/const.py

@@ -19,13 +19,12 @@
 HOLIDAY_MODE_SWITCH_LABEL = "Holiday mode"
 ALARM_SERVER_SENSOR_LABEL_FORMAT = "Alarm Server {}"
 
-DEVICE_TYPE_IP_CAMERA = "IPCamera"
-DEVICE_TYPE_ANALOG_CAMERA = "AnalogCamera"
-DEVICE_TYPE_NVR = "NVR"
+CONNECTION_TYPE_DIRECT = "Direct"
+CONNECTION_TYPE_PROXIED = "Proxied"
 
 HIKVISION_EVENT = f"{DOMAIN}_event"
 EVENT_BASIC: Final = "basic"
-EVENT_NVR_BASIC: Final = "nvr_basic"
+EVENT_IO: Final = "io"
 EVENT_SMART: Final = "smart"
 EVENTS = {
     "motiondetection": {
@@ -81,11 +80,11 @@
         "device_class": BinarySensorDeviceClass.MOTION,
     },
     "io": {
-        "type": EVENT_NVR_BASIC,
+        "type": EVENT_IO,
         "label": "Alarm Input",
-        "channel_attr": "inputIOPortID",
-        "url_path": "IO/inputs",
-        "slug": "IOInputPort",
+        "slug": "inputs",
+        "direct_node": "IOInputPort",
+        "proxied_node": "IOProxyInputPort",
         "device_class": BinarySensorDeviceClass.MOTION
     }
 }

custom_components/hikvision_next/coordinator.py

@@ -57,16 +57,15 @@ async def _async_update_data(self):
                 except Exception as ex:  # pylint: disable=broad-except
                     self.isapi.handle_exception(ex, f"Cannot fetch state for {event.id}")
 
-            # Get NVR outputs status
-            if self.isapi.device_info.is_nvr:
-                for i in range(1, self.isapi.device_info.output_ports + 1):
-                    try:
-                        entity_id = ENTITY_ID_FORMAT.format(
-                            f"{slugify(self.isapi.device_info.serial_no.lower())}_{i}_alarm_output"
-                        )
-                        data[entity_id] = await self.isapi.get_port_status("output", i)
-                    except Exception as ex:  # pylint: disable=broad-except
-                        self.isapi.handle_exception(ex, f"Cannot fetch state for {event.id}")
+            # Get output port(s) status
+            for i in range(1, self.isapi.device_info.output_ports + 1):
+                try:
+                    entity_id = ENTITY_ID_FORMAT.format(
+                        f"{slugify(self.isapi.device_info.serial_no.lower())}_{i}_alarm_output"
+                    )
+                    data[entity_id] = await self.isapi.get_port_status("output", i)
+                except Exception as ex:  # pylint: disable=broad-except
+                    self.isapi.handle_exception(ex, f"Cannot fetch state for {event.id}")
 
             # Refresh HDD data
             try:

custom_components/hikvision_next/diagnostics.py

@@ -63,44 +63,36 @@ async def _async_get_diagnostics(
     info.update({"Entity Data": to_json(coordinator.data)})
 
     # Add raw device info
-    info.update(await get_isapi_data("RAW Device Info", isapi.isapi.System.deviceInfo, "DeviceInfo"))
-
-    # Add raw camera info
-    info.update(
-        await get_isapi_data(
-            "RAW Analog Camera Info",
-            isapi.isapi.System.Video.inputs.channels,
-            "VideoInputChannelList",
-        )
-    )
-    info.update(
-        await get_isapi_data(
-            "RAW IP Camera Info",
-            isapi.isapi.ContentMgmt.InputProxy.channels,
-            "InputProxyChannelList",
-        )
-    )
+    info.update(await get_isapi_data("RAW Device Info", isapi.isapi.System.deviceInfo))
+
+    # Add raw camera info - Direct connected
+    info.update(await get_isapi_data("RAW Analog Camera Info", isapi.isapi.System.Video.inputs.channels))
+
+    # Add raw camera info - Proxy connected
+    info.update(await get_isapi_data("RAW IP Camera Info", isapi.isapi.ContentMgmt.InputProxy.channels))
 
     # Add raw capabilities
-    info.update(await get_isapi_data("RAW Capabilities Info", isapi.isapi.System.capabilities, "DeviceCap"))
+    info.update(await get_isapi_data("RAW Capabilities Info", isapi.isapi.System.capabilities))
 
     # Add raw supported events
-    info.update(await get_isapi_data("RAW Events Info", isapi.isapi.Event.triggers, ""))
+    info.update(await get_isapi_data("RAW Events Info", isapi.isapi.Event.triggers))
+
+    # Add IO info - direct connected
+    info.update(await get_isapi_data("Direct IO Inputs", isapi.isapi.System.IO.inputs))
+    info.update(await get_isapi_data("Direct IO Outputs", isapi.isapi.System.IO.outputs))
+
+    # Add IO info - proxy connected
+    info.update(await get_isapi_data("Proxied IO Inputs", isapi.isapi.ContentMgmt.IOProxy.inputs))
+    info.update(await get_isapi_data("Proxied IO Outputs", isapi.isapi.ContentMgmt.IOProxy.outputs))
 
     # Add raw streams info
-    info.update(await get_isapi_data("RAW Streams Info", isapi.isapi.Streaming.channels, "StreamingChannelList"))
+    info.update(await get_isapi_data("RAW Streams Info", isapi.isapi.Streaming.channels))
 
     # Add raw holiday info
-    info.update(await get_isapi_data("RAW Holiday Info", isapi.isapi.System.Holidays, "HolidayList"))
+    info.update(await get_isapi_data("RAW Holiday Info", isapi.isapi.System.Holidays))
 
     # Add alarms server info
-    info.update(
-        await get_isapi_data(
-            "RAW Alarm Server Info",
-            isapi.isapi.Event.notification.httpHosts,
-            "HttpHostNotificationList",
-        )
-    )
+    info.update(await get_isapi_data("RAW Alarm Server Info", isapi.isapi.Event.notification.httpHosts))
 
     return info