97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from Brain.yovision_brain.domain import (
|
|
Box,
|
|
Detection,
|
|
Point,
|
|
Zone,
|
|
ZoneEntryEvaluator,
|
|
point_in_polygon,
|
|
zone_from_payload,
|
|
)
|
|
|
|
|
|
class GeometryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.zone = Zone(
|
|
"zone-1",
|
|
"危险区域",
|
|
1,
|
|
(Point(0.4, 0.2), Point(0.8, 0.2), Point(0.8, 0.8), Point(0.4, 0.8)),
|
|
)
|
|
|
|
def test_point_in_polygon_includes_boundary(self) -> None:
|
|
self.assertTrue(point_in_polygon(Point(0.6, 0.5), self.zone.points))
|
|
self.assertTrue(point_in_polygon(Point(0.4, 0.5), self.zone.points))
|
|
self.assertFalse(point_in_polygon(Point(0.2, 0.5), self.zone.points))
|
|
|
|
def test_zone_requires_normalized_three_to_thirty_two_points(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "3 to 32"):
|
|
Zone("zone-1", "bad", 1, (Point(0, 0), Point(1, 1)))
|
|
with self.assertRaisesRegex(ValueError, "within"):
|
|
Point(1.1, 0.5)
|
|
with self.assertRaisesRegex(ValueError, "within"):
|
|
Point(float("nan"), 0.5)
|
|
with self.assertRaisesRegex(ValueError, "within"):
|
|
Box(0.1, 0.1, float("inf"), 0.9)
|
|
|
|
def test_zone_payload_rejects_unknown_fields_and_increments_version(self) -> None:
|
|
updated = zone_from_payload(
|
|
{"name": "新区域", "points": [{"x": 0.1, "y": 0.1}, {"x": 0.9, "y": 0.1}, {"x": 0.5, "y": 0.9}]},
|
|
self.zone,
|
|
)
|
|
self.assertEqual(updated.version, 2)
|
|
self.assertEqual(updated.name, "新区域")
|
|
with self.assertRaisesRegex(ValueError, "unknown"):
|
|
zone_from_payload({"points": [], "tenant_id": "must-not-be-here"}, self.zone)
|
|
|
|
|
|
class ZoneEntryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.zone = Zone(
|
|
"zone-1",
|
|
"危险区域",
|
|
1,
|
|
(Point(0.5, 0.2), Point(0.9, 0.2), Point(0.9, 0.9), Point(0.5, 0.9)),
|
|
)
|
|
ids = iter(("BRN-0001", "BRN-0002", "BRN-0003"))
|
|
self.evaluator = ZoneEntryEvaluator("device-ref", True, track_ttl_frames=2, event_id_factory=lambda: next(ids))
|
|
self.now = datetime(2026, 8, 11, 1, 2, 3, tzinfo=timezone.utc)
|
|
|
|
@staticmethod
|
|
def detection(track: str, center_x: float) -> Detection:
|
|
return Detection(track, "person", Box(center_x - 0.05, 0.3, center_x + 0.05, 0.8))
|
|
|
|
def test_first_seen_inside_does_not_fake_an_entry(self) -> None:
|
|
events, states = self.evaluator.evaluate(1, self.now, self.zone, [self.detection("P-1", 0.7)])
|
|
self.assertEqual(events, [])
|
|
self.assertTrue(states["P-1"])
|
|
|
|
def test_entry_fires_once_until_track_exits_and_reenters(self) -> None:
|
|
self.evaluator.evaluate(1, self.now, self.zone, [self.detection("P-1", 0.3)])
|
|
events, _ = self.evaluator.evaluate(2, self.now, self.zone, [self.detection("P-1", 0.6)])
|
|
repeated, _ = self.evaluator.evaluate(3, self.now, self.zone, [self.detection("P-1", 0.7)])
|
|
self.evaluator.evaluate(4, self.now, self.zone, [self.detection("P-1", 0.3)])
|
|
reentered, _ = self.evaluator.evaluate(5, self.now, self.zone, [self.detection("P-1", 0.6)])
|
|
|
|
self.assertEqual([item.source_event_id for item in events], ["BRN-0001"])
|
|
self.assertEqual(repeated, [])
|
|
self.assertEqual([item.source_event_id for item in reentered], ["BRN-0002"])
|
|
payload = events[0].as_dict()
|
|
self.assertNotIn("id", payload)
|
|
self.assertEqual(payload["confidence"], None)
|
|
self.assertTrue(payload["fixture"])
|
|
|
|
def test_expired_track_reappearing_inside_is_not_an_entry(self) -> None:
|
|
self.evaluator.evaluate(1, self.now, self.zone, [self.detection("P-1", 0.3)])
|
|
self.evaluator.evaluate(4, self.now, self.zone, [])
|
|
events, _ = self.evaluator.evaluate(5, self.now, self.zone, [self.detection("P-1", 0.7)])
|
|
self.assertEqual(events, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|