From 2a92fa51cdf04a139c84436204166bca7ceb54ca Mon Sep 17 00:00:00 2001 From: Paul Tarjan Date: Tue, 1 Sep 2026 15:01:13 -0600 Subject: [PATCH 1/2] Keep alerts that leave an optional field out A packet missing eventState or activePostCount died on .text of a None find() result, and the one try/except around the whole block turned that into "Problem finding attribute: 'NoneType' object has no attribute 'text'" and discarded the event -- including its event type and channel, which were present. An empty activePostCount was worse: int('') raised ValueError out of process_stream(), and the caller reads ValueError as a dropped connection, so one malformed packet cost the whole stream a reconnect and its backoff. Read each field on its own instead. Only the event type is required; a missing state means the device only posts while the event is happening, so treat the packet as active and let update_stale() clear it, and an unusable post count is reported as 0. --- pyhik/hikvision.py | 88 +++++++++++++++++++++++++----------------- test/test_hikvision.py | 65 +++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 36 deletions(-) diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index 17864e4..0340b8a 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -458,6 +458,13 @@ def element_query(self, element, context): else: return '{%s}%s' % (XML_NAMESPACE, element) + def element_text(self, tree, element, context=CONTEXT_ALERT): + """Text of an element, or None when the device left it out.""" + node = tree.find(self.element_query(element, context)) + if node is None or node.text is None: + return None + return node.text.strip() or None + def fetch_namespace(self, tree, context): """Determine proper namespace to find given element.""" if context == CONTEXT_INFO: @@ -841,42 +848,48 @@ def process_stream(self, tree): if not self.namespace[CONTEXT_ALERT]: self.fetch_namespace(tree, CONTEXT_ALERT) + raw_etype = self.element_text(tree, 'eventType') + if raw_etype is None: + # Nothing identifies the event, so there is nothing to report. + _LOGGING.debug('Alert packet carries no event type, ignoring.') + return + try: - raw_etype = tree.find( - self.element_query('eventType', CONTEXT_ALERT)).text - try: - etype = SENSOR_MAP[raw_etype.lower()] - except KeyError: - # Event type we don't model (e.g. storageDetection); skip - # quietly rather than logging an error for every packet. - _LOGGING.debug('Unsupported event type "%s", ignoring.', - raw_etype) - return - - # Since this pasing is different and not really usefull for now, just return without error. - if len(etype) > 0 and etype == 'Ongoing Events': - return - - estate = tree.find( - self.element_query('eventState', CONTEXT_ALERT)).text - - for idtype in ID_TYPES: - echid = tree.find(self.element_query(idtype, CONTEXT_ALERT)) - if echid is not None: - try: - # Need to make sure this is actually a number - echid = int(echid.text) - break - except (ValueError, TypeError) as err: - # Field must not be an integer or is blank - pass - - ecount = tree.find( - self.element_query('activePostCount', CONTEXT_ALERT)).text - except (AttributeError, KeyError, IndexError) as err: - _LOGGING.error('Problem finding attribute: %s', err) + etype = SENSOR_MAP[raw_etype.lower()] + except KeyError: + # Event type we don't model (e.g. storageDetection); skip + # quietly rather than logging an error for every packet. + _LOGGING.debug('Unsupported event type "%s", ignoring.', + raw_etype) + return + + # Since this pasing is different and not really usefull for now, just return without error. + if len(etype) > 0 and etype == 'Ongoing Events': return + # Every remaining field is optional in practice: devices leave any of + # them out of a packet. None of them identify the event, so a missing + # one must not cost us the alert itself. + estate = self.element_text(tree, 'eventState') + + echid = None + for idtype in ID_TYPES: + raw_echid = self.element_text(tree, idtype) + if raw_echid is None: + continue + try: + # Need to make sure this is actually a number + echid = int(raw_echid) + break + except ValueError: + # Field must not be an integer + continue + + try: + ecount = int(self.element_text(tree, 'activePostCount')) + except (TypeError, ValueError): + ecount = 0 + # Optional smart-event detection target (human, vehicle, ...). It can # appear at the top level or nested in a detection region entry. etarget = tree.find( @@ -895,10 +908,13 @@ def process_stream(self, tree): state = self.fetch_attributes(etype, echid) if state: # Determine if state has changed - # If so, publish, otherwise do nothing - estate = (estate == 'active') + # If so, publish, otherwise do nothing. + # A device that reports no state only posts while the event + # is happening, so the packet itself is the active signal; + # update_stale() clears it once the posts stop. + estate = estate is None or estate == 'active' old_state = state[0] - attr = [estate, echid, int(ecount), + attr = [estate, echid, ecount, datetime.datetime.now(), target_type] self.update_attributes(etype, echid, attr) diff --git a/test/test_hikvision.py b/test/test_hikvision.py index ad6962b..a6c15a8 100644 --- a/test/test_hikvision.py +++ b/test/test_hikvision.py @@ -978,5 +978,70 @@ def test_response_that_is_not_xml(self): self.assertIsNone(camera.current_motion_detection_state) +class PartialAlertPacketTestCase(unittest.TestCase): + """Devices leave optional fields out of alert packets. The event they do + describe still has to reach the consumer.""" + + NS = "{http://www.hikvision.com/ver20/XMLSchema}" + + def _packet(self, drop=(), blank=()): + tree = alert() + for name in drop: + tree.remove(tree.find(self.NS + name)) + for name in blank: + tree.find(self.NS + name).text = "" + return tree + + def _process(self, tree): + camera = make_camera() + published = [] + camera.add_update_callback(published.append, + "%s.Motion.1" % camera.cam_id) + with self.assertNoLogs("pyhik.hikvision", level=logging.ERROR): + camera.process_stream(tree) + return camera, published + + def test_missing_event_state_reports_the_event_as_active(self): + # Only 'active' packets are posted by devices that omit the state; + # update_stale() is what clears them. + camera, published = self._process(self._packet(drop=["eventState"])) + + self.assertTrue(camera.fetch_attributes("Motion", 1)[0]) + self.assertEqual(len(published), 1) + + def test_missing_post_count_still_publishes_the_event(self): + camera, published = self._process( + self._packet(drop=["activePostCount"])) + + self.assertTrue(camera.fetch_attributes("Motion", 1)[0]) + self.assertEqual(camera.fetch_attributes("Motion", 1)[2], 0) + self.assertEqual(len(published), 1) + + def test_unparseable_post_count_does_not_escape(self): + # process_stream() runs inside alert_stream()'s except ValueError, + # so an exception here is read as a dropped connection and costs the + # whole stream a reconnect. + camera, published = self._process( + self._packet(blank=["activePostCount"])) + + self.assertTrue(camera.fetch_attributes("Motion", 1)[0]) + self.assertEqual(len(published), 1) + + def test_missing_event_type_is_ignored_quietly(self): + camera, published = self._process(self._packet(drop=["eventType"])) + + self.assertFalse(camera.fetch_attributes("Motion", 1)[0]) + self.assertEqual(published, []) + + def test_non_numeric_channel_id_falls_through_to_the_next_id_type(self): + tree = self._packet(blank=["channelID"]) + ET.SubElement(tree, self.NS + "dynChannelID").text = "1" + + camera, published = self._process(tree) + + self.assertTrue(camera.fetch_attributes("Motion", 1)[0]) + self.assertEqual(len(published), 1) + + if __name__ == "__main__": unittest.main() From e00efe832cd94c7980da61cdf3bad0dadb857680 Mon Sep 17 00:00:00 2001 From: Paul Tarjan Date: Tue, 1 Sep 2026 17:03:58 -0600 Subject: [PATCH 2/2] Run the tests on Python 3.9 assertNoLogs landed in 3.10, so the four tests on master that use it, and the five added here, error out on the 3.9 leg of the matrix. Install a context-manager equivalent when the real one is missing. --- test/test_hikvision.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/test/test_hikvision.py b/test/test_hikvision.py index a6c15a8..03bc1f3 100644 --- a/test/test_hikvision.py +++ b/test/test_hikvision.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 +import contextlib import datetime import io import logging @@ -16,6 +17,29 @@ CONNECT_TIMEOUT, DOWNLOAD_TIMEOUT, NVR_DEVICE, VALID_NOTIFICATION_METHODS ) +if not hasattr(unittest.TestCase, "assertNoLogs"): + # unittest grew assertNoLogs in 3.10; the library still supports 3.9. + @contextlib.contextmanager + def _assert_no_logs(self, logger, level): + log = logging.getLogger(logger) + captured = [] + handler = logging.Handler() + handler.emit = captured.append + original_level = log.level + log.addHandler(handler) + log.setLevel(level) + try: + yield + finally: + log.removeHandler(handler) + log.setLevel(original_level) + + unexpected = [r.getMessage() for r in captured if r.levelno >= level] + if unexpected: + self.fail("unexpected logs on %s: %s" % (logger, unexpected)) + + unittest.TestCase.assertNoLogs = _assert_no_logs + XML = """ {} true