move_detect/edge/daemon.py

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 cloud 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.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))
    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 = CloudClient(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))