Add get_nvr_events and inject_events_into_camera

pyhik/__init__.py

@@ -0,0 +1,6 @@
+"""pyhik - Python library for Hikvision camera/NVR events."""
+
+from pyhik.hikvision import HikCamera, get_nvr_events, inject_events_into_camera
+from pyhik.constants import __version__
+
+__all__ = ['HikCamera', 'get_nvr_events', 'inject_events_into_camera', '__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

@@ -37,7 +37,7 @@
     DEFAULT_PORT, DEFAULT_HEADERS, XML_NAMESPACE, SENSOR_MAP,
     CAM_DEVICE, NVR_DEVICE, CONNECT_TIMEOUT, READ_TIMEOUT, CONTEXT_INFO,
     CONTEXT_TRIG, CONTEXT_MOTION, CONTEXT_ALERT, CHANNEL_NAMES, ID_TYPES,
-    __version__)
+    VALID_NOTIFICATION_METHODS, __version__)
 
 
 _LOGGING = logging.getLogger(__name__)
@@ -677,3 +677,149 @@ 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 get_nvr_events(host, port=DEFAULT_PORT, usr=None, pwd=None, verify_ssl=True):
+    """Fetch events from NVR with broader notification method support.
+
+    This function extends the standard event detection by also accepting
+    'record', 'email', and 'beep' notification methods, which are commonly
+    used on NVRs but ignored by the standard get_event_triggers method.
+
+    Args:
+        host: The host URL (e.g., 'http://192.168.1.64').
+        port: The port number (default 80).
+        usr: Username for authentication.
+        pwd: Password for authentication.
+        verify_ssl: Whether to verify SSL certificates.
+
+    Returns:
+        Dict mapping event type names to lists of channel numbers.
+    """
+    root_url = '{}:{}'.format(host, port)
+    events = {}
+
+    session = requests.Session()
+    session.verify = verify_ssl
+    session.auth = HTTPDigestAuth(usr, pwd)
+    session.headers.update(DEFAULT_HEADERS)
+
+    urls = [
+        '%s/ISAPI/Event/triggers' % root_url,
+        '%s/Event/triggers' % root_url,
+    ]
+
+    response = None
+    for url in urls:
+        try:
+            response = session.get(url, timeout=CONNECT_TIMEOUT)
+            if response.status_code == requests.codes.ok:
+                break
+        except (requests.exceptions.RequestException,
+                requests.exceptions.ConnectionError):
+            continue
+
+    if response is None or response.status_code != requests.codes.ok:
+        _LOGGING.warning('Unable to fetch event triggers from NVR')
+        session.close()
+        return events
+
+    try:
+        tree = ET.fromstring(response.text)
+    except ET.ParseError as err:
+        _LOGGING.error('Failed to parse event triggers XML: %s', err)
+        session.close()
+        return events
+
+    # Find namespace from root tag
+    namespace = ''
+    root_tag = tree.tag
+    if root_tag.startswith('{'):
+        namespace = root_tag.split('}')[0] + '}'
+
+    # Try different XML structures (camera vs NVR)
+    event_triggers = tree.findall('.//{0}EventTrigger'.format(namespace))
+
+    for trigger in event_triggers:
+        # Get event type
+        event_type_elem = trigger.find('{0}eventType'.format(namespace))
+        if event_type_elem is None or not event_type_elem.text:
+            continue
+
+        event_type = event_type_elem.text.lower()
+
+        # Skip videoloss as it's used for watchdog
+        if event_type == 'videoloss':
+            continue
+
+        # Get channel number
+        channel_num = 0
+        for channel_name in CHANNEL_NAMES:
+            channel_elem = trigger.find('{0}{1}'.format(namespace, channel_name))
+            if channel_elem is not None and channel_elem.text:
+                try:
+                    channel_num = int(channel_elem.text)
+                    break
+                except ValueError:
+                    continue
+
+        # Check if any valid notification method is configured
+        notification_list = trigger.find(
+            '{0}EventTriggerNotificationList'.format(namespace))
+        has_valid_notification = False
+
+        if notification_list is not None:
+            for notification in notification_list:
+                method_elem = notification.find(
+                    '{0}notificationMethod'.format(namespace))
+                if method_elem is not None and method_elem.text:
+                    if method_elem.text.lower() in {m.lower() for m in VALID_NOTIFICATION_METHODS}:
+                        has_valid_notification = True
+                        break
+
+        if has_valid_notification:
+            # Map to friendly name
+            friendly_name = SENSOR_MAP.get(event_type)
+            if friendly_name:
+                events.setdefault(friendly_name, []).append(channel_num)
+
+    session.close()
+    return events
+
+
+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)