move_detect/edge/probe.py

"""Camera probe: open every configured camera, grab a frame, report what came back.

Run by deploy/setup_online.sh BEFORE the service is enabled, so a miswired ribbon or a
USB camera that enumerated on a different /dev/video index fails during the install --
while an operator is still standing at the Pi -- instead of becoming a Restart=always
unit that monitors one camera less and looks perfectly healthy on the dashboard.

Standalone too:
    /opt/move-detect/venv/bin/python -m edge.probe                 # uses CAMERAS from env
    CAMERAS=uvc:2:cam-02 python -m edge.probe --save /tmp/probe    # write preview PNGs
"""

from __future__ import annotations

import argparse
import os
import sys
import time
from dataclasses import dataclass

from edge.daemon import parse_cameras

# A camera can open and still hand back nothing for the first few reads while the sensor
# settles (UVC auto-exposure, CSI pipeline start). Retry rather than fail a healthy camera.
_READ_ATTEMPTS = 8
_READ_DELAY_S = 0.25


@dataclass(frozen=True)
class ProbeResult:
    camera_id: str
    backend: str
    device: str
    ok: bool
    detail: str


def _describe(frame) -> str:
    shape = getattr(frame, "shape", None)
    if not shape or len(shape) < 2:
        return "frame of unknown shape"
    h, w = shape[0], shape[1]
    ch = shape[2] if len(shape) > 2 else 1
    return f"{w}x{h}, {ch} channel(s)"


def _default_build_source(backend: str, device: str):
    """Same construction the daemon uses (edge/daemon.py build_source)."""
    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 probe_one(backend: str, device: str, camera_id: str, *, build_source=None,
              save_dir: str | None = None) -> ProbeResult:
    """Open one camera, read a frame, optionally save a preview PNG."""
    build_source = build_source or _default_build_source
    try:
        source = build_source(backend, device)
    except Exception as exc:  # device missing, wrong index, picamera2 not installed
        return ProbeResult(camera_id, backend, device, False, f"open failed: {exc}")

    try:
        for attempt in range(_READ_ATTEMPTS):
            ok, frame = source.read()
            if ok and frame is not None:
                detail = _describe(frame)
                if save_dir:
                    detail += f"; {_save_preview(frame, save_dir, camera_id)}"
                return ProbeResult(camera_id, backend, device, True, detail)
            if attempt < _READ_ATTEMPTS - 1:
                time.sleep(_READ_DELAY_S)
        return ProbeResult(camera_id, backend, device, False,
                           f"opened but returned no frame in {_READ_ATTEMPTS} reads")
    finally:
        close = getattr(source, "close", None) or getattr(source, "release", None)
        if callable(close):
            try:
                close()
            except Exception:
                pass  # a probe must never fail the install on a teardown hiccup


def _save_preview(frame, save_dir: str, camera_id: str) -> str:
    try:
        import cv2
        os.makedirs(save_dir, exist_ok=True)
        path = os.path.join(save_dir, f"{camera_id}.png")
        cv2.imwrite(path, frame)
        return f"preview {path}"
    except Exception as exc:
        return f"preview failed: {exc}"


def probe_all(env=None, *, build_source=None, save_dir: str | None = None) -> list[ProbeResult]:
    env = dict(os.environ if env is None else env)
    cams = parse_cameras(env, env.get("DEVICE_ID", "cam-01"))
    return [probe_one(be, dev, cid, build_source=build_source, save_dir=save_dir)
            for be, dev, cid in cams]


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="probe every camera in CAMERAS")
    ap.add_argument("--save", metavar="DIR", help="write a preview PNG per camera")
    args = ap.parse_args(argv)

    try:
        results = probe_all(save_dir=args.save)
    except ValueError as exc:            # malformed CAMERAS -- a config error, not a camera fault
        print(f"CAMERAS is invalid: {exc}", file=sys.stderr)
        return 2

    for r in results:
        mark = "OK  " if r.ok else "FAIL"
        print(f"  [{mark}] {r.camera_id}  {r.backend}:{r.device}  -- {r.detail}")

    failed = [r for r in results if not r.ok]
    if failed:
        print(f"{len(failed)}/{len(results)} camera(s) failed.", file=sys.stderr)
        if any(r.backend == "uvc" for r in failed):
            print("  USB cameras enumerate in boot order -- check `v4l2-ctl --list-devices` "
                  "and fix the index in CAMERAS.", file=sys.stderr)
        if any(r.backend == "csi" for r in failed):
            print("  CSI: check the ribbon seating/orientation and `rpicam-hello --list-cameras`.",
                  file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())