frigate/frigate/config/camera/motion.py
Dmitry Ilyin b420efdebd Native ONVIF cell-motion ingest
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.
2026-05-30 19:43:33 +03:00

102 lines
4.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from enum import Enum
from typing import Any, Optional
from pydantic import Field, field_serializer
from ..base import FrigateBaseModel
from .mask import MotionMaskConfig
__all__ = ["MotionConfig", "MotionSourceEnum"]
class MotionSourceEnum(str, Enum):
internal = "internal"
onvif = "onvif"
class MotionConfig(FrigateBaseModel):
enabled: bool = Field(
default=True,
title="Enable motion detection",
description="Enable or disable motion detection for all cameras; can be overridden per-camera.",
)
source: MotionSourceEnum = Field(
default=MotionSourceEnum.internal,
title="Motion source",
description="Where motion state comes from: Frigate's internal frame analyser, or the camera's ONVIF cell-motion events (requires onvif.events.enabled).",
)
threshold: int = Field(
default=30,
title="Motion threshold",
description="Pixel difference threshold used by the motion detector; higher values reduce sensitivity (range 1-255).",
ge=1,
le=255,
)
lightning_threshold: float = Field(
default=0.8,
title="Lightning threshold",
description="Threshold to detect and ignore brief lighting spikes (lower is more sensitive, values between 0.3 and 1.0). This does not prevent motion detection entirely; it merely causes the detector to stop analyzing additional frames once the threshold is exceeded. Motion-based recordings are still created during these events.",
ge=0.3,
le=1.0,
)
skip_motion_threshold: Optional[float] = Field(
default=None,
title="Skip motion threshold",
description="If set to a value between 0.0 and 1.0, and more than this fraction of the image changes in a single frame, the detector will return no motion boxes and immediately recalibrate. This can save CPU and reduce false positives during lightning, storms, etc., but may miss real events such as a PTZ camera autotracking an object. The tradeoff is between dropping a few megabytes of recordings versus reviewing a couple short clips. Leave unset (None) to disable this feature.",
ge=0.0,
le=1.0,
)
improve_contrast: bool = Field(
default=True,
title="Improve contrast",
description="Apply contrast improvement to frames before motion analysis to help detection.",
)
contour_area: Optional[int] = Field(
default=10,
title="Contour area",
description="Minimum contour area in pixels required for a motion contour to be counted.",
)
delta_alpha: float = Field(
default=0.2,
title="Delta alpha",
description="Alpha blending factor used in frame differencing for motion calculation.",
)
frame_alpha: float = Field(
default=0.01,
title="Frame alpha",
description="Alpha value used when blending frames for motion preprocessing.",
)
frame_height: Optional[int] = Field(
default=100,
title="Frame height",
description="Height in pixels to scale frames to when computing motion.",
)
mask: dict[str, Optional[MotionMaskConfig]] = Field(
default_factory=dict,
title="Mask coordinates",
description="Ordered x,y coordinates defining the motion mask polygon used to include/exclude areas.",
)
mqtt_off_delay: int = Field(
default=30,
title="MQTT off delay",
description="Seconds to wait after last motion before publishing an MQTT 'off' state.",
)
enabled_in_config: Optional[bool] = Field(
default=None,
title="Original motion state",
description="Indicates whether motion detection was enabled in the original static configuration.",
)
raw_mask: dict[str, Optional[MotionMaskConfig]] = Field(
default_factory=dict, exclude=True
)
@field_serializer("mask", when_used="json")
def serialize_mask(self, value: Any, info):
if self.raw_mask:
return self.raw_mask
return value
@field_serializer("raw_mask", when_used="json")
def serialize_raw_mask(self, value: Any, info):
return None