diff --git a/pyhik/__init__.py b/pyhik/__init__.py index e69de29..bac8386 100644 --- a/pyhik/__init__.py +++ b/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__' +] diff --git a/pyhik/constants.py b/pyhik/constants.py index fa00c90..dc3fc71 100755 --- a/pyhik/constants.py +++ b/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'} diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index e93b306..1834ea8 100755 --- a/pyhik/hikvision.py +++ b/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) diff --git a/test/test_hikvision.py b/test/test_hikvision.py index 21d4687..dc7d25b 100644 --- a/test/test_hikvision.py +++ b/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 = """ {} @@ -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 = """ + + + 1 + VMD + 1 + + + 1 + record + + + + + 2 + linedetection + 2 + + + 1 + email + + + + + 3 + fielddetection + 3 + + + 1 + beep + + + + + 4 + VMD + 4 + + + 1 + center + + + + + 5 + VMD + 5 + + + 1 + HTTP + + + +""" + + +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()