Event Detector
Event Detector Analyzer Module Overview
This module provides an analyzer (EventDetector
) for converting analyzer outputs into high-level,
human-readable events. It enables detection of complex temporal patterns and conditions based on
metrics from other analyzers like zone counters and line counters.
Key Features
Metric-Based Events: Convert analyzer metrics into meaningful events
Temporal Patterns: Detect conditions that must hold for specific durations
Complex Conditions: Combine multiple metrics with logical operators
Data-Driven: Configure events using YAML or dictionary definitions
Ring Buffer: Internal state management for temporal conditions
Integration Support: Works with any analyzer that produces metrics
Schema Validation: Ensures event definitions match required format
Typical Usage
Configure auxiliary analyzers (e.g., ZoneCounter, LineCounter) for required metrics
Create an EventDetector instance with event definitions
Attach it to a model or compound model
Access detected events via result.events_detected
Use EventNotifier for event-based notifications
Integration Notes
Requires metrics from other analyzers to be present in results
Event definitions must match the event_definition_schema
Supports both YAML and dictionary-based configuration
Events are stored in result.events_detected for downstream use
Key Classes
EventDetector
: Main analyzer class for detecting eventsEventDefinitionSchema
: Schema for validating event definitions
Configuration Options
event_definitions
: YAML file or dictionary containing event definitionsmetrics
: Dictionary mapping metric names to their sourcestemporal_window
: Default duration for evaluating conditionsquantifier
: Default proportion/duration for event triggerscomparator
: Default operator for comparing metrics to thresholds
Classes
EventDetector
EventDetector
Analyzes inference results over time to detect high-level events based on metric conditions.
This analyzer monitors a chosen metric (e.g., ZoneCount, LineCount, or ObjectCount) over a sliding time window and triggers an event when a specified condition is satisfied. The condition consists of a comparison of the metric value against a threshold, and a temporal requirement that the condition holds for a certain duration or proportion of the window. When the condition is met, the event name is added to the events_detected
set in the result.
For example, you can detect an event "PersonInZone" when the count of persons in a region (ZoneCount
) remains above 0 for N seconds, or a "VehicleCountExceeded" event when a line-crossing count exceeds a threshold within a frame. Multiple EventDetector
instances can be attached to the same model to detect different events in parallel.
Attributes:
key_events_detected
str
Name of the result attribute that stores detected event names (defaults to "events_detected").
comparators
Dict[str, Callable[[float, float], bool]]
Mapping of comparator keywords (e.g., "is greater than") to the corresponding comparison functions used internally.
Note
Ensure that required analyzers (such as ZoneCounter
or LineCounter
) are attached before this detector so that the necessary metric data is present in result
. The EventDetector
maintains all state internally (using a ring buffer for timing) and is therefore stateless to callers. It is safe to use in a single-threaded inference pipeline (thread-safe per inference thread).
Functions
__init__(event_description, ...)
__init__(event_description, *, show_overlay=True, annotation_color=None, annotation_font_scale=None, annotation_pos=AnchorPoint.BOTTOM_LEFT)
Initializes an EventDetector with a given event description and overlay settings.
The event_description
defines the event's trigger name, metric, comparison, and timing requirements.
It can be provided as a YAML string or an equivalent dictionary and must conform to the expected schema (event_definition_schema
).
The description includes these key components:
Trigger: Name of the event to detect
when: Metric to evaluate (one of "ZoneCount", "LineCount", or "ObjectCount")
Comparator: A comparison operator (e.g., "is greater than") with a threshold value
during: Duration of the sliding window as
[value, unit]
(unit can be "frames" or "seconds")for at least / for at most (optional): Required portion of the window that the condition must hold true to trigger the event
Event Definition Schema (YAML):
Parameters:
event_description
Union[str, dict]
YAML string or dictionary defining the event conditions (must match the schema above).
required
show_overlay
bool
Whether to draw a label on the frame when the event fires. Defaults to True.
True
annotation_color
tuple
RGB color for the label background. If None, a contrasting color is auto-chosen. Defaults to None.
None
annotation_font_scale
float
Font scale for the overlay text. If None, uses a default scale. Defaults to None.
None
annotation_pos
Union[AnchorPoint, tuple]
Position for the overlay label (an AnchorPoint
or (x,y) coordinate). Defaults to AnchorPoint.BOTTOM_LEFT
.
BOTTOM_LEFT
Raises:
ValidationError
If event_description
does not conform to the required schema for events.
ValueError
If no comparison operator is specified in the event description.
analyze(result)
analyze(result)
Evaluates the configured event condition on an inference result and updates the result if the event is detected.
The method computes the metric value, compares it to the threshold, maintains a history of condition states, and triggers the event when the condition holds for the required duration.
Parameters:
result
InferenceResults
The inference result to analyze (must contain necessary metrics from prior analyzers).
required
Returns:
None
This method modifies the result
in-place by adding to its events_detected
attribute.
Raises:
AttributeError
If the required metric data is missing in result
(e.g., using ZoneCount without attaching ZoneCounter
).
Functions
ZoneCount(result, ...)
ZoneCount(result, params)
Computes the number of detected objects inside a specified zone or zones.
Requires that result.zone_counts
is present (produced by a ZoneCounter
analyzer). This function can filter the count by object class and aggregate counts across multiple zones if needed.
Parameters:
result
InferenceResults
Inference results object containing zone count data.
required
params
dict
Additional parameters to filter/aggregate the count. classes (List[str]): Class labels to include. If None, all classes are counted. index (int): Zone index to count. If None, all zones are included. aggregation (str): Aggregation function to apply across zones. One of 'sum', 'max', 'min', 'mean', 'std'. Defaults to 'sum'.
required
Returns:
Union[int, float]: Total count of matching objects in the selected zone(s).
Raises:
AttributeError
If result.zone_counts
is missing (no ZoneCounter applied upstream).
ValueError
If a specified zone index is out of range for the available zones.
LineCount(result, ...)
LineCount(result, params)
Computes the number of object crossings on a specified line (or across all lines).
Relies on a result.line_counts
attribute (produced by a LineCounter
analyzer). This function can filter count by object class and crossing direction for fine-grained event definitions.
Parameters:
result
InferenceResults
Inference results object containing line crossing counts.
required
params
dict
Filter parameters from the event's "with" clause. index (int): Line index to count. If None, all lines are considered. classes (List[str]): Class labels to include. If None, all detected classes are counted. directions (List[str]): Directions of crossing to include. One of 'left', 'right', 'top', 'bottom'. If None, all directions are counted.
required
Returns:
count
int
Number of line-crossing events that match the specified filters.
Raises:
AttributeError
If result.line_counts
is missing (no LineCounter applied upstream).
ValueError
If a specified line index is out of range for the available lines.
ObjectCount(result, ...)
ObjectCount(result, params)
Counts the detected objects in the result, with optional class and score filtering.
This metric does not require any auxiliary analyzer; it simply tallies detections, optionally constrained by object class and minimum confidence score.
Parameters:
result
InferenceResults
The inference results object containing detections.
required
params
dict
Filter parameters. classes (List[str]): Class labels to include. If None, all detected classes are counted. min_score (float): Minimum confidence score required for counting. If None, no score threshold is applied.
required
Returns:
count
int
Number of detected objects that meet the specified class and score criteria.
Last updated
Was this helpful?