feat: temporal validation of alerts on pi

.env.example

@@ -19,3 +19,6 @@ PYRO_ENGINE_VERSION=latest
 # Stuck-PTZ detector (auto-reboots a PTZ camera that stops rotating during patrol).
 # Enabled by default. Set to "false" to disable.
 ENABLE_STUCK_DETECTOR=true
+# Optional temporal validation of alerts: both lines together, plus data/model_onnx.zip
+# COMPOSE_PROFILES=temporal
+# TEMPORAL_API_URL=http://localhost:8082

.github/workflows/build-push-image.yml

@@ -21,6 +21,9 @@ jobs:
           - image: pyronear/pyro-camera-api
             context: pyro_camera_api
             dockerfile: pyro_camera_api/Dockerfile
+          - image: pyronear/pyro-temporal-api
+            context: pyro_temporal_api
+            dockerfile: pyro_temporal_api/Dockerfile
 
     steps:
       - name: Checkout repository

.github/workflows/tests.yml

@@ -112,5 +112,5 @@ jobs:
           license: 'Apache-2.0'
           owner: 'Pyronear'
           starting-year: 2022
-          folders: 'pyroengine,docs,scripts,.github,src,pyro_camera_api,pyro-predictor/pyro_predictor'
+          folders: 'pyroengine,docs,scripts,.github,src,pyro_camera_api,pyro_temporal_api,pyro-predictor/pyro_predictor'
           ignore-files: 'version.py,__init__.py'

CLAUDE.md

@@ -4,12 +4,13 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## Project Overview
 
-PyroEngine is a wildfire detection system for edge devices (Raspberry Pi, etc.). It has two main packages:
+PyroEngine is a wildfire detection system for edge devices (Raspberry Pi, etc.). It has three packages:
 
 - **`pyroengine/`** — Core detection engine: runs YOLO model inference on camera images, manages alert states, communicates with the PyroNear API.
 - **`pyro_camera_api/`** — FastAPI service: unified REST interface for controlling heterogeneous cameras (Reolink, Linovision/Hikvision, RTSP, HTTP URL, generic REST/JSON snapshot API).
+- **`pyro_temporal_api/`** — Optional FastAPI service: validates ongoing alerts with the temporal smoke model (`pyronear/temporal-model`, ONNX runtime, no torch) before the engine sends them to the PyroNear API.
 
-These two services run as separate Docker containers and communicate over localhost (host network mode). The engine calls the camera API to capture frames and manage PTZ patrols.
+These services run as separate Docker containers and communicate over localhost (host network mode). The engine calls the camera API to capture frames and manage PTZ patrols.
 
 ## Common Commands
 
@@ -81,6 +82,10 @@ pytest tests/test_engine.py -v
 
 For each `cam_id`, the engine periodically fetches a JSON file at `{bbox_mask_url}_{pose_id}.json` from a remote URL to get a dict of bounding boxes marking permanently occluded regions. Predictions with IoU > 0.1 against any occlusion box are dropped before confidence scoring.
 
+### Temporal validation (optional)
+
+The service is behind the compose profile `temporal` (`COMPOSE_PROFILES=temporal` in `.env`). When `TEMPORAL_API_URL` is set, `Engine` keeps the last `temporal_window` (10) inference JPEGs and their YOLO boxes per `cam_id`. Once an alert is ongoing, `Engine._temporal_gate` submits that window to `pyro_temporal_api` (`POST /jobs`, returns immediately) and reads the verdict on the next round (`GET /jobs/{id}`). Frames are only staged for upload after a positive verdict; a negative verdict resubmits the window with the new frame; a pending job holds the alert one more round. Service errors fail open (alert sent unvalidated). The service loads `data/model_onnx.zip` (`TEMPORAL_MODEL_PATH`), exported from a temporal-model release with `temporal-export-onnx`.
+
 ### Stream-awareness
 
 `SystemController.inference_loop` calls `_any_stream_active()` before and during every camera loop. If an active RTSP/SRT pipeline is detected (via `/stream/status`), the entire inference pass is skipped to avoid interfering with live streaming.
@@ -91,7 +96,7 @@ Day is determined by IR-channel analysis (`is_day_time(strategy="ir")`): if `max
 
 ### Environment Variables (`.env`)
 
-Key vars used at runtime: `LAT`, `LON`, `API_URL`, `API_TOKEN`, `CAM_USER`, `CAM_PWD`, `MEDIAMTX_SERVER_IP`, `ROUTER_IP`, `ROUTER_USER`, `ROUTER_PASSWORD`, `ENABLE_ROUTER_REBOOT`.
+Key vars used at runtime: `LAT`, `LON`, `API_URL`, `API_TOKEN`, `CAM_USER`, `CAM_PWD`, `MEDIAMTX_SERVER_IP`, `ROUTER_IP`, `ROUTER_USER`, `ROUTER_PASSWORD`, `ENABLE_ROUTER_REBOOT`, `COMPOSE_PROFILES=temporal` + `TEMPORAL_API_URL` (e.g. `http://localhost:8082`; both unset = no temporal validation), `TEMPORAL_MODEL_PATH`.
 
 ### Legacy direct-camera module
 

