Merge feat/p1e-multicam: run multiple cameras in one edge daemon

CSI + USB (or any mix) at once: one PipelineRuntime per camera with its own
camera_id, sharing the pose adapter (Hailo round-robin), offline queue, and cloud
sync; zones routed per camera. run_loop generalized to N cameras with per-camera
error isolation; single-camera path unchanged. Adversarial review clean.

deploy/README.md

@@ -105,6 +105,17 @@ Validate on device per `docs/hardware-bringup.md` before trusting output (the Ha
 decode is unit-locked, but the HailoRT inference + picamera2 color order are hardware-
 validated). USB webcam stays available anytime via `CAMERA_BACKEND=uvc`.
 
+**Multiple cameras at once.** Set `CAMERAS=backend:device:camera_id,...` to run several
+cameras in one daemon — each gets its own detection pipeline and `camera_id`, they share
+the Hailo device (round-robin) and the offline queue, and cloud zones are routed to the
+matching camera. Example — CSI + USB together:
+
+```ini
+CAMERAS=csi:0:cam-01,uvc:0:cam-02
+```
+
+Each `camera_id` must match the `camera_id` on that camera's zones in the cloud.
+
 > **Air gap:** `hailo-all` / `python3-picamera2` are the ONE piece this bundle can't
 > carry (Hailo's own driver + apt tree). Pre-install them on the Pi image, or transfer
 > their `.deb`s separately, before running `install_offline.sh`. The pose HEF is

deploy/systemd/move-detect-edge.env.example

@@ -28,6 +28,10 @@ CAMERA_BACKEND=uvc
 UVC_DEVICE=0
 # CSI_DEVICE: picamera2 camera_num for the CSI camera (used when CAMERA_BACKEND=csi).
 CSI_DEVICE=0
+# CAMERAS (optional): run several cameras at once; overrides CAMERA_BACKEND/*_DEVICE.
+# Format backend:device:camera_id, comma-separated. Each camera_id must match its
+# cloud zones. Example — CSI + USB together on a Hailo Pi:
+#CAMERAS=csi:0:cam-01,uvc:0:cam-02
 # TARGET_FPS: frame pacing = 1/TARGET_FPS.
 TARGET_FPS=12
 

edge/daemon.py

@@ -3,22 +3,31 @@ import os
 
 def run_loop(runtime, client, queue, *, clock, sleep, iterations=None,
              sync_every=150, heartbeat_every=900, flush_every=300,
-             frame_interval=0.0):
+             frame_interval=0.0, zones_for=None):
+    # `runtime` may be a single PipelineRuntime or a list of them (multi-camera). All
+    # cameras share this loop, one shared clock read per tick, and the periodic
+    # sync/heartbeat/flush. A per-camera step exception is counted, never fatal.
+    runtimes = list(runtime) if isinstance(runtime, (list, tuple)) else [runtime]
+    if zones_for is None:
+        from edge.zones import zones_from_config
+        zones_for = lambda cfg, rt: zones_from_config(cfg)
     counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0}
     i = 0
     while iterations is None or i < iterations:
         i += 1
-        try:
-            runtime.step(clock())
-            counters["frames"] += 1
-        except Exception:
-            counters["errors"] += 1
+        now = clock()  # one timestamp per tick, shared across cameras
+        for rt in runtimes:
+            try:
+                rt.step(now)
+                counters["frames"] += 1
+            except Exception:
+                counters["errors"] += 1
         try:
             if sync_every and i % sync_every == 0:
                 cfg = client.fetch_config()
                 if cfg is not None:
-                    from edge.zones import zones_from_config
-                    runtime.apply_zones(zones_from_config(cfg))
+                    for rt in runtimes:
+                        rt.apply_zones(zones_for(cfg, rt))
                     counters["syncs"] += 1
             if heartbeat_every and i % heartbeat_every == 0:
                 if client.send_heartbeat({"fps": 0.0, "frames": counters["frames"]}):
