Add get_nvr_events and inject_events_into_camera

pyhik/__init__.py

@@ -0,0 +1,11 @@
+"""pyhik - Python library for Hikvision camera/NVR events."""
+
+from pyhik.hikvision import HikCamera, inject_events_into_camera
+from pyhik.constants import __version__, VALID_NOTIFICATION_METHODS
+
+__all__ = [
+    'HikCamera',
+    'inject_events_into_camera',
+    'VALID_NOTIFICATION_METHODS',
+    '__version__'
+]

pyhik/constants.py

@@ -67,3 +67,7 @@
 CONTEXT_TRIG = 'TRIGGERS'
 CONTEXT_ALERT = 'ALERTS'
 CONTEXT_MOTION = 'MOTION'
+
+# Notification methods that indicate the event is active/configured
+# Expanded from original list of just 'center' and 'HTTP' to support NVRs
+VALID_NOTIFICATION_METHODS = {'center', 'HTTP', 'record', 'email', 'beep'}

pyhik/hikvision.py

@@ -367,12 +367,22 @@ def get_device_info(self):
             _LOGGING.error('There was a problem: %s', err)
             return None
 
-    def get_event_triggers(self):
+    def get_event_triggers(self, notification_methods=None):
         """
         Returns dict of supported events.
         Key = Event Type
         List = Channels that have that event activated
+
+        Args:
+            notification_methods: Set of notification method strings to accept.
+                Defaults to {'center', 'HTTP'}. For NVRs, you may want to include
+                additional methods like 'record', 'email', 'beep'.
         """
+        if notification_methods is None:
+            notification_methods = {'center', 'HTTP'}
+        # Normalize to lowercase for comparison
+        notification_methods_lower = {m.lower() for m in notification_methods}
+
         events = {}
         nvrflag = False
         event_xml = []
@@ -445,11 +455,9 @@ def get_event_triggers(self):
                     for notifytrigger in etnotify:
                         ntype = notifytrigger.find(
                             self.element_query('notificationMethod', CONTEXT_TRIG))
-                        if ntype.text == 'center' or ntype.text == 'HTTP':
-                            """
-                            If we got this far we found an event that we want
-                            to track.
-                            """
+                        if ntype is not None and ntype.text and \
+                                ntype.text.lower() in notification_methods_lower:
+                            # Found an event with a valid notification method
                             # Catch events with bad IDs
                             if etchannel_num == 0 : etchannel_num = 1
                             events.setdefault(ettype.text, []) \
@@ -679,3 +687,41 @@ def update_attributes(self, event, channel, attr):
         except KeyError:
             _LOGGING.debug('Error updating attributes for: (%s, %s)',
                            event, channel)
+
+    def inject_events(self, events):
+        """Inject discovered events into the camera's event_states.
+
+        This allows the camera to track events that wouldn't normally be
+        detected, such as those from NVRs with non-standard notification
+        methods.
+
+        Args:
+            events: Dict mapping event type names to lists of channel numbers.
+        """
+        for event_name, channels in events.items():
+            for channel in channels:
+                # Only add if not already present
+                if event_name not in self.event_states:
+                    self.event_states[event_name] = []
+
+                # Check if this channel is already tracked
+                channel_exists = any(
+                    sensor[1] == channel for sensor in self.event_states[event_name]
+                )
+                if not channel_exists:
+                    # Add the event state: [is_active, channel, count, last_update_time]
+                    self.event_states[event_name].append(
+                        [False, channel, 0, datetime.datetime.now()]
+                    )
+
+
+def inject_events_into_camera(camera, events):
+    """Inject discovered events into the pyhik camera's event_states.
+
+    This allows the camera to track events that wouldn't normally be detected.
+
+    Args:
+        camera: A HikCamera instance.
+        events: Dict mapping event type names to lists of channel numbers.
+    """
+    camera.inject_events(events)

test/test_hikvision.py

@@ -5,8 +5,8 @@
 import unittest
 
 from unittest.mock import MagicMock, patch, PropertyMock
