diff --git a/deploy/README.md b/deploy/README.md index 808581a..e164b6a 100644 --- a/deploy/README.md +++ b/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 diff --git a/deploy/systemd/move-detect-edge.env.example b/deploy/systemd/move-detect-edge.env.example index b152ba0..9ed4b81 100644 --- a/deploy/systemd/move-detect-edge.env.example +++ b/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 diff --git a/edge/daemon.py b/edge/daemon.py index 256c176..036f2bd 100644 --- a/edge/daemon.py +++ b/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)) diff --git a/edge/runtime.py b/edge/runtime.py index 7b76a59..fc0d1bd 100644 --- a/edge/runtime.py +++ b/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 diff --git a/edge/zones.py b/edge/zones.py index 23718ff..8c49342 100644 --- a/edge/zones.py +++ b/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 diff --git a/tests/test_multicam.py b/tests/test_multicam.py new file mode 100644 index 0000000..4fc6fb9 --- /dev/null +++ b/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