Dockerfile

@@ -1,5 +1,5 @@
 # ---- Builder: only used for git-based deps (needs git) ----
-FROM python:3.11.13-slim-bullseye AS git-deps
+FROM python:3.11.13-slim-bookworm AS git-deps
 
 RUN apt-get update && \
     apt-get install -y --no-install-recommends git \
@@ -12,7 +12,7 @@ COPY ./requirements-git.txt /tmp/requirements-git.txt
 RUN uv pip install --no-cache --target=/tmp/git-packages -r /tmp/requirements-git.txt
 
 # ---- Runtime ----
-FROM python:3.11.13-slim-bullseye
+FROM python:3.11.13-slim-bookworm
 
 ENV LANG="C.UTF-8" \
     PYTHONUNBUFFERED=1 \

Makefile

@@ -37,6 +37,12 @@ build-api:
 	docker build -f pyro_camera_api/Dockerfile pyro_camera_api -t pyronear/pyro-camera-api:latest
 
 # Build the engine Docker image
+build-temporal:
+	cd pyro_temporal_api && \
+		uv lock && \
+		uv export --no-hashes --no-emit-project --no-default-groups --no-dev --format requirements-txt -o requirements.txt
+	docker build -f pyro_temporal_api/Dockerfile pyro_temporal_api -t pyronear/pyro-temporal-api:latest
+
 build-app:
 	docker build . -t pyronear/pyro-engine:latest
 
@@ -50,16 +56,21 @@ build-optional-lib:
 run:
 	docker pull pyronear/pyro-engine:latest
 	docker pull pyronear/pyro-camera-api:latest
+	docker pull pyronear/pyro-temporal-api:latest
 	docker compose up -d
 
 # Build images locally and run the stack
-run_local: build-api build-app
+run_local: build-api build-temporal build-app
 	docker compose up -d
 
 # Get log from engine wrapper
 log:
 	docker logs -f --tail 50 engine
 
+# Get log from temporal API wrapper
+log-temporal:
+	docker logs -f --tail 50 pyro-temporal-api
+
 # Get log from camera API wrapper
 log-api:
 	docker logs -f --tail 50 pyro-camera-api

README.md