-from pyhik.hikvision import HikCamera
-from pyhik.constants import (CONNECT_TIMEOUT)
+from pyhik.hikvision import HikCamera, inject_events_into_camera
+from pyhik.constants import CONNECT_TIMEOUT, VALID_NOTIFICATION_METHODS
 
 XML = """<MotionDetection xmlns="http://www.hikvision.com/ver20/XMLSchema" version="2.0">
     <enabled>{}</enabled>
@@ -82,5 +82,235 @@ def change_get_response(url, data,timeout):
         self.assertFalse(device.current_motion_detection_state)
 
 
+# XML for testing get_event_triggers with various notification methods
+EVENT_TRIGGERS_XML = """<?xml version="1.0" encoding="UTF-8"?>
+<EventTriggerList xmlns="http://www.hikvision.com/ver20/XMLSchema" version="2.0">
+    <EventTrigger version="2.0">
+        <id>1</id>
+        <eventType>VMD</eventType>
+        <videoInputChannelID>1</videoInputChannelID>
+        <EventTriggerNotificationList>
+            <EventTriggerNotification>
+                <id>1</id>
+                <notificationMethod>record</notificationMethod>
+            </EventTriggerNotification>
+        </EventTriggerNotificationList>
+    </EventTrigger>
+    <EventTrigger version="2.0">
+        <id>2</id>
+        <eventType>linedetection</eventType>
+        <videoInputChannelID>2</videoInputChannelID>
+        <EventTriggerNotificationList>
+            <EventTriggerNotification>
+                <id>1</id>
+                <notificationMethod>email</notificationMethod>
+            </EventTriggerNotification>
+        </EventTriggerNotificationList>
+    </EventTrigger>
+    <EventTrigger version="2.0">
+        <id>3</id>
+        <eventType>fielddetection</eventType>
+        <videoInputChannelID>3</videoInputChannelID>
+        <EventTriggerNotificationList>
+            <EventTriggerNotification>
+                <id>1</id>
+                <notificationMethod>beep</notificationMethod>
+            </EventTriggerNotification>
+        </EventTriggerNotificationList>
+    </EventTrigger>
+    <EventTrigger version="2.0">
+        <id>4</id>
+        <eventType>VMD</eventType>
+        <videoInputChannelID>4</videoInputChannelID>
+        <EventTriggerNotificationList>
+            <EventTriggerNotification>
+                <id>1</id>
+                <notificationMethod>center</notificationMethod>
+            </EventTriggerNotification>
+        </EventTriggerNotificationList>
+    </EventTrigger>
+    <EventTrigger version="2.0">
+        <id>5</id>
+        <eventType>VMD</eventType>
+        <videoInputChannelID>5</videoInputChannelID>
+        <EventTriggerNotificationList>
+            <EventTriggerNotification>
+                <id>1</id>
+                <notificationMethod>HTTP</notificationMethod>
+            </EventTriggerNotification>
+        </EventTriggerNotificationList>
+    </EventTrigger>
+</EventTriggerList>"""
+
+
+class GetEventTriggersTestCase(unittest.TestCase):
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    def test_default_notification_methods(self, mock_info, mock_session):
+        """Test that get_event_triggers defaults to center and HTTP only."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        session = mock_session.return_value
+        response = MagicMock()
+        response.status_code = requests.codes.ok
+        response.text = EVENT_TRIGGERS_XML
+        session.get.return_value = response
+
+        camera = HikCamera(host="localhost")
+        # Call get_event_triggers with default (no args)
+        events = camera.get_event_triggers()
+
+        # Should only find VMD on channels 4 and 5 (center and HTTP)
+        self.assertIn("VMD", events)
+        self.assertEqual(sorted(events["VMD"]), [4, 5])
+
+        # Should NOT find events with record, email, beep notification methods
+        self.assertNotIn("linedetection", events)
+        self.assertNotIn("fielddetection", events)
+
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    def test_custom_notification_methods(self, mock_info, mock_session):
+        """Test that get_event_triggers accepts custom notification methods."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        session = mock_session.return_value
+        response = MagicMock()
+        response.status_code = requests.codes.ok
+        response.text = EVENT_TRIGGERS_XML
+        session.get.return_value = response
+
+        camera = HikCamera(host="localhost")
+        # Call get_event_triggers with expanded notification methods
+        events = camera.get_event_triggers(
+            notification_methods={'center', 'HTTP', 'record', 'email', 'beep'}
+        )
+
+        # Should find VMD on channels 1, 4, 5 (record, center, HTTP)
+        self.assertIn("VMD", events)
+        self.assertEqual(sorted(events["VMD"]), [1, 4, 5])
+
+        # Should find linedetection on channel 2 (email)
+        self.assertIn("linedetection", events)
+        self.assertEqual(events["linedetection"], [2])
+
+        # Should find fielddetection on channel 3 (beep)
+        self.assertIn("fielddetection", events)
+        self.assertEqual(events["fielddetection"], [3])
+
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    def test_valid_notification_methods_constant(self, mock_info, mock_session):
+        """Test using VALID_NOTIFICATION_METHODS constant."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        session = mock_session.return_value
+        response = MagicMock()
+        response.status_code = requests.codes.ok
+        response.text = EVENT_TRIGGERS_XML
+        session.get.return_value = response
+
+        camera = HikCamera(host="localhost")
+        # Use the exported constant
+        events = camera.get_event_triggers(
+            notification_methods=VALID_NOTIFICATION_METHODS
+        )
+
+        # Should find all events
+        self.assertIn("VMD", events)
+        self.assertEqual(sorted(events["VMD"]), [1, 4, 5])
+        self.assertIn("linedetection", events)
+        self.assertIn("fielddetection", events)
+
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    def test_case_insensitive_notification_methods(self, mock_info, mock_session):
+        """Test that notification method matching is case insensitive."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        session = mock_session.return_value
+        response = MagicMock()
+        response.status_code = requests.codes.ok
+        response.text = EVENT_TRIGGERS_XML
+        session.get.return_value = response
+
+        camera = HikCamera(host="localhost")
+        # Use uppercase - should still match lowercase in XML
+        events = camera.get_event_triggers(
+            notification_methods={'CENTER', 'http', 'RECORD'}
+        )
+
+        # Should find VMD on channels 1, 4, 5
+        self.assertIn("VMD", events)
+        self.assertEqual(sorted(events["VMD"]), [1, 4, 5])
+
+
+class InjectEventsTestCase(unittest.TestCase):
+    def test_inject_events_adds_new_events(self):
+        """Test that inject_events adds new events to camera event_states."""
+        camera = MagicMock()
+        camera.event_states = {}
+
+        events = {
+            "Motion": [1, 2],
+            "Line Crossing": [3]
+        }
+
+        inject_events_into_camera(camera, events)
+
+        camera.inject_events.assert_called_once_with(events)
+
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    @patch("pyhik.hikvision.HikCamera.get_event_triggers")
+    def test_inject_events_method(self, mock_triggers, mock_info, mock_session):
+        """Test that HikCamera.inject_events correctly adds events."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        mock_triggers.return_value = {}
+        session = mock_session.return_value
+        session.get.return_value = MagicMock(status_code=requests.codes.not_found)
+
+        camera = HikCamera(host="localhost")
+        camera.event_states = {}
+
+        # Inject events
+        events = {
+            "Motion": [1, 2],
+            "Line Crossing": [3]
+        }
+        camera.inject_events(events)
+
+        # Verify events were added
+        self.assertIn("Motion", camera.event_states)
+        self.assertEqual(len(camera.event_states["Motion"]), 2)
+        self.assertEqual(camera.event_states["Motion"][0][1], 1)  # channel 1
+        self.assertEqual(camera.event_states["Motion"][1][1], 2)  # channel 2
+        self.assertFalse(camera.event_states["Motion"][0][0])  # not active
+
+        self.assertIn("Line Crossing", camera.event_states)
+        self.assertEqual(len(camera.event_states["Line Crossing"]), 1)
+        self.assertEqual(camera.event_states["Line Crossing"][0][1], 3)  # channel 3
+
+    @patch("pyhik.hikvision.requests.Session")
+    @patch("pyhik.hikvision.HikCamera.get_device_info")
+    @patch("pyhik.hikvision.HikCamera.get_event_triggers")
+    def test_inject_events_does_not_duplicate(self, mock_triggers, mock_info, mock_session):
+        """Test that inject_events doesn't add duplicate channel events."""
+        mock_info.return_value = {"deviceName": "Test", "deviceID": "12345678901"}
+        mock_triggers.return_value = {}
+        session = mock_session.return_value
+        session.get.return_value = MagicMock(status_code=requests.codes.not_found)
+
+        camera = HikCamera(host="localhost")
+        camera.event_states = {
+            "Motion": [[False, 1, 0, None]]  # Already has channel 1
+        }
+
+        # Try to inject event for same channel
+        events = {"Motion": [1, 2]}
+        camera.inject_events(events)
+
+        # Should only have 2 entries (original + channel 2, not duplicate of 1)
+        self.assertEqual(len(camera.event_states["Motion"]), 2)
+        channels = [sensor[1] for sensor in camera.event_states["Motion"]]
+        self.assertEqual(sorted(channels), [1, 2])
+
+
 if __name__ == "__main__":
     unittest.main()