@@ -33,13 +42,55 @@ def run_loop(runtime, client, queue, *, clock, sleep, iterations=None,
     return counters
 
 
+def parse_cameras(env, default_camera_id):
+    """CAMERAS='backend:device:camera_id,...' -> [(backend, device, camera_id), ...].
+    When CAMERAS is unset, fall back to a single camera from
+    CAMERA_BACKEND + {UVC,CSI}_DEVICE (using default_camera_id)."""
+    raw = (env.get("CAMERAS") or "").strip()
+    if raw:
+        cams = []
+        for spec in raw.split(","):
+            spec = spec.strip()
+            if not spec:
+                continue
+            parts = [p.strip() for p in spec.split(":")]
+            if len(parts) != 3:
+                raise ValueError(f"bad CAMERAS entry {spec!r} (want backend:device:camera_id)")
+            cams.append((parts[0].lower(), parts[1], parts[2]))
+        if not cams:
+            raise ValueError("CAMERAS is set but has no valid entries")  # never monitor zero cameras
+        return cams
+    backend = (env.get("CAMERA_BACKEND") or "uvc").lower()
+    device = env.get("CSI_DEVICE" if backend == "csi" else "UVC_DEVICE") or "0"
+    return [(backend, device, default_camera_id)]
+
+
 def main() -> None:  # pragma: no cover (integration entrypoint, exercised on device)
     import time
+    from dataclasses import replace
     from edge.config import load_config
     from edge.cloudclient import CloudClient
     from edge.alert import PostClient
     from edge.queue import OfflineQueue
     from edge.runtime import PipelineRuntime
+    from edge.zones import zones_from_config
+
+    def build_source(backend, device):
+        if backend == "csi":
+            from edge.adapters.picamera_source import PiCameraSource
+            return PiCameraSource(int(device))
+        from edge.adapters.uvc_source import UvcSource
+        return UvcSource(int(device))
+
+    def build_pose():
+        # POSE_BACKEND: "hailo" (Pi5 + Hailo-8, needs HAILO_HEF) or CPU "movenet"
+        # (needs MOVENET_ONNX). One shared adapter serves every camera (the Hailo
+        # device / onnxruntime session is reused per frame, round-robin across cams).
+        if os.environ.get("POSE_BACKEND", "movenet").lower() == "hailo":
+            from edge.adapters.hailo_pose import HailoPoseAdapter
+            return HailoPoseAdapter(os.environ["HAILO_HEF"])
+        from edge.adapters.movenet_pose import MoveNetPoseAdapter
+        return MoveNetPoseAdapter(os.environ["MOVENET_ONNX"])
 
     cfg = load_config(dict(os.environ))
     nonce = lambda: os.urandom(8).hex()
@@ -48,22 +99,14 @@ def main() -> None:  # pragma: no cover (integration entrypoint, exercised on de
     client = CloudClient(base, cfg.device_id, cfg.secret, nonce, now)
     post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now)
     queue = OfflineQueue(os.environ.get("QUEUE_DB", "queue.db"), post)
-    # CAMERA_BACKEND selects the capture source: "csi" (Pi CSI cam via picamera2,
-    # uses CSI_DEVICE) or the default "uvc" (USB webcam via OpenCV, uses UVC_DEVICE).
-    if os.environ.get("CAMERA_BACKEND", "uvc").lower() == "csi":
-        from edge.adapters.picamera_source import PiCameraSource
-        source = PiCameraSource(int(os.environ.get("CSI_DEVICE", "0")))
-    else:
-        from edge.adapters.uvc_source import UvcSource
-        source = UvcSource(int(os.environ.get("UVC_DEVICE", "0")))
-    # POSE_BACKEND selects the accelerator path: "hailo" (Pi5 + Hailo-8, needs HAILO_HEF)
-    # or the default CPU "movenet" (needs MOVENET_ONNX).
-    if os.environ.get("POSE_BACKEND", "movenet").lower() == "hailo":
-        from edge.adapters.hailo_pose import HailoPoseAdapter
-        pose = HailoPoseAdapter(os.environ["HAILO_HEF"])
-    else:
-        from edge.adapters.movenet_pose import MoveNetPoseAdapter
-        pose = MoveNetPoseAdapter(os.environ["MOVENET_ONNX"])
-    runtime = PipelineRuntime(cfg, source, pose, queue)
-    run_loop(runtime, client, queue, clock=now, sleep=time.sleep, iterations=None,
-             frame_interval=1.0 / float(os.environ.get("TARGET_FPS", "12")))
+
+    pose = build_pose()  # shared across cameras
+    runtimes = []
+    for backend, device, cam in parse_cameras(dict(os.environ), cfg.camera_id):
+        # Per-camera config so events are stamped with that camera's id.
+        cam_cfg = replace(cfg, camera_id=cam, engine=replace(cfg.engine, camera_id=cam))
+        runtimes.append(PipelineRuntime(cam_cfg, build_source(backend, device), pose, queue))
+
+    run_loop(runtimes, client, queue, clock=now, sleep=time.sleep, iterations=None,
+             frame_interval=1.0 / float(os.environ.get("TARGET_FPS", "12")),
+             zones_for=lambda c, rt: zones_from_config(c, rt.camera_id))

edge/runtime.py

@@ -12,6 +12,7 @@ from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
 class PipelineRuntime:
     def __init__(self, config, source, pose, sink):
         self.config = config
+        self.camera_id = config.camera_id  # for per-camera zone routing (multi-cam)
         self.source = source
         self.pose = pose
         self.sink = sink

edge/zones.py

@@ -3,20 +3,24 @@ import json
 from edge.detectors.zone import Zone
 
 
-def zones_from_config(config: dict) -> list[Zone]:
+def zones_from_config(config: dict, camera_id: str | None = None) -> list[Zone]:
+    """Parse cloud config zone rows into Zone objects. When camera_id is given, only
+    rows for that camera are kept (multi-camera routing); None keeps all rows."""
     out: list[Zone] = []
     rows = config.get("zones", []) or []
     if not isinstance(rows, list):
         return []
     for row in rows:
         try:
