mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-06-26 06:11:54 +03:00
b420efdebd
Adds a per-camera ONVIF subscriber that lets cameras with native
hardware motion detection (e.g. OpenIPC firmware for HiSilicon,
Ingenic and SigmaStar SoCs; many ONVIF Profile-M devices) replace
Frigate's per-frame CPU motion analysis. Two standard ONVIF
transports are consumed in parallel:
- WS-BaseNotification PullPoint for the binary motion state
(tns1:RuleEngine/CellMotionDetector/Motion IsMotion=true|false,
with tns1:VideoSource/MotionAlarm State=true|false accepted as
a fallback for cameras that only publish the legacy topic).
- RTSP analytics metadata stream (application/vnd.onvif.metadata)
for the per-frame cell grid (tt:MotionInCells, base64 + PackBits
bit-packed bitmap). Cell layout is discovered once at startup via
AnalyticsService.GetAnalyticsModules and the camera's CellLayout
transformation is used to map cells to detect-frame pixel
rectangles via connected-components.
New config:
onvif.events.{enabled, subscription_timeout, use_metadata_stream}
motion.source: internal (default) | onvif
When motion.source: onvif, ImprovedMotionDetector is skipped and
motion_boxes come from the camera. Internal motion remains the
default; the new path is fully opt-in.
73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""Unit tests for the ONVIF PullPoint motion-state parser."""
|
|
|
|
import unittest
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from frigate.ptz.onvif_events import _parse_motion_state
|
|
|
|
|
|
class FakeMessage:
|
|
"""Mimic the zeep NotificationMessage shape: a Message attribute holding
|
|
an object whose `_value_1` is an lxml/etree element."""
|
|
|
|
class _Body:
|
|
def __init__(self, element):
|
|
self._value_1 = element
|
|
|
|
def __init__(self, xml: str):
|
|
self.Message = self._Body(ET.fromstring(xml))
|
|
|
|
|
|
_NS = 'xmlns:tt="http://www.onvif.org/ver10/schema"'
|
|
|
|
|
|
def _build_msg(name: str, value: str) -> FakeMessage:
|
|
xml = (
|
|
f"<tt:Message {_NS}>"
|
|
"<tt:Source>"
|
|
'<tt:SimpleItem Name="Source" Value="VideoSourceToken"/>'
|
|
"</tt:Source>"
|
|
"<tt:Data>"
|
|
f'<tt:SimpleItem Name="{name}" Value="{value}"/>'
|
|
"</tt:Data>"
|
|
"</tt:Message>"
|
|
)
|
|
return FakeMessage(xml)
|
|
|
|
|
|
class TestParseMotionState(unittest.TestCase):
|
|
def test_is_motion_true(self):
|
|
self.assertTrue(_parse_motion_state(_build_msg("IsMotion", "true")))
|
|
|
|
def test_is_motion_false(self):
|
|
self.assertFalse(_parse_motion_state(_build_msg("IsMotion", "false")))
|
|
|
|
def test_legacy_state_topic_name(self):
|
|
# The legacy tns1:VideoSource/MotionAlarm payload uses "State" instead
|
|
# of the spec-compliant "IsMotion"; we accept either.
|
|
self.assertTrue(_parse_motion_state(_build_msg("State", "true")))
|
|
self.assertFalse(_parse_motion_state(_build_msg("State", "false")))
|
|
|
|
def test_boolean_aliases(self):
|
|
self.assertTrue(_parse_motion_state(_build_msg("IsMotion", "1")))
|
|
self.assertFalse(_parse_motion_state(_build_msg("IsMotion", "0")))
|
|
|
|
def test_no_state_returns_none(self):
|
|
# Missing the State/IsMotion SimpleItem.
|
|
xml = (
|
|
f"<tt:Message {_NS}>"
|
|
'<tt:Data><tt:SimpleItem Name="Other" Value="yes"/></tt:Data>'
|
|
"</tt:Message>"
|
|
)
|
|
self.assertIsNone(_parse_motion_state(FakeMessage(xml)))
|
|
|
|
def test_no_message_returns_none(self):
|
|
class Empty:
|
|
pass
|
|
|
|
self.assertIsNone(_parse_motion_state(Empty()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|