feat(camera-api): reboot frozen or silent cameras, with a daily budget

.env.example

@@ -16,6 +16,11 @@ MEDIAMTX_SERVER_IP=1.2.3.4
 # Set to a specific version for pinned deployments, e.g. PYRO_ENGINE_VERSION=1.2.0
 PYRO_ENGINE_VERSION=latest
 
-# Stuck-PTZ detector (auto-reboots a PTZ camera that stops rotating during patrol).
+# Stuck-camera detector: reboots a camera through its API when it stops giving fresh
+# images, freezes, or (PTZ) stops rotating during patrol. At most 3 reboots per day.
 # Enabled by default. Set to "false" to disable.
 ENABLE_STUCK_DETECTOR=true
+
+# Testing only: never reboot a camera that looks frozen or does not move, to work
+# with a still PTZ at home or a mock camera. Logged as a warning at startup.
+ALLOW_STILL_CAMERA=false

docker-compose.yml

@@ -13,6 +13,8 @@ services:
       CAM_PWD: ${CAM_PWD}
       MEDIAMTX_SERVER_IP: ${MEDIAMTX_SERVER_IP}
       ENABLE_STUCK_DETECTOR: ${ENABLE_STUCK_DETECTOR:-true}
+      # Testing only: never reboot a camera that looks frozen or does not move
+      ALLOW_STILL_CAMERA: ${ALLOW_STILL_CAMERA:-false}
     volumes:
       - ./data:/usr/src/app/data
     restart: always

pyro_camera_api/pyro_camera_api/api/routes_patrol.py

@@ -19,6 +19,7 @@
     static_loop,
 )
 from pyro_camera_api.camera.registry import CAMERA_REGISTRY, PATROL_FLAGS, PATROL_THREADS
+from pyro_camera_api.camera.stuck_detector import recovery_status
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
@@ -110,6 +111,7 @@ def patrol_status(camera_ip: str):
         - type of patrol loop (static or preset patrol)
         - number of recent patrol failures
         - timestamp at which retry should resume (skip-until logic)
+        - stuck detector state: detected problem, reboot attempts in the last 24 h, last attempt (UTC)
 
     This endpoint is safe to call periodically from UI dashboards.
     """
@@ -131,4 +133,5 @@ def patrol_status(camera_ip: str):
         "loop_type": loop_type,
         "failures": FAILURE_COUNT.get(camera_ip, 0),
         "skip_until": int(SKIP_UNTIL.get(camera_ip, 0.0)),
+        "recovery": recovery_status(camera_ip),
     }

pyro_camera_api/pyro_camera_api/camera/stuck_detector.py

@@ -5,43 +5,77 @@
 
 
 """
-PTZ stuck-camera detector.
+Stuck-camera detector and recovery.
 
-Every CHECK_INTERVAL seconds, compute pairwise pHash distances across the
-most recent per-pose images produced by the patrol loop. A turret that has
-frozen returns near-identical frames for all poses, giving a very small
-maximum pairwise distance. After CONSECUTIVE_HITS_BEFORE_REBOOT consecutive
-low-distance checks, reboot the camera.
+Every CHECK_INTERVAL seconds, compare the images the patrol stored for a camera
+with the ones seen at the previous check:
 