+            if camera_id is not None and str(row["camera_id"]) != str(camera_id):
+                continue
             raw = row["polygon_json"]
             pts = json.loads(raw) if isinstance(raw, str) else raw
             polygon = [(float(p[0]), float(p[1])) for p in pts]
             if len(polygon) < 3:
                 continue
             out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon))
-        except (KeyError, ValueError, TypeError, IndexError, json.JSONDecodeError):
+        except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError):
             continue
     return out
 

tests/test_multicam.py

@@ -0,0 +1,82 @@
+"""TDD for multi-camera support: per-camera zone routing, a run_loop that steps
+N cameras per tick, and CAMERAS env parsing. All with fakes (no hardware)."""
+from edge.zones import zones_from_config
+from edge.daemon import run_loop, parse_cameras
+
+
+def _cfg_two_cams():
+    return {"zones": [
+        {"zone_id": "z1", "camera_id": "cam-01", "level": "restricted",
+         "polygon_json": "[[0,0],[9,0],[9,9]]"},
+        {"zone_id": "z2", "camera_id": "cam-02", "level": "warning",
+         "polygon_json": "[[0,0],[5,0],[5,5]]"},
+    ], "thresholds": {}}
+
+
+def test_zones_filtered_by_camera():
+    cfg = _cfg_two_cams()
+    assert [z.zone_id for z in zones_from_config(cfg, "cam-01")] == ["z1"]
+    assert [z.zone_id for z in zones_from_config(cfg, "cam-02")] == ["z2"]
+    assert len(zones_from_config(cfg)) == 2          # no filter -> all (backward compat)
+    assert zones_from_config(cfg, "cam-99") == []    # unknown camera -> none
+
+
+class _RT:
+    def __init__(self, cam): self.camera_id = cam; self.steps = 0; self.applied = None
+    def step(self, now): self.steps += 1; return []
+    def apply_zones(self, z): self.applied = z
+
+
+class _Client:
+    def __init__(self, cfg): self._cfg = cfg; self.configs = 0; self.beats = 0
+    def fetch_config(self): self.configs += 1; return self._cfg
+    def send_heartbeat(self, m): self.beats += 1; return True
+
+
+class _Q:
+    def __init__(self): self.flushes = 0
+    def flush(self): self.flushes += 1; return 0
+
+
+def test_run_loop_steps_all_cameras_and_routes_zones_per_camera():
+    cfg = _cfg_two_cams()
+    r1, r2 = _RT("cam-01"), _RT("cam-02")
+    run_loop([r1, r2], _Client(cfg), _Q(), clock=lambda: 0.0, sleep=lambda s: None,
+             iterations=5, sync_every=5, heartbeat_every=99, flush_every=99,
+             zones_for=lambda c, rt: zones_from_config(c, rt.camera_id))
+    assert r1.steps == 5 and r2.steps == 5
+    assert [z.zone_id for z in r1.applied] == ["z1"]
+    assert [z.zone_id for z in r2.applied] == ["z2"]
+
+
+def test_run_loop_one_camera_glitch_does_not_stop_the_other():
+    class Boom(_RT):
+        def step(self, now):
+            self.steps += 1
+            if self.steps == 2:
+                raise RuntimeError("cam glitch")
+            return []
+    r1, r2 = Boom("cam-01"), _RT("cam-02")
+    counters = run_loop([r1, r2], _Client(_cfg_two_cams()), _Q(),
+                        clock=lambda: 0.0, sleep=lambda s: None, iterations=3,
+                        sync_every=99, heartbeat_every=99, flush_every=99)
+    assert r1.steps == 3 and r2.steps == 3   # both kept going
+    assert counters["errors"] == 1
+
+
+def test_parse_cameras_multi_and_single():
+    assert parse_cameras({"CAMERAS": "csi:0:cam-01, uvc:1:cam-02"}, "def") == \
+        [("csi", "0", "cam-01"), ("uvc", "1", "cam-02")]
+    assert parse_cameras({"CAMERA_BACKEND": "csi", "CSI_DEVICE": "2"}, "cam-09") == \
+        [("csi", "2", "cam-09")]
+    assert parse_cameras({}, "cam-01") == [("uvc", "0", "cam-01")]
+
+
+def test_parse_cameras_rejects_blank_entries_never_zero_cameras():
+    import pytest
+    # A non-empty-but-all-blank CAMERAS must fail loudly, never yield zero cameras
+    # (a safety daemon monitoring nothing while looking healthy).
+    with pytest.raises(ValueError):
+        parse_cameras({"CAMERAS": " , "}, "cam-01")
+    with pytest.raises(ValueError):
+        parse_cameras({"CAMERAS": "csi:0"}, "cam-01")   # wrong arity