feat: 实现 Brain 区域与方向越线判定 (#15)

Brain/src/yovision_brain/rules/__init__.py

@@ -0,0 +1,21 @@
+"""Brain-internal anonymous area and directional-line rules."""
+
+from .engine import RuleEngine
+from .models import (
+    AreaDefinition,
+    DirectionalLineDefinition,
+    NormalizedPoint,
+    RuleConfigError,
+    RuleDecision,
+    RuleSet,
+)
+
+__all__ = [
+    "AreaDefinition",
+    "DirectionalLineDefinition",
+    "NormalizedPoint",
+    "RuleConfigError",
+    "RuleDecision",
+    "RuleEngine",
+    "RuleSet",
+]

Brain/src/yovision_brain/rules/engine.py

@@ -0,0 +1,112 @@
+"""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)

Brain/src/yovision_brain/rules/models.py

@@ -0,0 +1,84 @@
+"""Versioned Brain-internal rule configuration and decisions."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+class RuleConfigError(ValueError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class NormalizedPoint:
+    x: float
+    y: float
+
+    def __post_init__(self) -> None:
+        if not 0.0 <= self.x <= 1.0 or not 0.0 <= self.y <= 1.0:
+            raise RuleConfigError("rule coordinates must be normalized to 0..1")
+
+
+@dataclass(frozen=True, slots=True)
+class AreaDefinition:
+    rule_id: str
+    points: tuple[NormalizedPoint, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class DirectionalLineDefinition:
+    rule_id: str
+    start: NormalizedPoint
+    end: NormalizedPoint
+    trigger_direction: str
+    deadband: float = 0.005
+
+
+@dataclass(frozen=True, slots=True)
+class RuleSet:
+    version: str
+    profile_id: str
+    width: int
+    height: int
+    areas: tuple[AreaDefinition, ...] = ()
+    directional_lines: tuple[DirectionalLineDefinition, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.version or not self.profile_id or self.width <= 0 or self.height <= 0:
+            raise RuleConfigError("rule version, profile and dimensions are required")
+        identifiers = [rule.rule_id for rule in self.areas] + [rule.rule_id for rule in self.directional_lines]
+        if any(not identifier for identifier in identifiers) or len(set(identifiers)) != len(identifiers):
+            raise RuleConfigError("rule ids must be non-empty and unique")
+        for area in self.areas:
+            if len(area.points) < 3 or abs(_polygon_area(area.points)) < 1e-9:
+                raise RuleConfigError(f"area {area.rule_id!r} must be a non-degenerate polygon")
+        for line in self.directional_lines:
+            if line.start == line.end:
+                raise RuleConfigError(f"line {line.rule_id!r} must have distinct endpoints")
+            if line.trigger_direction not in {"left_to_right", "right_to_left"}:
+                raise RuleConfigError(f"line {line.rule_id!r} has invalid trigger direction")
+            if not 0.0 <= line.deadband < 0.5:
+                raise RuleConfigError(f"line {line.rule_id!r} has invalid deadband")
+
+
+def _polygon_area(points: tuple[NormalizedPoint, ...]) -> float:
+    return sum(
+        first.x * second.y - second.x * first.y
+        for first, second in zip(points, points[1:] + points[:1])
+    ) / 2.0
+
+
+@dataclass(frozen=True, slots=True)
+class RuleDecision:
+    rule_id: str
+    rule_type: str
+    track_id: str
+    state: str
+    triggered: bool
+    reason: str
+    config_version: str
+    profile_id: str
+    width: int
+    height: int
+    anchor: NormalizedPoint
+    timestamp_ns: int