113 lines
4.9 KiB
Python
113 lines
4.9 KiB
Python
"""Stateful, explainable area and directional-line evaluation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from yovision_brain.vision import TrackedObject
|
|
|
|
from .models import NormalizedPoint, RuleConfigError, RuleDecision, RuleSet
|
|
|
|
_EPSILON = 1e-9
|
|
|
|
|
|
def _anchor(track: TrackedObject, width: int, height: int) -> NormalizedPoint:
|
|
x = (track.box.left + track.box.right) / (2.0 * width)
|
|
y = track.box.bottom / height
|
|
try:
|
|
return NormalizedPoint(x, y)
|
|
except RuleConfigError as exc:
|
|
raise RuleConfigError(f"track {track.track_id!r} anchor is outside the configured frame") from exc
|
|
|
|
|
|
def _on_segment(point: NormalizedPoint, first: NormalizedPoint, second: NormalizedPoint) -> bool:
|
|
cross = (second.x - first.x) * (point.y - first.y) - (second.y - first.y) * (point.x - first.x)
|
|
return abs(cross) <= _EPSILON and min(first.x, second.x) - _EPSILON <= point.x <= max(first.x, second.x) + _EPSILON and min(first.y, second.y) - _EPSILON <= point.y <= max(first.y, second.y) + _EPSILON
|
|
|
|
|
|
def _inside(point: NormalizedPoint, polygon: tuple[NormalizedPoint, ...]) -> bool:
|
|
inside = False
|
|
previous = polygon[-1]
|
|
for current in polygon:
|
|
if _on_segment(point, previous, current):
|
|
return True
|
|
if (current.y > point.y) != (previous.y > point.y):
|
|
crossing_x = (previous.x - current.x) * (point.y - current.y) / (previous.y - current.y) + current.x
|
|
if point.x < crossing_x:
|
|
inside = not inside
|
|
previous = current
|
|
return inside
|
|
|
|
|
|
def _side(point: NormalizedPoint, start: NormalizedPoint, end: NormalizedPoint) -> float:
|
|
return (end.x - start.x) * (point.y - start.y) - (end.y - start.y) * (point.x - start.x)
|
|
|
|
|
|
class RuleEngine:
|
|
"""Evaluates one versioned rule set against one stream session."""
|
|
|
|
def __init__(self, rules: RuleSet) -> None:
|
|
self._rules = rules
|
|
self._area_inside: dict[tuple[str, str], bool] = {}
|
|
self._line_side: dict[tuple[str, str], int] = {}
|
|
|
|
def evaluate(
|
|
self,
|
|
tracks: tuple[TrackedObject, ...],
|
|
*,
|
|
profile_id: str,
|
|
width: int,
|
|
height: int,
|
|
) -> tuple[RuleDecision, ...]:
|
|
if (profile_id, width, height) != (self._rules.profile_id, self._rules.width, self._rules.height):
|
|
raise RuleConfigError("track Profile/resolution does not match the versioned rule configuration")
|
|
decisions: list[RuleDecision] = []
|
|
for track in tracks:
|
|
anchor = _anchor(track, width, height)
|
|
common = dict(
|
|
track_id=track.track_id,
|
|
config_version=self._rules.version,
|
|
profile_id=profile_id,
|
|
width=width,
|
|
height=height,
|
|
anchor=anchor,
|
|
timestamp_ns=track.timestamp_ns,
|
|
)
|
|
for area in self._rules.areas:
|
|
key = (track.track_id, area.rule_id)
|
|
current = _inside(anchor, area.points)
|
|
previous = self._area_inside.get(key, False)
|
|
state = "entered" if current and not previous else "inside" if current else "outside"
|
|
self._area_inside[key] = current
|
|
decisions.append(RuleDecision(
|
|
rule_id=area.rule_id,
|
|
rule_type="danger_area",
|
|
state=state,
|
|
triggered=state == "entered",
|
|
reason=f"bottom-center anchor is {state} the configured polygon",
|
|
**common,
|
|
))
|
|
for line in self._rules.directional_lines:
|
|
key = (track.track_id, line.rule_id)
|
|
value = _side(anchor, line.start, line.end)
|
|
if abs(value) <= line.deadband:
|
|
decisions.append(RuleDecision(
|
|
rule_id=line.rule_id, rule_type="directional_line", state="on_line",
|
|
triggered=False, reason="anchor is inside the line deadband; previous significant side is retained",
|
|
**common,
|
|
))
|
|
continue
|
|
current_side = 1 if value > 0 else -1
|
|
previous_side = self._line_side.get(key)
|
|
self._line_side[key] = current_side
|
|
wanted = (previous_side, current_side) == ((1, -1) if line.trigger_direction == "left_to_right" else (-1, 1))
|
|
crossed = previous_side is not None and previous_side != current_side
|
|
state = "triggered" if wanted else "reverse_crossing" if crossed else "same_side"
|
|
decisions.append(RuleDecision(
|
|
rule_id=line.rule_id,
|
|
rule_type="directional_line",
|
|
state=state,
|
|
triggered=wanted,
|
|
reason=f"directed side transition {previous_side!r}->{current_side}; expected {line.trigger_direction}",
|
|
**common,
|
|
))
|
|
return tuple(decisions)
|