Fix/event swtich for multi channel camera (#272)

* added entities for each channel of multichannel cam

* added is_multi_channel flag to isapi client

* alarm output switch state fix

custom_components/hikvision_next/diagnostics.py

@@ -87,6 +87,7 @@ async def _async_get_diagnostics(
         "ContentMgmt/Storage",
         "Security/adminAccesses",
         "Event/triggers",
+        "Event/channels/capabilities",
         "Event/triggers/scenechangedetection-1",
         "Event/notification/httpHosts",
         "Streaming/channels",

custom_components/hikvision_next/isapi/isapi.py

@@ -101,8 +101,8 @@ async def get_hardware_info(self):
         await self.get_device_info()
         capabilities = (await self.request(GET, "System/capabilities")).get("DeviceCap", {})
 
-        self.capabilities.support_analog_cameras = int(deep_get(capabilities, "SysCap.VideoCap.videoInputPortNums", 0))
-        self.capabilities.support_digital_cameras = int(deep_get(capabilities, "RacmCap.inputProxyNums", 0))
+        self.capabilities.analog_cameras_inputs = int(deep_get(capabilities, "SysCap.VideoCap.videoInputPortNums", 0))
+        self.capabilities.digital_cameras_inputs = int(deep_get(capabilities, "RacmCap.inputProxyNums", 0))
         self.capabilities.support_holiday_mode = str_to_bool(deep_get(capabilities, "SysCap.isSupportHolidy", "false"))
         self.capabilities.support_channel_zero = str_to_bool(
             deep_get(capabilities, "RacmCap.isSupportZeroChan", "false")
@@ -116,7 +116,7 @@ async def get_hardware_info(self):
 
         # Set if NVR based on whether more than 1 supported IP or analog cameras
         # Single IP camera will show 0 supported devices in total
-        if self.capabilities.support_analog_cameras + self.capabilities.support_digital_cameras > 1:
+        if self.capabilities.analog_cameras_inputs + self.capabilities.digital_cameras_inputs > 1:
             self.device_info.is_nvr = True
 
         await self.get_cameras()
@@ -141,6 +141,7 @@ async def get_cameras(self):
                 channel_id = int(deep_get(streaming_channel, "Video.videoInputChannelID", 1))
                 channel_ids.add(channel_id)
 
+            self.capabilities.is_multi_channel = len(channel_ids) > 1
             for channel_id in sorted(channel_ids):
                 # Determine camera name
                 if len(channel_ids) > 1:
@@ -162,7 +163,7 @@ async def get_cameras(self):
                 self.cameras.append(camera)
         else:
             # Get analog and digital cameras attached to NVR
-            if self.capabilities.support_digital_cameras > 0:
+            if self.capabilities.digital_cameras_inputs > 0:
                 digital_cameras = deep_get(
                     (await self.request(GET, "ContentMgmt/InputProxy/channels")),
                     "InputProxyChannelList.InputProxyChannel",
@@ -196,7 +197,7 @@ async def get_cameras(self):
                     )
 
             # Get analog cameras
-            if self.capabilities.support_analog_cameras > 0:
+            if self.capabilities.analog_cameras_inputs > 0:
                 analog_cameras = deep_get(
                     (await self.request(GET, "System/Video/inputs/channels")),
                     "VideoInputChannelList.VideoInputChannel",
@@ -238,7 +239,7 @@ async def get_protocols(self):
     async def get_supported_events(self, system_capabilities: dict) -> list[EventInfo]:
         """Get list of all supported events available."""
 
-        def get_event(event_trigger: dict):
+        def create_event_info(event_trigger: dict):
             notification_list = event_trigger.get("EventTriggerNotificationList", {}) or {}
 
             event_type = event_trigger.get("eventType")
@@ -283,15 +284,17 @@ def get_event(event_trigger: dict):
             )
 
         events = []
+
+        # Get events from Event/triggers
         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", [])
+            available_events = deep_get(event_notification, "EventTriggerList.EventTrigger", [])
         else:
-            supported_events = deep_get(event_triggers, "EventTriggerList.EventTrigger", [])
+            available_events = deep_get(event_triggers, "EventTriggerList.EventTrigger", [])
 
-        for event_trigger in supported_events:
-            if event := get_event(event_trigger):
+        for event_trigger in available_events:
+            if event := create_event_info(event_trigger):
                 events.append(event)
 
         # some devices do not have scenechangedetection in Event/triggers
@@ -300,9 +303,28 @@ def get_event(event_trigger: dict):
             if is_supported:
                 event_trigger = await self.request(GET, "Event/triggers/scenechangedetection-1")
                 event_trigger = deep_get(event_trigger, "EventTrigger", {})
-                if event := get_event(event_trigger):
+                if event := create_event_info(event_trigger):
                     events.append(event)
 
+        # multichannel camera needs to fetch events for each channel
+        if self.capabilities.is_multi_channel:
+            channels_capabilities = await self.request(GET, "Event/channels/capabilities")
+            channel_events = deep_get(channels_capabilities, "ChannelEventCapList.ChannelEventCap", [])
+            for event_cap in channel_events:
+                event_types = deep_get(event_cap, "eventType").get("@opt", "").split(",")
+                channel_id = int(event_cap.get("channelID"))
+                for event_type in event_types:
+                    event_id = event_type.lower()
+                    if event_id in EVENTS_ALTERNATE_ID:
+                        event_id = EVENTS_ALTERNATE_ID[event_id]
+                    if event_id not in EVENTS:
+                        continue
+                    if not [e for e in events if (e.id == event_id and e.channel_id == channel_id)]:
+                        event_trigger = await self.request(GET, f"Event/triggers/{event_id}-{channel_id}")
+                        event_trigger = deep_get(event_trigger, "EventTrigger", {})
+                        if event := create_event_info(event_trigger):
+                            events.append(event)
+
         return events
 
     def get_event_url(self, event_id: str, channel_id: int, io_port_id: int, is_proxy: bool) -> str | None:

custom_components/hikvision_next/isapi/models.py

@@ -96,8 +96,9 @@ class ISAPIDeviceInfo:
 class CapabilitiesInfo:
     """Holds info of an NVR/DVR or single IP Camera."""
 
-    support_analog_cameras: int = 0
-    support_digital_cameras: int = 0
+    analog_cameras_inputs: int = 0  # number of analog cameras connected to NVR/DVR
+    digital_cameras_inputs: int = 0  # number of digital cameras connected to NVR/DVR
+    is_multi_channel: bool = False  # if camera has multiple channels
     support_holiday_mode: bool = False
     support_alarm_server: bool = False
     support_channel_zero: bool = False

custom_components/hikvision_next/switch.py

@@ -116,7 +116,7 @@ def __init__(self, coordinator, port_no: int) -> None:
     @property
     def is_on(self) -> bool | None:
         """Turn on."""
-        return self.coordinator.data.get(self.unique_id) == "active"
+        return self.coordinator.data.get(self.unique_id)
 
     async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn on."""