-Thresholds were calibrated on real sdis-77 captures: stuck-episode max
-pairwise Hamming <= 6, working-patrol min pairwise Hamming >= 17.
+- no_fresh_image: no pose got a new image, snapshots are failing.
+- frozen: every pose that got a new image has the exact same pixels as before.
+  A live sensor never returns the same bytes twice, a frozen camera does.
+- not_moving: all PTZ poses look alike, the turret is stuck on one view.
+  Thresholds were calibrated on real sdis-77 captures: stuck-episode max
+  pairwise Hamming <= 6, working-patrol min pairwise Hamming >= 17.
+
+After CONSECUTIVE_HITS_BEFORE_REBOOT checks in a row, reboot the camera through
+its API, at most MAX_REBOOTS_PER_DAY attempts at least REBOOT_COOLDOWN apart.
+A camera that does not answer at all is power-cycled by the Shelly watchdog
+(watchdog/shelly/main_pi), not here.
+
+ALLOW_STILL_CAMERA=true ignores frozen and not_moving, to work with a camera
+that does not move (at home, or a mock). Never set it on a station.
 """
 
 from __future__ import annotations
 
+import hashlib
+import json
 import logging
+import os
 import threading
-from typing import Dict, List
+import time
+from collections import deque
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Deque, Dict, List, Optional, Tuple
 
 import cv2
 import numpy as np
 from PIL import Image
 
+from pyro_camera_api.camera import patrol
 from pyro_camera_api.camera.registry import CAMERA_REGISTRY, PATROL_FLAGS, PATROL_THREADS
+from pyro_camera_api.services.stream import is_camera_streaming
 
 logger = logging.getLogger(__name__)
 
-CHECK_INTERVAL = 30 * 60.0  # seconds between checks
-INITIAL_DELAY = 3 * 60.0  # delay before the first check, lets patrol populate last_images
+CHECK_INTERVAL = 5 * 60.0  # seconds between checks
 STUCK_MAX_HAMMING = 10  # max pairwise distance below which we suspect stuck
 CONSECUTIVE_HITS_BEFORE_REBOOT = 2
 MIN_POSES_FOR_CHECK = 2
 # Fog / low-light scenes collapse to near-uniform gray and produce unstable pHashes.
 # Skip the check when the mean per-image gray variance falls below this threshold.
 # Calibrated on sun_test data: foggy rounds mean variance <= 332, stuck rounds mean >= 854.
 MIN_MEAN_VARIANCE_FOR_CHECK = 500.0
+# Same limits as the Shelly watchdog. Attempts are wall-clock times kept in STATE_FILE, so a
+# container restart (healthcheck, update) does not hand out a fresh budget.
+MAX_REBOOTS_PER_DAY = 3
+REBOOT_COOLDOWN = 30 * 60.0
+DAY = 24 * 3600.0
+STATE_FILE = os.getenv("RECOVERY_STATE_FILE", "data/camera_recovery.json")
+# The static loop skips a camera 30 min after two failed captures: give a rebooting camera
+# this long to come back before its captures count again.
+REBOOT_GRACE = 90.0
+
+ALLOW_STILL_CAMERA = os.getenv("ALLOW_STILL_CAMERA", "false").strip().lower() in ("1", "true", "yes", "on")
 
-CONSECUTIVE_HITS: Dict[str, int] = {}
+# Per camera recovery state, exposed by /patrol_status. The lock covers the attempts deque,
+# appended by the detector thread and read by the API.
+RECOVERY_STATE: Dict[str, Dict[str, Any]] = {}
+_LOCK = threading.Lock()
+
+# pose -> (time the image was stored, md5 of its pixels)
+Snapshot = Dict[int, Tuple[float, bytes]]
 
 
 def _phash(img: Image.Image, hash_size: int = 8, highfreq_factor: int = 4) -> np.ndarray:
@@ -75,115 +109,172 @@ def _patrol_is_running(camera_ip: str) -> bool:
     return bool(thr and thr.is_alive() and flag and not flag.is_set())
 
 
+def snapshot(images: Dict[int, Optional[Image.Image]], stored_at: Dict[int, float]) -> Snapshot:
+    return {
+        pose: (stored_at[pose], hashlib.md5(image.tobytes(), usedforsecurity=False).digest())
+        for pose, image in images.items()
+        if image is not None and pose in stored_at
+    }
+
+
+def detect(prev: Snapshot, snap: Snapshot, ptz_images: List[Image.Image]) -> Optional[str]:
+    """Return the problem seen since the previous check, or None when the camera looks fine."""
+    advanced = [pose for pose, (ts, _) in snap.items() if pose not in prev or prev[pose][0] != ts]
+    if not advanced:
+        return "no_fresh_image"
+    if all(pose in prev and prev[pose][1] == snap[pose][1] for pose in advanced):
+        return "frozen"
+    if (
+        len(ptz_images) >= MIN_POSES_FOR_CHECK
+        and _mean_gray_variance(ptz_images) >= MIN_MEAN_VARIANCE_FOR_CHECK
+        and _max_pairwise_hamming(ptz_images) < STUCK_MAX_HAMMING
+    ):
+        return "not_moving"
+    return None
+
+
+def reboot_allowed(attempts: Deque[float], now: float) -> bool:
+    """Budget on reboot attempts, a rejected or timed-out request may still reboot the camera."""
+    while attempts and now - attempts[0] > DAY:
+        attempts.popleft()
+    return len(attempts) < MAX_REBOOTS_PER_DAY and (not attempts or now - attempts[-1] >= REBOOT_COOLDOWN)
+
+
+def load_attempts(camera_ip: str) -> Deque[float]:
+    """Reboot attempts of the last 24 h saved by a previous run, none when the file is unusable."""
+    try:
+        saved = json.loads(Path(STATE_FILE).read_text()).get(camera_ip, [])
+    except (OSError, ValueError, AttributeError):
+        return deque()
+    now = time.time()
+    return deque(float(t) for t in saved if now - float(t) <= DAY)
+
+
+def save_attempts(camera_ip: str, attempts: Deque[float]) -> None:
+    try:
+        path = Path(STATE_FILE)
+        try:
+            state = json.loads(path.read_text())
+        except (OSError, ValueError):
+            state = {}
+        state[camera_ip] = list(attempts)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(state))
+    except (OSError, TypeError) as exc:
+        logger.warning("[%s] Cannot save reboot attempts to %s: %s", camera_ip, STATE_FILE, exc)
+
+
+def recovery_status(camera_ip: str) -> Dict[str, Any]:
+    state = RECOVERY_STATE.get(camera_ip)
+    if state is None:
+        return {"status": None, "reboot_attempts_24h": 0, "last_reboot_attempt": None}
+    now = time.time()
+    with _LOCK:
+        attempts = list(state["attempts"])
+    return {
+        "status": state["status"],
+        "reboot_attempts_24h": sum(1 for t in attempts if now - t <= DAY),
+        "last_reboot_attempt": state["last_reboot_attempt"],
+    }
+
+
 def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None:
     cam = CAMERA_REGISTRY[camera_ip]
-
-    if not hasattr(cam, "reboot_camera"):
-        logger.info("[%s] Stuck detector disabled: adapter does not support reboot", camera_ip)
-        return
+    can_reboot = hasattr(cam, "reboot_camera")
+    is_ptz = getattr(cam, "cam_type", "static") == "ptz"
+    attempts = load_attempts(camera_ip)
+    state: Dict[str, Any] = {
+        "status": "ok",
+        "attempts": attempts,
+        "last_reboot_attempt": datetime.fromtimestamp(attempts[-1], timezone.utc).isoformat() if attempts else None,
+    }
+    RECOVERY_STATE[camera_ip] = state
 
     logger.info(
-        "[%s] Stuck detector started (initial=%ds, interval=%ds, threshold=%d, consecutive=%d)",
+        "[%s] Stuck detector started (interval=%ds, consecutive=%d, reboot=%s)",
         camera_ip,
-        int(INITIAL_DELAY),
         int(CHECK_INTERVAL),
-        STUCK_MAX_HAMMING,
         CONSECUTIVE_HITS_BEFORE_REBOOT,
+        "api" if can_reboot else "unsupported",
     )
 
-    CONSECUTIVE_HITS[camera_ip] = 0
-    next_delay = INITIAL_DELAY
+    prev: Optional[Snapshot] = None
+    hits = 0
 
-    while not stop_flag.wait(next_delay):
-        next_delay = CHECK_INTERVAL
-        if not _patrol_is_running(camera_ip):
-            logger.info("[%s] Stuck check skipped: patrol not running", camera_ip)
-            CONSECUTIVE_HITS[camera_ip] = 0
+    while not stop_flag.wait(CHECK_INTERVAL):
+        if not _patrol_is_running(camera_ip) or is_camera_streaming(camera_ip):
+            # Night, autofocus or a live stream: no image is expected, start over from a new baseline
+            prev, hits = None, 0
+            state["status"] = "idle"
             continue
 
-        poses = [p for p in getattr(cam, "cam_poses", []) or [] if p != -1]
-        images = [im for im in (cam.last_images.get(p) for p in poses) if im is not None]
-        if len(images) < MIN_POSES_FOR_CHECK:
-            logger.info(
-                "[%s] Stuck check skipped: only %d pose images available",
-                camera_ip,
-                len(images),
-            )
+        # One copy, the patrol thread keeps storing images while we read them
+        images = dict(cam.last_images)
+        snap = snapshot(images, dict(cam.last_image_at))
+        if prev is None:
+            prev = snap
+            state["status"] = "ok"
             continue
 
+        ptz_images = [img for pose, img in images.items() if is_ptz and pose != -1 and img is not None]
         try:
-            mean_var = _mean_gray_variance(images)
+            problem = detect(prev, snap, ptz_images)
         except Exception as exc:
-            logger.warning("[%s] Stuck check failed computing variance: %s", camera_ip, exc)
+            logger.warning("[%s] Stuck check failed: %s", camera_ip, exc)
             continue
+        prev = snap
 
-        if mean_var < MIN_MEAN_VARIANCE_FOR_CHECK:
-            logger.info(
-                "[%s] Stuck check skipped: low-variance scene (mean=%.0f < %d, likely fog/night)",
-                camera_ip,
-                mean_var,
-                int(MIN_MEAN_VARIANCE_FOR_CHECK),
-            )
-            CONSECUTIVE_HITS[camera_ip] = 0
-            continue
+        if problem in ("frozen", "not_moving") and ALLOW_STILL_CAMERA:
+            logger.warning("[%s] Camera looks %s, ignored because ALLOW_STILL_CAMERA is on", camera_ip, problem)
+            problem = None
 
-        try:
-            max_dist = _max_pairwise_hamming(images)
-        except Exception as exc:
-            logger.warning("[%s] Stuck check failed: %s", camera_ip, exc)
+        state["status"] = problem or "ok"
+        if problem is None:
+            if hits:
+                logger.info("[%s] Stuck detector cleared", camera_ip)
+            hits = 0
             continue
 
-        logger.info(
-            "[%s] Stuck check: max pHash distance=%d across %d poses (threshold=%d)",
-            camera_ip,
-            max_dist,
-            len(images),
-            STUCK_MAX_HAMMING,
-        )
-
-        if max_dist < STUCK_MAX_HAMMING:
-            CONSECUTIVE_HITS[camera_ip] += 1
-            logger.warning(
-                "[%s] Possible stuck PTZ: max pHash distance=%d across %d poses (hit %d/%d)",
-                camera_ip,
-                max_dist,
-                len(images),
-                CONSECUTIVE_HITS[camera_ip],
-                CONSECUTIVE_HITS_BEFORE_REBOOT,
-            )
-            if CONSECUTIVE_HITS[camera_ip] >= CONSECUTIVE_HITS_BEFORE_REBOOT:
+        hits += 1
+        logger.warning("[%s] Camera %s (hit %d/%d)", camera_ip, problem, hits, CONSECUTIVE_HITS_BEFORE_REBOOT)
+        if hits < CONSECUTIVE_HITS_BEFORE_REBOOT:
+            continue
+        if not can_reboot:
+            logger.error("[%s] Camera %s and its adapter cannot reboot it", camera_ip, problem)
+            continue
+        with _LOCK:
+            allowed = reboot_allowed(state["attempts"], time.time())
+        if not allowed:
+            # Logged once per episode: hits keep counting silently until the cooldown ends
+            if hits == CONSECUTIVE_HITS_BEFORE_REBOOT:
                 logger.error(
-                    "[%s] Rebooting camera due to stuck PTZ detection (max distance=%d)",
+                    "[%s] Camera %s, reboot budget spent (%d per 24 h, %d min apart)",
                     camera_ip,
-                    max_dist,
+                    problem,
+                    MAX_REBOOTS_PER_DAY,
+                    int(REBOOT_COOLDOWN // 60),
                 )
-                ok = False
-                try:
-                    ok = bool(cam.reboot_camera())
-                except Exception as exc:
-                    logger.error("[%s] Reboot raised: %s", camera_ip, exc)
-
-                if ok:
-                    for p in poses:
-                        cam.last_images[p] = None
-                    CONSECUTIVE_HITS[camera_ip] = 0
-                else:
-                    logger.error(
-                        "[%s] Reboot command rejected by camera, will retry on next check",
-                        camera_ip,
-                    )
-                    # keep the hit counter so we retry, use fast-confirm cadence
-                    next_delay = INITIAL_DELAY
-            else:
-                # confirm the hit quickly rather than waiting a full interval
-                next_delay = INITIAL_DELAY
+            continue
+
+        logger.error("[%s] Rebooting camera, %s", camera_ip, problem)
+        with _LOCK:
+            state["attempts"].append(time.time())
+            save_attempts(camera_ip, state["attempts"])
+        state["last_reboot_attempt"] = datetime.now(timezone.utc).isoformat()
+        ok = False
+        try:
+            ok = bool(cam.reboot_camera())
+        except Exception as exc:
+            # The exception text carries the request URL, with the camera password in it
+            logger.error("[%s] Reboot raised %s", camera_ip, type(exc).__name__)
+        if ok:
+            for pose in list(cam.last_images):
+                cam.last_images[pose] = None
+            # Let the camera boot before its failed captures count towards the 30 min skip
+            patrol.FAILURE_COUNT[camera_ip] = 0
+            patrol.SKIP_UNTIL[camera_ip] = time.time() + REBOOT_GRACE
         else:
-            if CONSECUTIVE_HITS[camera_ip] > 0:
-                logger.info(
-                    "[%s] Stuck detector cleared: max distance=%d",
-                    camera_ip,
-                    max_dist,
-                )
-            CONSECUTIVE_HITS[camera_ip] = 0
+            logger.error("[%s] Reboot command rejected by camera", camera_ip)
+        prev, hits = None, 0
 
     logger.info("[%s] Stuck detector exited cleanly", camera_ip)