From f6f561f2e275974512a0a486419fe7bab08729a6 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Fri, 28 Aug 2026 23:07:46 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20Brain=20=E5=8C=BA?= =?UTF-8?q?=E5=9F=9F=E4=B8=8E=E6=96=B9=E5=90=91=E8=B6=8A=E7=BA=BF=E5=88=A4?= =?UTF-8?q?=E5=AE=9A=20(#15)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Brain/src/yovision_brain/rules/__init__.py | 21 ++++ Brain/src/yovision_brain/rules/engine.py | 112 +++++++++++++++++++++ Brain/src/yovision_brain/rules/models.py | 84 ++++++++++++++++ Brain/tests/fixtures/rules/README.md | 5 + Brain/tests/rules/test_engine.py | 81 +++++++++++++++ 5 files changed, 303 insertions(+) create mode 100644 Brain/src/yovision_brain/rules/__init__.py create mode 100644 Brain/src/yovision_brain/rules/engine.py create mode 100644 Brain/src/yovision_brain/rules/models.py create mode 100644 Brain/tests/fixtures/rules/README.md create mode 100644 Brain/tests/rules/test_engine.py diff --git a/Brain/src/yovision_brain/rules/__init__.py b/Brain/src/yovision_brain/rules/__init__.py new file mode 100644 index 0000000..f0c4789 --- /dev/null +++ b/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", +] diff --git a/Brain/src/yovision_brain/rules/engine.py b/Brain/src/yovision_brain/rules/engine.py new file mode 100644 index 0000000..6d4346e --- /dev/null +++ b/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) diff --git a/Brain/src/yovision_brain/rules/models.py b/Brain/src/yovision_brain/rules/models.py new file mode 100644 index 0000000..4487f3e --- /dev/null +++ b/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 diff --git a/Brain/tests/fixtures/rules/README.md b/Brain/tests/fixtures/rules/README.md new file mode 100644 index 0000000..1390d46 --- /dev/null +++ b/Brain/tests/fixtures/rules/README.md @@ -0,0 +1,5 @@ +# Brain rule fixtures + +Rule tests use normalized synthetic geometry and anonymous track IDs only. Do +not add customer site layouts, camera paths, identities, credentials, or a +copy of a future cross-project contract. diff --git a/Brain/tests/rules/test_engine.py b/Brain/tests/rules/test_engine.py new file mode 100644 index 0000000..1ba8fb4 --- /dev/null +++ b/Brain/tests/rules/test_engine.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import pytest + +from yovision_brain.rules import ( + AreaDefinition, + DirectionalLineDefinition, + NormalizedPoint, + RuleConfigError, + RuleEngine, + RuleSet, +) +from yovision_brain.vision import BoundingBox, TrackedObject + + +def point(x: float, y: float) -> NormalizedPoint: + return NormalizedPoint(x, y) + + +def rules() -> RuleSet: + return RuleSet( + version="rules-v7", + profile_id="main", + width=100, + height=100, + areas=(AreaDefinition("yard", (point(0.2, 0.2), point(0.8, 0.2), point(0.8, 0.8), point(0.2, 0.8))),), + directional_lines=(DirectionalLineDefinition("gate", point(0.5, 0.1), point(0.5, 0.9), "left_to_right", 0.01),), + ) + + +def track(track_id: str, anchor_x: int, anchor_y: int, sequence: int = 0) -> TrackedObject: + return TrackedObject(track_id, BoundingBox(anchor_x - 1, anchor_y - 2, anchor_x + 1, anchor_y), "anonymous_target", 1.0, sequence, sequence) + + +def decisions(engine: RuleEngine, item: TrackedObject): + return engine.evaluate((item,), profile_id="main", width=100, height=100) + + +def test_area_outside_entered_inside_and_boundary() -> None: + engine = RuleEngine(rules()) + assert decisions(engine, track("one", 10, 50))[0].state == "outside" + entered = decisions(engine, track("one", 20, 50, 1))[0] + assert (entered.state, entered.triggered) == ("entered", True) + inside = decisions(engine, track("one", 50, 50, 2))[0] + assert (inside.state, inside.triggered) == ("inside", False) + assert inside.config_version == "rules-v7" + + +def test_direction_and_reverse_crossing_are_distinct() -> None: + engine = RuleEngine(rules()) + decisions(engine, track("one", 40, 50)) + forward = decisions(engine, track("one", 60, 50, 1))[1] + assert (forward.state, forward.triggered) == ("triggered", True) + + reverse_engine = RuleEngine(rules()) + decisions(reverse_engine, track("two", 60, 50)) + reverse = decisions(reverse_engine, track("two", 40, 50, 1))[1] + assert (reverse.state, reverse.triggered) == ("reverse_crossing", False) + + +def test_line_deadband_prevents_jitter_trigger() -> None: + engine = RuleEngine(rules()) + decisions(engine, track("one", 40, 50)) + on_line = decisions(engine, track("one", 50, 50, 1))[1] + assert (on_line.state, on_line.triggered) == ("on_line", False) + triggered = decisions(engine, track("one", 60, 50, 2))[1] + assert triggered.triggered is True + + +def test_profile_resolution_mismatch_is_rejected() -> None: + with pytest.raises(RuleConfigError, match="Profile/resolution"): + RuleEngine(rules()).evaluate((track("one", 20, 20),), profile_id="sub", width=100, height=100) + + +def test_invalid_polygon_line_and_duplicate_ids_are_rejected() -> None: + with pytest.raises(RuleConfigError, match="non-degenerate"): + RuleSet("v", "main", 10, 10, areas=(AreaDefinition("bad", (point(0, 0), point(0.5, 0.5), point(1, 1))),)) + with pytest.raises(RuleConfigError, match="distinct endpoints"): + RuleSet("v", "main", 10, 10, directional_lines=(DirectionalLineDefinition("bad", point(0, 0), point(0, 0), "left_to_right"),)) + with pytest.raises(RuleConfigError, match="unique"): + RuleSet("v", "main", 10, 10, areas=(AreaDefinition("same", (point(0, 0), point(1, 0), point(0, 1))),), directional_lines=(DirectionalLineDefinition("same", point(0, 0), point(1, 1), "left_to_right"),))