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 9089d05c..09641ee5 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..7d2c999e 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, 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), } 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..c5d31615 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,51 @@ """ -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 @@ -40,8 +57,25 @@ # 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) 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 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..ea3bb1ac --- /dev/null +++ b/pyro_camera_api/tests/test_stuck_detector.py @@ -0,0 +1,169 @@ +# 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 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 + + +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.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}" + 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, 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) + # 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)