feat(brain): add single-stream visual prototype (T-017)

Brain/tests/test_domain.py

@@ -0,0 +1,96 @@
+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()

Brain/tests/test_runtime.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import unittest
+
+from Brain.yovision_brain.runtime import DemoEngine
+from Brain.yovision_brain.source import SyntheticSource, cv2
+
+
+@unittest.skipIf(cv2 is None, "pinned OpenCV package is not installed")
+class RuntimeTests(unittest.TestCase):
+    def test_rejects_non_finite_fps(self) -> None:
+        with self.assertRaisesRegex(ValueError, "fps"):
+            DemoEngine(SyntheticSource(width=320, height=180), fps=float("nan"))
+
+    def test_synthetic_step_produces_safe_state_and_jpeg(self) -> None:
+        engine = DemoEngine(SyntheticSource(width=320, height=180), fps=2.0)
+        engine.step()
+        state = engine.state()
+        self.assertTrue(state["source"]["fixture"])
+        self.assertEqual(state["detector"]["name"], "scripted_fixture")
+        self.assertEqual(state["frame"]["width"], 320)
+        self.assertGreater(len(engine.frame_jpeg() or b""), 100)
+        self.assertNotIn("url", state["source"])
+
+    def test_zone_update_increments_version(self) -> None:
+        engine = DemoEngine(SyntheticSource(width=320, height=180))
+        updated = engine.update_zone({"name": "新区域", "points": [{"x": 0.1, "y": 0.1}, {"x": 0.9, "y": 0.1}, {"x": 0.5, "y": 0.9}]})
+        self.assertEqual(updated.version, 2)
+        self.assertEqual(engine.state()["zone"]["name"], "新区域")
+
+
+if __name__ == "__main__":
+    unittest.main()

Brain/tests/test_server.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import http.client
+import json
+import threading
+import unittest
+
+from Brain.yovision_brain.domain import Point, Zone, zone_from_payload
+from Brain.yovision_brain.server import DemoHTTPServer, DemoHandler, index_path, parse_bind
+
+
+class FakeEngine:
+    def __init__(self) -> None:
+        self.zone = Zone("zone-demo-01", "测试区域", 1, (Point(0.1, 0.1), Point(0.9, 0.1), Point(0.5, 0.9)))
+
+    def state(self) -> dict[str, object]:
+        return {"source": {"label": "safe", "connected": True}, "events": [], "zone": {"version": self.zone.version}}
+
+    def frame_jpeg(self) -> bytes:
+        return b"\xff\xd8safe-jpeg\xff\xd9"
+
+    def update_zone(self, payload: object) -> Zone:
+        self.zone = zone_from_payload(payload, self.zone)
+        return self.zone
+
+
+class BindTests(unittest.TestCase):
+    def test_only_explicit_loopback_is_allowed(self) -> None:
+        self.assertEqual(parse_bind("127.0.0.1:8090"), ("127.0.0.1", 8090))
+        self.assertEqual(parse_bind("localhost:8090"), ("localhost", 8090))
+        with self.assertRaisesRegex(ValueError, "loopback"):
+            parse_bind("0.0.0.0:8090")
+        with self.assertRaisesRegex(ValueError, "loopback"):
+            parse_bind("192.168.1.10:8090")
+
+
+class HTTPTests(unittest.TestCase):
+    def setUp(self) -> None:
+        html = index_path().read_text(encoding="utf-8")
+        self.server = DemoHTTPServer(("127.0.0.1", 0), FakeEngine(), "test-token", html)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.connection = http.client.HTTPConnection("127.0.0.1", self.server.server_port, timeout=2)
+
+    def tearDown(self) -> None:
+        self.connection.close()
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=2)
+
+    def test_index_substitutes_token_and_sets_security_headers(self) -> None:
+        self.connection.request("GET", "/brain-demo")
+        response = self.connection.getresponse()
+        body = response.read().decode("utf-8")
+        self.assertEqual(response.status, 200)
+        self.assertIn('content="test-token"', body)
+        self.assertNotIn("__BRAIN_DEMO_TOKEN__", body)
+        self.assertEqual(response.getheader("X-Frame-Options"), "DENY")
+        self.assertIn("default-src 'self'", response.getheader("Content-Security-Policy"))
+
+    def test_zone_write_requires_token_and_rejects_unknown_fields(self) -> None:
+        body = json.dumps({"name": "新区域", "points": [{"x": 0.1, "y": 0.1}, {"x": 0.9, "y": 0.1}, {"x": 0.5, "y": 0.9}]})
+        self.connection.request("PUT", "/api/v1/zones/active", body=body, headers={"Content-Type": "application/json"})
+        forbidden = self.connection.getresponse()
+        forbidden.read()
+        self.assertEqual(forbidden.status, 403)
+
+        self.connection.request("PUT", "/api/v1/zones/active", body=body, headers={"Content-Type": "application/json", "X-Brain-Demo-Token": "test-token"})
+        accepted = self.connection.getresponse()
+        payload = json.loads(accepted.read())
+        self.assertEqual(accepted.status, 200)
+        self.assertEqual(payload["version"], 2)
+
+        invalid = json.dumps({"name": "bad", "points": [], "tenant_id": "leak"})
+        self.connection.request("PUT", "/api/v1/zones/active", body=invalid, headers={"Content-Type": "application/json", "X-Brain-Demo-Token": "test-token"})
+        rejected = self.connection.getresponse()
+        self.assertEqual(rejected.status, 400)
+        self.assertEqual(json.loads(rejected.read())["code"], "invalid_zone")
+
+
+if __name__ == "__main__":
+    unittest.main()

