From 184c2dd5dcf7fc6945d040c703651db1f14679bc Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 26 Oct 2026 19:46:03 +0100 Subject: [PATCH 1/5] feat(camera-api): detect frozen cameras and missing images, cap reboots --- .../pyro_camera_api/camera/stuck_detector.py | 233 ++++++++++-------- pyro_camera_api/tests/test_stuck_detector.py | 95 +++++++ 2 files changed, 231 insertions(+), 97 deletions(-) create mode 100644 pyro_camera_api/tests/test_stuck_detector.py diff --git a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py index 47668d62..ae5bf3c6 100644 --- a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py +++ b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py @@ -5,34 +5,48 @@ """ -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 logging +import os import threading -from typing import Dict, List +import time +from collections import deque +from datetime import datetime, timezone +from typing import Any, Deque, Dict, List, Optional, Tuple import cv2 import numpy as np from PIL import Image 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 @@ -40,8 +54,19 @@ # 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. +# ponytail: in-memory budget, a container restart resets it; persist under data/ if restarts loop +MAX_REBOOTS_PER_DAY = 3 +REBOOT_COOLDOWN = 30 * 60.0 +DAY = 24 * 3600.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 +RECOVERY_STATE: Dict[str, Dict[str, Any]] = {} + +# 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 +100,129 @@ 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 recovery_status(camera_ip: str) -> Dict[str, Any]: + state = RECOVERY_STATE.get(camera_ip) + if state is None: + return {"status": None, "reboots_24h": 0, "last_reboot": None} + now = time.monotonic() + return { + "status": state["status"], + "reboots_24h": sum(1 for t in state["attempts"] if now - t <= DAY), + "last_reboot": state["last_reboot"], + } + + 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" + state: Dict[str, Any] = {"status": "ok", "attempts": deque(), "last_reboot": 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), - ) + snap = snapshot(dict(cam.last_images), dict(cam.last_image_at)) + if prev is None: + prev = snap + state["status"] = "ok" continue + ptz_images = [img for pose, img in cam.last_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)", + 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 + + state["status"] = problem or "ok" + if problem is None: + if hits: + logger.info("[%s] Stuck detector cleared", camera_ip) + hits = 0 + continue + + 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 + if not reboot_allowed(state["attempts"], time.monotonic()): + logger.error( + "[%s] Camera %s, reboot budget spent (%d per 24 h, %d min apart)", camera_ip, - mean_var, - int(MIN_MEAN_VARIANCE_FOR_CHECK), + problem, + MAX_REBOOTS_PER_DAY, + int(REBOOT_COOLDOWN // 60), ) - CONSECUTIVE_HITS[camera_ip] = 0 continue + logger.error("[%s] Rebooting camera, %s", camera_ip, problem) + state["attempts"].append(time.monotonic()) + state["last_reboot"] = datetime.now(timezone.utc).isoformat() + ok = False try: - max_dist = _max_pairwise_hamming(images) + ok = bool(cam.reboot_camera()) except Exception as exc: - logger.warning("[%s] Stuck check failed: %s", camera_ip, exc) - 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: - logger.error( - "[%s] Rebooting camera due to stuck PTZ detection (max distance=%d)", - camera_ip, - max_dist, - ) - 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 + logger.error("[%s] Reboot raised: %s", camera_ip, exc) + if ok: + for pose in list(cam.last_images): + cam.last_images[pose] = None 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) diff --git a/pyro_camera_api/tests/test_stuck_detector.py b/pyro_camera_api/tests/test_stuck_detector.py new file mode 100644 index 00000000..5ec7d826 --- /dev/null +++ b/pyro_camera_api/tests/test_stuck_detector.py @@ -0,0 +1,95 @@ +# Copyright (C) 2022-2026, Pyronear. + +# This program is licensed under the Apache License 2.0. +# See LICENSE or go to for full license details. + +import threading +import time +from collections import deque +from unittest.mock import MagicMock + +import numpy as np +import pytest +from PIL import Image + +from pyro_camera_api.camera import stuck_detector as sd +from pyro_camera_api.camera.adapters.mock import MockCamera +from pyro_camera_api.camera.registry import CAMERA_REGISTRY + + +def _noise(seed): + return Image.fromarray(np.random.default_rng(seed).integers(0, 255, (64, 64, 3), dtype=np.uint8)) + + +def test_detect_no_fresh_image(): + prev = sd.snapshot({0: _noise(0)}, {0: 1.0}) + assert sd.detect(prev, prev, []) == "no_fresh_image" + # A camera that never produced an image is caught too + assert sd.detect({}, {}, []) == "no_fresh_image" + + +def test_detect_frozen_same_pixels_stored_again(): + # Each pose keeps its own frozen image: the Epinay August case, not seen by pHash + prev = sd.snapshot({0: _noise(0), 1: _noise(1)}, {0: 1.0, 1: 1.0}) + snap = sd.snapshot({0: _noise(0), 1: _noise(1)}, {0: 2.0, 1: 1.0}) + assert sd.detect(prev, snap, []) == "frozen" + + +def test_detect_healthy_and_not_moving(): + prev = sd.snapshot({0: _noise(0), 1: _noise(1)}, {0: 1.0, 1: 1.0}) + images = {0: _noise(2), 1: _noise(3)} + snap = sd.snapshot(images, {0: 2.0, 1: 2.0}) + assert sd.detect(prev, snap, list(images.values())) is None + # Live pixels but every pose shows the same view: the turret does not move + same_view = [_noise(2), _noise(2)] + assert sd.detect(prev, snap, same_view) == "not_moving" + + +def test_reboot_budget(): + attempts: deque[float] = deque() + assert sd.reboot_allowed(attempts, 0.0) + attempts.append(0.0) + assert not sd.reboot_allowed(attempts, sd.REBOOT_COOLDOWN - 1) + for t in (sd.REBOOT_COOLDOWN, 2 * sd.REBOOT_COOLDOWN): + assert sd.reboot_allowed(attempts, t) + attempts.append(t) + assert not sd.reboot_allowed(attempts, 10 * sd.REBOOT_COOLDOWN) + # The oldest attempt leaves the 24 h window + assert sd.reboot_allowed(attempts, sd.DAY + 1) + + +@pytest.mark.parametrize("allow_still", [False, True]) +def test_frozen_camera_is_rebooted_once_unless_allowed(monkeypatch, allow_still): + cam_id = f"frozen-{allow_still}" + cam = MockCamera(camera_id=cam_id, cam_type="static") + cam.reboot_camera = MagicMock(return_value=True) + CAMERA_REGISTRY[cam_id] = cam + monkeypatch.setattr(sd, "CHECK_INTERVAL", 0.02) + monkeypatch.setattr(sd, "ALLOW_STILL_CAMERA", allow_still) + monkeypatch.setattr(sd, "_patrol_is_running", lambda _ip: True) + monkeypatch.setattr(sd, "is_camera_streaming", lambda _ip: False) + + stop = threading.Event() + frozen = _noise(0) + + def frozen_patrol(): + while not stop.wait(0.005): + cam.store_image(-1, frozen) + + threads = [ + threading.Thread(target=frozen_patrol, daemon=True), + threading.Thread(target=sd.stuck_check_loop, args=(cam_id, stop), daemon=True), + ] + try: + for thread in threads: + thread.start() + time.sleep(0.5) + finally: + stop.set() + for thread in threads: + thread.join(timeout=2.0) + CAMERA_REGISTRY.pop(cam_id, None) + + # The cooldown keeps it to a single attempt + assert cam.reboot_camera.call_count == (0 if allow_still else 1) + assert sd.recovery_status(cam_id)["reboots_24h"] == (0 if allow_still else 1) From 578ba4717247dc6b63cb3539ed8c52fde8f37eb9 Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 26 Oct 2026 19:46:03 +0100 Subject: [PATCH 2/5] feat(camera-api): run the stuck detector on every camera, add ALLOW_STILL_CAMERA --- .env.example | 7 ++++++- docker-compose.yml | 2 ++ .../pyro_camera_api/api/routes_patrol.py | 3 +++ pyro_camera_api/pyro_camera_api/main.py | 13 +++++++++---- 4 files changed, 20 insertions(+), 5 deletions(-) diff --git a/.env.example b/.env.example index a40d4113..2e1e3abe 100644 --- a/.env.example +++ b/.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 diff --git a/docker-compose.yml b/docker-compose.yml index 951773ec..656e3558 100644 --- a/docker-compose.yml +++ b/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 diff --git a/pyro_camera_api/pyro_camera_api/api/routes_patrol.py b/pyro_camera_api/pyro_camera_api/api/routes_patrol.py index 2af9695c..a93460fe 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_patrol.py +++ b/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, reboots in the last 24 h, last reboot (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), } diff --git a/pyro_camera_api/pyro_camera_api/main.py b/pyro_camera_api/pyro_camera_api/main.py index e5081d98..311b56cd 100644 --- a/pyro_camera_api/pyro_camera_api/main.py +++ b/pyro_camera_api/pyro_camera_api/main.py @@ -29,7 +29,7 @@ STUCK_CHECK_FLAGS, STUCK_CHECK_THREADS, ) -from pyro_camera_api.camera.stuck_detector import stuck_check_loop +from pyro_camera_api.camera.stuck_detector import ALLOW_STILL_CAMERA, stuck_check_loop from pyro_camera_api.core.logging import setup_logging from pyro_camera_api.services.anonymizer_rtsp import AnonymizerWorker, BoxStore, LastFrameStore from pyro_camera_api.services.stream import set_app_for_stream, stop_stream_if_idle @@ -81,16 +81,21 @@ async def lifespan(app: FastAPI): "yes", "on", ) - if stuck_detector_enabled and getattr(cam, "cam_type", "static") == "ptz" and hasattr(cam, "reboot_camera"): + if stuck_detector_enabled: stuck_flag = threading.Event() stuck_thread = threading.Thread(target=stuck_check_loop, args=(cam_id, stuck_flag), daemon=True) STUCK_CHECK_THREADS[cam_id] = stuck_thread STUCK_CHECK_FLAGS[cam_id] = stuck_flag stuck_thread.start() - logger.info("Starting stuck detector for PTZ camera %s", cam_id) - elif not stuck_detector_enabled and getattr(cam, "cam_type", "static") == "ptz": + logger.info("Starting stuck detector for camera %s", cam_id) + else: logger.info("Stuck detector disabled by ENABLE_STUCK_DETECTOR for %s", cam_id) + if ALLOW_STILL_CAMERA: + logger.warning( + "ALLOW_STILL_CAMERA is on: frozen or not moving cameras are never rebooted. Do not use on a station." + ) + threading.Thread(target=stop_stream_if_idle, daemon=True).start() # Resolve pose azimuths from the platform API in the background (retries From da60cc62a417cbaaaa34751dd82b3f9e26fc6a45 Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 26 Oct 2026 19:54:11 +0100 Subject: [PATCH 3/5] fix(camera-api): read stored images from one copy, report reboot attempts --- .../pyro_camera_api/api/routes_patrol.py | 2 +- .../pyro_camera_api/camera/stuck_detector.py | 16 +++++++++------- pyro_camera_api/tests/test_stuck_detector.py | 2 +- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/pyro_camera_api/pyro_camera_api/api/routes_patrol.py b/pyro_camera_api/pyro_camera_api/api/routes_patrol.py index a93460fe..7d2c999e 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_patrol.py +++ b/pyro_camera_api/pyro_camera_api/api/routes_patrol.py @@ -111,7 +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, reboots in the last 24 h, last reboot (UTC) + - 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. """ diff --git a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py index ae5bf3c6..a314afad 100644 --- a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py +++ b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py @@ -134,12 +134,12 @@ def reboot_allowed(attempts: Deque[float], now: float) -> bool: def recovery_status(camera_ip: str) -> Dict[str, Any]: state = RECOVERY_STATE.get(camera_ip) if state is None: - return {"status": None, "reboots_24h": 0, "last_reboot": None} + return {"status": None, "reboot_attempts_24h": 0, "last_reboot_attempt": None} now = time.monotonic() return { "status": state["status"], - "reboots_24h": sum(1 for t in state["attempts"] if now - t <= DAY), - "last_reboot": state["last_reboot"], + "reboot_attempts_24h": sum(1 for t in state["attempts"] if now - t <= DAY), + "last_reboot_attempt": state["last_reboot_attempt"], } @@ -147,7 +147,7 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: cam = CAMERA_REGISTRY[camera_ip] can_reboot = hasattr(cam, "reboot_camera") is_ptz = getattr(cam, "cam_type", "static") == "ptz" - state: Dict[str, Any] = {"status": "ok", "attempts": deque(), "last_reboot": None} + state: Dict[str, Any] = {"status": "ok", "attempts": deque(), "last_reboot_attempt": None} RECOVERY_STATE[camera_ip] = state logger.info( @@ -168,13 +168,15 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: state["status"] = "idle" continue - snap = snapshot(dict(cam.last_images), dict(cam.last_image_at)) + # 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 cam.last_images.items() if is_ptz and pose != -1 and img is not None] + ptz_images = [img for pose, img in images.items() if is_ptz and pose != -1 and img is not None] try: problem = detect(prev, snap, ptz_images) except Exception as exc: @@ -212,7 +214,7 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: logger.error("[%s] Rebooting camera, %s", camera_ip, problem) state["attempts"].append(time.monotonic()) - state["last_reboot"] = datetime.now(timezone.utc).isoformat() + state["last_reboot_attempt"] = datetime.now(timezone.utc).isoformat() ok = False try: ok = bool(cam.reboot_camera()) diff --git a/pyro_camera_api/tests/test_stuck_detector.py b/pyro_camera_api/tests/test_stuck_detector.py index 5ec7d826..c2354808 100644 --- a/pyro_camera_api/tests/test_stuck_detector.py +++ b/pyro_camera_api/tests/test_stuck_detector.py @@ -92,4 +92,4 @@ def frozen_patrol(): # The cooldown keeps it to a single attempt assert cam.reboot_camera.call_count == (0 if allow_still else 1) - assert sd.recovery_status(cam_id)["reboots_24h"] == (0 if allow_still else 1) + assert sd.recovery_status(cam_id)["reboot_attempts_24h"] == (0 if allow_still else 1) From 13c529dfdcc9bdae5e0d58206b0544c2b8e7e7c2 Mon Sep 17 00:00:00 2001 From: Mateo Lostanlen Date: Wed, 28 Oct 2026 12:26:14 +0100 Subject: [PATCH 4/5] fix(camera-api): persist the reboot budget, hide the camera password, grace after reboot - Reboot attempts were in memory with monotonic times, so every container restart (healthcheck, update) handed out three new attempts. They are now wall-clock times saved in data/camera_recovery.json (RECOVERY_STATE_FILE) and loaded at startup, pruned to the last 24 h. - `Reboot raised: ` logged the Reolink URL with the camera password. Only the exception type is logged. - A static camera rebooted by the detector failed its next two captures while booting and entered the 30 min skip window. The detector now resets the failure count and gives it 90 s before captures count again. - /patrol_status iterated the attempts deque while the detector appended to it: both sides now hold a lock. - "reboot budget spent" is logged once per episode instead of every 5 min. --- .../pyro_camera_api/camera/stuck_detector.py | 77 +++++++++++++++---- pyro_camera_api/tests/test_stuck_detector.py | 67 +++++++++++++++- 2 files changed, 127 insertions(+), 17 deletions(-) diff --git a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py index a314afad..f39e577b 100644 --- a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py +++ b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py @@ -29,18 +29,21 @@ from __future__ import annotations import hashlib +import json import logging import os import threading 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 @@ -54,16 +57,22 @@ # 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. -# ponytail: in-memory budget, a container restart resets it; persist under data/ if restarts loop +# 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") -# Per camera recovery state, exposed by /patrol_status +# 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]] @@ -131,14 +140,40 @@ def reboot_allowed(attempts: Deque[float], now: float) -> bool: 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.monotonic() + now = time.time() + with _LOCK: + attempts = list(state["attempts"]) return { "status": state["status"], - "reboot_attempts_24h": sum(1 for t in state["attempts"] if now - t <= DAY), + "reboot_attempts_24h": sum(1 for t in attempts if now - t <= DAY), "last_reboot_attempt": state["last_reboot_attempt"], } @@ -147,7 +182,7 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: cam = CAMERA_REGISTRY[camera_ip] can_reboot = hasattr(cam, "reboot_camera") is_ptz = getattr(cam, "cam_type", "static") == "ptz" - state: Dict[str, Any] = {"status": "ok", "attempts": deque(), "last_reboot_attempt": None} + state: Dict[str, Any] = {"status": "ok", "attempts": load_attempts(camera_ip), "last_reboot_attempt": None} RECOVERY_STATE[camera_ip] = state logger.info( @@ -202,27 +237,37 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: if not can_reboot: logger.error("[%s] Camera %s and its adapter cannot reboot it", camera_ip, problem) continue - if not reboot_allowed(state["attempts"], time.monotonic()): - logger.error( - "[%s] Camera %s, reboot budget spent (%d per 24 h, %d min apart)", - camera_ip, - problem, - MAX_REBOOTS_PER_DAY, - int(REBOOT_COOLDOWN // 60), - ) + 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] Camera %s, reboot budget spent (%d per 24 h, %d min apart)", + camera_ip, + problem, + MAX_REBOOTS_PER_DAY, + int(REBOOT_COOLDOWN // 60), + ) continue logger.error("[%s] Rebooting camera, %s", camera_ip, problem) - state["attempts"].append(time.monotonic()) + 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: - logger.error("[%s] Reboot raised: %s", camera_ip, 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: logger.error("[%s] Reboot command rejected by camera", camera_ip) prev, hits = None, 0 diff --git a/pyro_camera_api/tests/test_stuck_detector.py b/pyro_camera_api/tests/test_stuck_detector.py index c2354808..76cf5d04 100644 --- a/pyro_camera_api/tests/test_stuck_detector.py +++ b/pyro_camera_api/tests/test_stuck_detector.py @@ -12,6 +12,7 @@ import pytest from PIL import Image +from pyro_camera_api.camera import patrol from pyro_camera_api.camera import stuck_detector as sd from pyro_camera_api.camera.adapters.mock import MockCamera from pyro_camera_api.camera.registry import CAMERA_REGISTRY @@ -58,6 +59,69 @@ def test_reboot_budget(): assert sd.reboot_allowed(attempts, sd.DAY + 1) +@pytest.fixture(autouse=True) +def state_file(tmp_path, monkeypatch): + monkeypatch.setattr(sd, "STATE_FILE", str(tmp_path / "recovery.json")) + return tmp_path / "recovery.json" + + +def test_attempts_survive_a_restart(state_file): + now = time.time() + sd.save_attempts("cam", deque([now - sd.DAY - 1, now - 60])) + # Only the attempt of the last 24 h is loaded, the old one is pruned + assert sd.load_attempts("cam") == deque([now - 60]) + assert sd.load_attempts("other") == deque() + state_file.write_text("not json") + assert sd.load_attempts("cam") == deque() + + +def _run_frozen_camera(monkeypatch, cam_id, reboot, allow_still=False, seconds=0.5): + cam = MockCamera(camera_id=cam_id, cam_type="static") + cam.reboot_camera = reboot + CAMERA_REGISTRY[cam_id] = cam + monkeypatch.setattr(sd, "CHECK_INTERVAL", 0.02) + monkeypatch.setattr(sd, "ALLOW_STILL_CAMERA", allow_still) + monkeypatch.setattr(sd, "_patrol_is_running", lambda _ip: True) + monkeypatch.setattr(sd, "is_camera_streaming", lambda _ip: False) + + stop = threading.Event() + frozen = _noise(0) + + def frozen_patrol(): + while not stop.wait(0.005): + cam.store_image(-1, frozen) + + threads = [ + threading.Thread(target=frozen_patrol, daemon=True), + threading.Thread(target=sd.stuck_check_loop, args=(cam_id, stop), daemon=True), + ] + try: + for thread in threads: + thread.start() + time.sleep(seconds) + finally: + stop.set() + for thread in threads: + thread.join(timeout=2.0) + CAMERA_REGISTRY.pop(cam_id, None) + return cam + + +def test_reboot_error_does_not_log_the_camera_password(monkeypatch, caplog): + reboot = MagicMock(side_effect=ConnectionError("https://cam/api.cgi?user=admin&password=s3cret")) + _run_frozen_camera(monkeypatch, "leak", reboot) + assert reboot.called + assert "Reboot raised ConnectionError" in caplog.text + assert "s3cret" not in caplog.text + + +def test_rebooted_camera_gets_a_grace_before_the_skip_window(monkeypatch): + patrol.FAILURE_COUNT["grace"] = 1 + _run_frozen_camera(monkeypatch, "grace", MagicMock(return_value=True)) + assert patrol.FAILURE_COUNT["grace"] == 0 + assert time.time() < patrol.SKIP_UNTIL["grace"] <= time.time() + sd.REBOOT_GRACE + + @pytest.mark.parametrize("allow_still", [False, True]) def test_frozen_camera_is_rebooted_once_unless_allowed(monkeypatch, allow_still): cam_id = f"frozen-{allow_still}" @@ -90,6 +154,7 @@ def frozen_patrol(): thread.join(timeout=2.0) CAMERA_REGISTRY.pop(cam_id, None) - # The cooldown keeps it to a single attempt + # The cooldown keeps it to a single attempt, saved for the next run assert cam.reboot_camera.call_count == (0 if allow_still else 1) assert sd.recovery_status(cam_id)["reboot_attempts_24h"] == (0 if allow_still else 1) + assert len(sd.load_attempts(cam_id)) == (0 if allow_still else 1) From a4a683a926d867eaa617b7fc26b7a6af2f5cba59 Mon Sep 17 00:00:00 2001 From: Mateo Lostanlen Date: Wed, 28 Oct 2026 12:56:51 +0100 Subject: [PATCH 5/5] fix(camera-api): report the saved reboot attempt after a restart /patrol_status showed one attempt in the last 24 h but no last_reboot_attempt once the detector restarted: derive it from the attempts loaded from disk. --- pyro_camera_api/pyro_camera_api/camera/stuck_detector.py | 7 ++++++- pyro_camera_api/tests/test_stuck_detector.py | 9 +++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py index f39e577b..c5d31615 100644 --- a/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py +++ b/pyro_camera_api/pyro_camera_api/camera/stuck_detector.py @@ -182,7 +182,12 @@ def stuck_check_loop(camera_ip: str, stop_flag: threading.Event) -> None: cam = CAMERA_REGISTRY[camera_ip] can_reboot = hasattr(cam, "reboot_camera") is_ptz = getattr(cam, "cam_type", "static") == "ptz" - state: Dict[str, Any] = {"status": "ok", "attempts": load_attempts(camera_ip), "last_reboot_attempt": None} + 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( diff --git a/pyro_camera_api/tests/test_stuck_detector.py b/pyro_camera_api/tests/test_stuck_detector.py index 76cf5d04..ea3bb1ac 100644 --- a/pyro_camera_api/tests/test_stuck_detector.py +++ b/pyro_camera_api/tests/test_stuck_detector.py @@ -158,3 +158,12 @@ def frozen_patrol(): assert cam.reboot_camera.call_count == (0 if allow_still else 1) assert sd.recovery_status(cam_id)["reboot_attempts_24h"] == (0 if allow_still else 1) assert len(sd.load_attempts(cam_id)) == (0 if allow_still else 1) + # A restarted detector reports the saved attempt + CAMERA_REGISTRY[cam_id] = cam + stop = threading.Event() + stop.set() + sd.stuck_check_loop(cam_id, stop) + CAMERA_REGISTRY.pop(cam_id, None) + status = sd.recovery_status(cam_id) + assert status["reboot_attempts_24h"] == (0 if allow_still else 1) + assert (status["last_reboot_attempt"] is not None) == (not allow_still)