feat(edge): PipelineRuntime assembles components + hot-swaps zones

edge/runtime.py

@@ -0,0 +1,37 @@
+from edge.app import run_once
+from edge.tracking import IouTracker
+from edge.smoothing import TrackSmootherPool
+from edge.detectors.fall import FallDetector
+from edge.detectors.immobility import ImmobilityWatchdog
+from edge.detectors.ghost import GhostIncidentDetector
+from edge.detectors.zone import ZoneIntrusionDetector
+from edge.engine import EventEngine
+from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
+
+
+class PipelineRuntime:
+    def __init__(self, config, source, pose, sink):
+        self.config = config
+        self.source = source
+        self.pose = pose
+        self.sink = sink
+        self.tracker = IouTracker()
+        self.pool = TrackSmootherPool()
+        self.zone_detector = ZoneIntrusionDetector([], config.zone)
+        self.engine = EventEngine(
+            FallDetector(config.fall),
+            ImmobilityWatchdog(config.immobility),
+            GhostIncidentDetector(config.ghost),
+            self.zone_detector,
+            config.engine,
+        )
+        self.ring = RingBuffer()
+        self.snapshotter = Snapshotter(FaceBlurrer())
+
+    def step(self, now: float) -> list:
+        return run_once(self.source, self.pose, self.tracker, self.pool,
+                        self.engine, self.ring, self.snapshotter, self.sink,
+                        now_fn=lambda: now)
+
+    def apply_zones(self, zones: list) -> None:
+        self.zone_detector.zones = list(zones)

tests/test_runtime.py

@@ -0,0 +1,43 @@
+import numpy as np
+from edge.config import load_config
+from edge.models import Person, BBox, Keypoint
+from edge.detectors.zone import Zone
+from edge.runtime import PipelineRuntime
+
+
+def _lying_person():
+    kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
+    kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
+    kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
+    kps[0] = Keypoint(90, 320, 0.9)
+    return Person(BBox(90, 290, 210, 320), kps)
+
+
+class _Source:
+    def read(self): return True, np.full((360, 640, 3), 128, np.uint8)
+
+
+class _Pose:
+    def infer(self, frame): return [_lying_person()]
+
+
+class _Sink:
+    def __init__(self): self.events = []
+    def try_send(self, e, snaps): self.events.append(e); return True
+
+
+def test_step_runs_pipeline_and_emits_fall():
+    cfg = load_config({"FALL_PRONE_S": "0.5"})
+    sink = _Sink()
+    rt = PipelineRuntime(cfg, _Source(), _Pose(), sink)
+    rt.step(now=0.0)
+    rt.step(now=0.6)
+    from edge.events import EventType
+    assert any(e.event_type == EventType.FALL for e in sink.events)
+
+
+def test_apply_zones_hotswaps():
+    cfg = load_config({})
+    rt = PipelineRuntime(cfg, _Source(), _Pose(), _Sink())
+    rt.apply_zones([Zone("z1", "restricted", [(0, 0), (9, 0), (9, 9)])])
+    assert rt.zone_detector.zones[0].zone_id == "z1"