Brain/tests/test_source.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from Brain.yovision_brain.domain import Box
+from Brain.yovision_brain.source import CentroidTracker, read_stream_url
+
+
+class StreamURLTests(unittest.TestCase):
+    def test_requires_absolute_single_rtsp_url_file(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "stream.url"
+            path.write_text("rtsp://user:secret@127.0.0.1:8554/camera\n", encoding="utf-8")
+            self.assertEqual(read_stream_url(str(path)), "rtsp://user:secret@127.0.0.1:8554/camera")
+            path.write_text("rtsp://127.0.0.1/a\nrtsp://127.0.0.1/b\n", encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "exactly one") as caught:
+                read_stream_url(str(path))
+            self.assertNotIn("127.0.0.1", str(caught.exception))
+
+    def test_rejects_relative_path_without_echoing_input(self) -> None:
+        with self.assertRaisesRegex(ValueError, "absolute"):
+            read_stream_url("camera-secret.url")
+
+    def test_rejects_url_files_inside_repository(self) -> None:
+        repository_file = Path(__file__).resolve()
+        with self.assertRaisesRegex(ValueError, "outside the repository"):
+            read_stream_url(str(repository_file))
+
+
+class TrackerTests(unittest.TestCase):
+    def test_nearby_boxes_retain_track_and_distant_box_gets_new_track(self) -> None:
+        tracker = CentroidTracker(max_distance=0.2)
+        first = tracker.update([(Box(0.1, 0.1, 0.2, 0.4), 0.8)], 1)
+        nearby = tracker.update([(Box(0.12, 0.1, 0.22, 0.4), 0.7)], 2)
+        distant = tracker.update([(Box(0.7, 0.1, 0.8, 0.4), 0.9)], 3)
+        self.assertEqual(first[0].track_id, nearby[0].track_id)
+        self.assertNotEqual(nearby[0].track_id, distant[0].track_id)
+
+
+if __name__ == "__main__":
+    unittest.main()

Brain/tests/test_ui_contract.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import re
+import unittest
+from pathlib import Path
+
+
+HTML_PATH = Path(__file__).resolve().parents[2] / "docs" / "design" / "brain" / "index.html"
+
+
+class UIContractTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.html = HTML_PATH.read_text(encoding="utf-8")
+
+    def test_prototype_is_self_contained_and_labels_fixture_truthfully(self) -> None:
+        self.assertIn("合成回放不等于模型效果", self.html)
+        self.assertIn("离线 HTML 原型数据", self.html)
+        self.assertNotRegex(self.html, r'(?:src|href)=["\']https?://')
+        self.assertNotIn("@import url", self.html)
+
+    def test_accessibility_and_responsive_guards_are_present(self) -> None:
+        required = (
+            'name="viewport"',
+            'class="skip-link"',
+            'aria-live="polite"',
+            ':focus-visible',
+            'prefers-reduced-motion',
+            'min-height: 44px',
+            '@media (max-width: 420px)',
+            '.toolbar { display: grid; grid-template-columns: 1fr; }',
+            '键盘用户可使用右侧坐标表单',
+        )
+        for marker in required:
+            with self.subTest(marker=marker):
+                self.assertIn(marker, self.html)
+
+    def test_no_structural_emoji_or_unescaped_secret_placeholder_in_text(self) -> None:
+        visible_without_script = re.sub(r"<script[\s\S]*?</script>", "", self.html)
+        self.assertNotRegex(visible_without_script, r"[\U0001F300-\U0001FAFF]")
+        self.assertEqual(self.html.count("__BRAIN_DEMO_TOKEN__"), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()