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, 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 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: 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"]}): counters["beats"] += 1 if flush_every and i % flush_every == 0: queue.flush() counters["flushes"] += 1 except Exception: counters["errors"] += 1 if frame_interval: sleep(frame_interval) 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() now = time.time base = cfg.cafe24_url.rstrip("/") 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) 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))