@@ -115,7 +115,13 @@ MEDIAMTX_SERVER_IP=1.2.3.4
 PYRO_ENGINE_VERSION=latest
 ```
 
-`PYRO_ENGINE_VERSION` controls which Docker image tag is pulled for both services (defaults to `latest` if unset).
+`PYRO_ENGINE_VERSION` controls which Docker image tag is pulled for the services (defaults to `latest` if unset).
+
+Optional temporal validation of alerts: with `COMPOSE_PROFILES=temporal` and `TEMPORAL_API_URL=http://localhost:8082`
+in `.env`, the stack also starts `pyro_temporal_api`. While an alert is ongoing, the engine sends the last frames and
+boxes of that camera pose to it and only uploads the alert once the temporal smoke model confirms it. Leave both unset
+to send alerts as before. The service needs `data/model_onnx.zip`, exported from a
+[temporal-model](https://github.com/pyronear/temporal-model) release with `temporal-export-onnx`.
 
 ### Data directory
 

docker-compose.yml

@@ -30,6 +30,31 @@ services:
         max-file: "5"
 
 
+  # Optional: started only with COMPOSE_PROFILES=temporal (needs data/model_onnx.zip)
+  pyro-temporal-api:
+    profiles: ["temporal"]
+    build:
+      context: ./pyro_temporal_api
+    image: pyronear/pyro-temporal-api:${PYRO_ENGINE_VERSION:-latest}
+    container_name: pyro-temporal-api
+    environment:
+      TEMPORAL_MODEL_PATH: ${TEMPORAL_MODEL_PATH:-data/model_onnx.zip}
+    volumes:
+      - ./data:/usr/src/app/data
+    restart: always
+    network_mode: host
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:8082/health || exit 1"]
+      interval: 30s
+      retries: 3
+      start_period: 30s
+      timeout: 5s
+    logging:
+      driver: json-file
+      options:
+        max-size: "100m"
+        max-file: "5"
+
   engine:
     build:
       context: ./engine
@@ -39,6 +64,7 @@ services:
       API_URL: ${API_URL}
       CAM_USER: ${CAM_USER}
       CAM_PWD: ${CAM_PWD}
+      TEMPORAL_API_URL: ${TEMPORAL_API_URL:-}
     volumes:
       - ./data:/usr/src/app/data
     command: >

pyproject.toml

@@ -80,7 +80,7 @@ source = ["pyroengine"]
 
 [tool.mypy]
 python_version = "3.11"
-files = "pyroengine/,pyro_camera_api/,pyro-predictor/pyro_predictor/"
+files = "pyroengine/,pyro_camera_api/,pyro_temporal_api/,pyro-predictor/pyro_predictor/"
 show_error_codes = true
 pretty = true
 warn_unused_ignores = true
@@ -135,7 +135,7 @@ exclude = [".git", "pyro-predictor/**", "docs/**"]
 docstring-quotes = "double"
 
 [tool.ruff.lint.isort]
-known-first-party = ["pyroengine", "pyro_camera_api"]
+known-first-party = ["pyroengine", "pyro_camera_api", "pyro_temporal_api"]
 known-third-party = ["pillow", "tqdm", "onnxruntime", "huggingface_hub"]
 
 [tool.ruff.lint.per-file-ignores]
@@ -145,6 +145,7 @@ known-third-party = ["pillow", "tqdm", "onnxruntime", "huggingface_hub"]
 ".github/**.py" = ["D", "T201", "ANN", "S", "PYI024"]
 "tests/**.py" = ["D103", "CPY001", "S101", "T201", "ANN001", "ANN201", "ANN202", "ARG001", "S113"]
 "pyro_camera_api/**.py" = ["D", "T201", "S101", "ANN", "BLE001", "S113", "S501", "S404", "S603", "S405", "S314", "E402", "RUF029"]
+"pyro_temporal_api/**.py" = ["D", "T201", "S101", "ANN", "BLE001", "S113", "S501", "E402", "RUF029"]
 "pyroengine/core.py" = ["BLE001"]
 "pyroengine/sensors.py" = ["S113", "S501", "ANN"]
 "watchdog/**.py" = ["CPY001", "S108", "S310", "S404", "S603", "S607", "BLE001", "LOG015"]

pyro_camera_api/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.9.16-slim
+FROM python:3.9-slim-bookworm
 
 ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \

pyro_temporal_api/Dockerfile

@@ -0,0 +1,26 @@
+FROM python:3.11.13-slim-bookworm
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/usr/src/app
+
+# Layer 1: git for the temporal-model dependency, curl for the healthcheck
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends curl git \
+    || { apt-get update && apt-get install -y --no-install-recommends --fix-missing curl git; } \
+    && apt-get clean && \
+    rm -rf /var/lib/apt/lists/*
+
+# Layer 2: Pip deps (onnxruntime + temporal-model-core, no torch)
+COPY --from=ghcr.io/astral-sh/uv:0.6.16 /uv /bin/uv
+COPY requirements.txt /tmp/requirements.txt
+RUN uv pip install --no-cache --system -r /tmp/requirements.txt && \
+    rm /bin/uv /tmp/requirements.txt
+
+# Layer 3: Source code
+WORKDIR /usr/src/app
+COPY pyproject.toml ./pyproject.toml
+COPY pyro_temporal_api ./pyro_temporal_api
+
+# host network mode: bind to loopback, only the engine on the same box talks to this service
+CMD ["uvicorn", "pyro_temporal_api.main:app", "--host", "127.0.0.1", "--port", "8082"]

pyro_temporal_api/pyproject.toml

@@ -0,0 +1,25 @@
+[project]
+name = "pyro_temporal_api"
+version = "0.1.0"
+description = "FastAPI service validating engine alerts with the temporal smoke model (ONNX, no torch)"
+authors = [{ name = "Pyronear", email = "contact@pyronear.org" }]
+requires-python = ">=3.11,<3.13"
+dependencies = [
+    "fastapi>=0.110.0,<0.111",
+    "uvicorn[standard]>=0.30.0,<0.31",
+    "python-multipart>=0.0.9,<0.1",
+    "pyyaml>=6.0",
+    "temporal-model-core[onnx] @ git+https://github.com/pyronear/temporal-model.git@fb7f6ca855cafa8a2ca0981ac54487c63892f6ad#subdirectory=core",
+]
+
+[dependency-groups]
+dev = [
+    "pytest>=8.0,<9",
+    "httpx>=0.27,<1",
+]
+
+[tool.uv]
+package = false
+
+[tool.pytest.ini_options]
+pythonpath = ["."]

pyro_temporal_api/pyro_temporal_api/__init__.py

@@ -0,0 +1,4 @@
+# Copyright (C) 2022-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.

pyro_temporal_api/pyro_temporal_api/jobs.py

@@ -0,0 +1,140 @@
+# Copyright (C) 2022-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+"""Validation jobs: one submitted window of frames + boxes, scored by the temporal model in a worker thread."""
+
+from __future__ import annotations
+
+import logging
+import queue
+import tempfile
+import threading
+import uuid
+from collections import OrderedDict
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from temporal_model.core import Detection, FrameDetections
+
+logger = logging.getLogger(__name__)
+
+MAX_JOBS_KEPT = 100
+# One job per camera pose per round is the normal load; more means a stuck worker or a flood.
+MAX_PENDING = 16
+
+
+class QueueFullError(RuntimeError):
+    """Raised by submit() when too many jobs are still waiting to be scored."""
+
+
+@dataclass
+class Job:
+    job_id: str
+    cam_id: str
+    frames: list[tuple[str, bytes]]  # (frame_id, jpeg bytes), oldest first
+    boxes: list[list[list[float]]]  # per frame: [[x1, y1, x2, y2, conf], ...] normalized xyxy
+    status: str = "pending"  # pending | done | error
+    verdict: dict[str, Any] | None = None
+    error: str | None = None
+    _done: threading.Event = field(default_factory=threading.Event)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "job_id": self.job_id,
+            "cam_id": self.cam_id,
+            "status": self.status,
+            "verdict": self.verdict,
+            "error": self.error,
+        }
+
+
+def _to_frame_detections(frames: list[Any], boxes: list[list[list[float]]]) -> dict[str, FrameDetections]:
+    """Engine boxes (normalized xyxy + conf) -> the temporal model's per-frame detections."""
+    out = {}
+    for idx, (frame, frame_boxes) in enumerate(zip(frames, boxes, strict=True)):
+        out[frame.frame_id] = FrameDetections(
+            frame_idx=idx,
+            frame_id=frame.frame_id,
+            timestamp=frame.timestamp,
+            detections=[
+                Detection(
+                    class_id=0,
+                    cx=(x1 + x2) / 2,
+                    cy=(y1 + y2) / 2,
+                    w=x2 - x1,
+                    h=y2 - y1,
+                    confidence=float(conf),
+                )
+                for x1, y1, x2, y2, conf in frame_boxes
+            ],
+        )
+    return out
+
+
+def score(model: Any, job: Job) -> dict[str, Any]:
+    """Run the temporal model on a job's window; frames live on disk only for the call."""
+    with tempfile.TemporaryDirectory(prefix="temporal_job_") as td:
+        paths = []
+        for frame_id, data in job.frames:
+            p = Path(td) / f"{frame_id}.jpg"
+            p.write_bytes(data)
+            paths.append(p)
+        frames = model.load_sequence(paths)
+        out = model.predict(frames, frame_detections=_to_frame_detections(frames, job.boxes))
+    kept = out.details["tubes"]["kept"]
+    probs = [t["probability"] for t in kept if t["probability"] is not None]
+    return {
+        "is_positive": bool(out.is_positive),
+        "probability": max(probs) if probs else None,
+        "n_tubes": len(kept),
+    }
+
+
+class JobStore:
+    """FIFO of jobs scored one at a time by a daemon thread; keeps the last MAX_JOBS_KEPT results."""
+
+    def __init__(self, model: Any) -> None:
+        self._model = model
+        self._jobs: OrderedDict[str, Job] = OrderedDict()
+        self._queue: queue.Queue[Job] = queue.Queue()
+        self._lock = threading.Lock()
+        self._worker = threading.Thread(target=self._run, name="temporal-worker", daemon=True)
+        self._worker.start()
+
+    def submit(self, cam_id: str, frames: list[tuple[str, bytes]], boxes: list[list[list[float]]]) -> Job:
+        if self._queue.qsize() >= MAX_PENDING:
+            raise QueueFullError(f"{MAX_PENDING} jobs already pending")
+        job = Job(job_id=uuid.uuid4().hex, cam_id=cam_id, frames=frames, boxes=boxes)
+        with self._lock:
+            self._jobs[job.job_id] = job
+            while len(self._jobs) > MAX_JOBS_KEPT:
+                self._jobs.popitem(last=False)
+        self._queue.put(job)
+        return job
+
+    def get(self, job_id: str) -> Job | None:
+        with self._lock:
+            return self._jobs.get(job_id)
+
+    def wait(self, job_id: str, timeout: float) -> bool:
+        """Block until the job finishes (tests and synchronous callers)."""
+        job = self.get(job_id)
+        return job is not None and job._done.wait(timeout)
+
+    def _run(self) -> None:
+        while True:
+            job = self._queue.get()
+            try:
+                job.verdict = score(self._model, job)
+                job.status = "done"
+                logger.info("job %s cam %s: %s", job.job_id, job.cam_id, job.verdict)
+            except Exception as exc:
+                job.status = "error"
+                job.error = f"{type(exc).__name__}: {exc}"
+                logger.exception("job %s cam %s failed", job.job_id, job.cam_id)
+            finally:
+                job.frames = []  # release the JPEGs once scored
+                job._done.set()