import os # Rate-limit for the swallowed-exception log: a camera failing every frame must be VISIBLE # (not a silent green unit) but must not spam the journal. Log the 1st of a run of consecutive # failures, then every LOG_EVERY-th, and again after any recovery. _LOG_EVERY = 300 def _default_log(msg, exc=None): # pragma: no cover (thin stderr wrapper) import sys import traceback print(msg, file=sys.stderr) if exc is not None: traceback.print_exception(type(exc), exc, exc.__traceback__, file=sys.stderr) 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, log=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 and rate-limit-logged, # never fatal — but it is no longer SILENT (a permanently broken pipeline used to look # perfectly green). 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) if log is None: log = _default_log counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0} streak = {} # id(rt)/"periodic" -> consecutive-error count, for rate-limited logging def note(key, label, exc): counters["errors"] += 1 n = streak.get(key, 0) + 1 streak[key] = n if n == 1 or n % _LOG_EVERY == 0: log(f"{label}: {type(exc).__name__}: {exc} (consecutive #{n})", exc) beat_anchor = None # (wall_time, frames) at the previous heartbeat, for measured fps i = 0 while iterations is None or i < iterations: i += 1 now = clock() # one timestamp per tick, shared across cameras if beat_anchor is None: beat_anchor = (now, 0) for rt in runtimes: try: rt.step(now) counters["frames"] += 1 streak.pop(id(rt), None) # recovered — reset the streak except Exception as e: cam = getattr(rt, "camera_id", "?") note(id(rt), f"camera {cam} step failed", e) 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: # Measured fps over the window since the last beat, and per-camera health, so # the server can tell a live device from a degraded/dead one (was hardcoded 0.0 # + a frames counter that ticked even when no frame was read). t0, f0 = beat_anchor elapsed = now - t0 fps = (counters["frames"] - f0) / elapsed if elapsed > 0 else 0.0 erroring = sum(1 for rt in runtimes if id(rt) in streak) metrics = {"fps": round(fps, 2), "frames": counters["frames"], "cameras_total": len(runtimes), "cameras_erroring": erroring, "errors": counters["errors"]} if client.send_heartbeat(metrics): counters["beats"] += 1 beat_anchor = (now, counters["frames"]) if flush_every and i % flush_every == 0: queue.flush() counters["flushes"] += 1 streak.pop("periodic", None) except Exception as e: note("periodic", "periodic op (sync/heartbeat/flush) failed", e) 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 build_runtimes(cameras, make_source, make_runtime, log=None): """Build one (source -> runtime) per camera. A camera that fails to CONSTRUCT (e.g. an unseated/absent CSI camera whose Picamera2(...) raises) is logged and SKIPPED so it never crash-loops the whole daemon and the other cameras still run. Raises if no camera comes up (never run monitoring nothing).""" if log is None: import sys log = lambda m: print(m, file=sys.stderr) runtimes = [] for backend, device, cam in cameras: try: runtimes.append(make_runtime(cam, make_source(backend, device))) except Exception as e: log(f"camera {cam} ({backend}:{device}) failed to start, skipping: {e}") if not runtimes: raise RuntimeError("no cameras could be started") return runtimes def main() -> None: # pragma: no cover (integration entrypoint, exercised on device) import time from dataclasses import replace from edge.config import (frame_interval_from_env, load_config, queue_db_path, require_upload_config) from edge.serverclient import ServerClient 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)) require_upload_config(cfg) # fail loud now, not silently upload nothing forever nonce = lambda: os.urandom(8).hex() now = time.time base = cfg.cafe24_url.rstrip("/") client = ServerClient(base, cfg.device_id, cfg.secret, nonce, now) post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now) queue = OfflineQueue(queue_db_path(dict(os.environ)), post) pose = build_pose() # shared across cameras def make_runtime(cam, source): # 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)) return PipelineRuntime(cam_cfg, source, pose, queue) # A missing/unseated camera is skipped (logged), not fatal — the others still run. runtimes = build_runtimes(parse_cameras(dict(os.environ), cfg.camera_id), build_source, make_runtime) run_loop(runtimes, client, queue, clock=now, sleep=time.sleep, iterations=None, frame_interval=frame_interval_from_env(dict(os.environ)), zones_for=lambda c, rt: zones_from_config(c, rt.camera_id))