Sync from temporal-model/demo

.dockerignore

@@ -1,12 +0,0 @@
-.git
-.gitignore
-.venv
-__pycache__/
-.ipynb_checkpoints/
-*.py[cod]
-*.ipynb
-.DS_Store
-debug_frames/
-force_06/
-out*.mp4
-output.png

.gitignore

@@ -1,5 +1,2 @@
-data/
-*.mp4
+.venv/
 __pycache__/
-*.ipynb
-acme.json

Dockerfile

@@ -1,24 +0,0 @@
-FROM python:3.11-slim
-
-ENV PYTHONDONTWRITEBYTECODE=1 \
-    PYTHONUNBUFFERED=1 \
-    PIP_NO_CACHE_DIR=1 \
-    STREAMLIT_SERVER_ADDRESS=0.0.0.0 \
-    STREAMLIT_SERVER_PORT=7860 \
-    STREAMLIT_BROWSER_GATHER_USAGE_STATS=false
-
-WORKDIR /app
-
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    ffmpeg \
-    libgomp1 \
-    && rm -rf /var/lib/apt/lists/*
-
-COPY requirements.txt .
-RUN pip install --upgrade pip && pip install -r requirements.txt
-
-COPY . .
-
-EXPOSE 7860
-
-CMD ["streamlit", "run", "app.py", "--server.address=0.0.0.0", "--server.port=7860"]

Makefile

@@ -1,21 +1,23 @@
-COMPOSE := docker compose
+# This folder is a mirror of the HuggingFace Space, so it is shaped like one
+# (app.py + requirements.txt + README.md), not like the repo's uv packages.
+SPACE ?= pyronear/Pyronear-Wildfire-Detection
 
-.PHONY: run run-bg stop down logs build ps
+.PHONY: install run lint format push
 
-run:
-	$(COMPOSE) up --build -d
+install: ## build .venv from requirements.txt (pulls torch — large)
+	uv venv --python 3.12
+	uv pip install --python .venv -r requirements.txt
 
-stop:
-	$(COMPOSE) stop
+run: ## serve the demo at http://localhost:7860
+	.venv/bin/python app.py
 
-down:
-	$(COMPOSE) down
+lint: ## ruff check
+	uv run --project ../core ruff check .
 
-logs:
-	$(COMPOSE) logs -f app
+format: ## ruff format
+	uv run --project ../core ruff format .
 
-build:
-	$(COMPOSE) build
-
-ps:
-	$(COMPOSE) ps
+push: ## mirror this folder to the Space (needs `hf auth login` or HF_TOKEN)
+	uvx --from huggingface_hub hf upload $(SPACE) . . \
+	    --repo-type space --exclude ".venv/*" --delete "*" \
+	    --commit-message "Sync from temporal-model/demo"

README.md

@@ -1,55 +1,66 @@
 ---
 title: Pyronear Wildfire Detection
-emoji: 🏆
+emoji: 🔥
 colorFrom: blue
 colorTo: pink
-sdk: streamlit
-python_version: 3.11
+sdk: gradio
+python_version: "3.12"
 app_file: app.py
 pinned: false
 license: apache-2.0
+models:
+  - pyronear/yolov11s
+  - pyronear/temporal-model
 ---
 
-# MP4 to 8 Frames + Wildfire Detection for Pyronear
+# Pyronear — smoke detection demo
 
-Upload an MP4, extract evenly spaced frames, run wildfire detection on each,
-and display the main detections (one image per main detection).
+Gradio Space to try the two Pyronear wildfire-smoke models side by side. Source
+lives in [`pyronear/temporal-model`](https://github.com/pyronear/temporal-model)
+under `demo/`; the Space is a mirror of that folder.
 
-## Requirements
-- Python 3.9+
-- Packages listed in `requirements.txt`
+| Tab | Model | Input | Output |
+|---|---|---|---|
+| Single frame — detection | [`pyronear/yolov11s`](https://huggingface.co/pyronear/yolov11s) `v8.2.0` | one image | boxes + confidences |
+| Sequence — temporal model | [`pyronear/temporal-model`](https://huggingface.co/pyronear/temporal-model) `v0.4.0` | ordered frames, or a video | smoke / no-smoke, per-tube probabilities, trigger frame |
 
-## Install
-```bash
-python -m venv .venv
-source .venv/bin/activate
-pip install -r requirements.txt
-```
+Both tabs run the detector at the pipeline's production settings
+(`conf=0.1`, `iou=0.2`, `imgsz=1024`). The temporal `model.zip` bundles its own
+pinned YOLO, so tab 2's boxes come from that one — they can differ from tab 1's.
 
-## Run
-```bash
-streamlit run app.py
-```
+Frames of a sequence are ordered **by filename**, the production convention
+(`<prefix>_<YYYY-MM-DDTHH-MM-SS>.jpg`). A video is sampled into evenly spaced
+frames instead.
+
+The trigger frame (earliest frame the model would have fired on) is off by
+default: it re-scores growing prefixes of each tube, which is slow on CPU.
 
-Streamlit will print a local URL (for example, `http://localhost:8501`). Open it
-in your browser and upload an MP4. Detection starts automatically after upload.
+## Run locally
 
-## Docker Compose + Make
-Run with:
 ```bash
-make run
+make install
+make run          # http://localhost:7860
 ```
 
-Other commands:
+`make install` builds `.venv` from `requirements.txt`, which pulls
+`temporal-model-core[torch]` (torch, timm, ultralytics) — the first run is a
+large download. Model weights are fetched from the Hub on first use and cached.
+
+Override the pinned models with `DETECTOR_REPO` / `DETECTOR_REVISION` /
+`TEMPORAL_REPO` / `TEMPORAL_REVISION`.
+
+## Examples
+
+Drop a folder of frames in `demo/examples/<name>/` and it shows up as a
+one-click example in the sequence tab. Committed examples are mirrored to the
+Space; none are bundled today.
+
+## Deploy
+
 ```bash
-make logs
-make stop
-make down
+hf auth login    # once, or set HF_TOKEN
+make push
 ```
 
-Then open `http://127.0.0.1:7860` in your browser.
-
-## Notes
-- The first run downloads the wildfire detection model from Hugging Face.
-- `ffmpeg`/`ffprobe` are required for frame extraction.
-- OpenCV is used for motion features and image processing.
+Mirrors this folder to the Space, deleting anything there that is no longer
+here.

app.py

@@ -1,1359 +1,322 @@
-import logging
-import os
-import shutil
-import subprocess
-import tempfile
-import time
-from hashlib import sha1
-from collections import deque
-from contextlib import contextmanager
-
-import cv2
-import numpy as np
-import streamlit as st
-from PIL import Image, ImageDraw
-
-from vision import Classifier
-from utils import box_iou, nms
-
-
-LOGGER = logging.getLogger(__name__)
-
-
-PYRONEAR_LOGO_URL = (
-    "https://raw.githubusercontent.com/pyronear/pyro-engine/develop/docs/source/_static/img/pyronear-logo-dark.png"
-)
-
-
-DEFAULT_SPLIT_CFG = {
-    "n_samples": 16,
-    "max_w": 400,
-    "crop_y": (0.25, 0.90),
-    "dx_threshold_px": 1.5,
-    "min_inlier_ratio": 0.20,
-    "min_stable_frames": 2,
-    "smooth_window": 2,
-    "orb_nfeatures": 800,
-    "orb_fast_threshold": 12,
-    "min_matches": 25,
-    "keep_ratio": 0.4,
-    "jump_meanabs_threshold": 18.0,
-    "progress_every": 0,
-}
-ENABLE_MOTION_SEGMENTATION = os.getenv("ENABLE_MOTION_SEGMENTATION", "0").strip().lower() in {
-    "1",
-    "true",
-    "yes",
-    "on",
-}
-FAST_N_SAMPLES = max(1, int(os.getenv("FAST_N_SAMPLES", "12")))
-INFER_BATCH_SIZE = max(1, int(os.getenv("INFER_BATCH_SIZE", "16")))
-MODEL_IMGSZ = max(320, int(os.getenv("MODEL_IMGSZ", "1024")))
-MAX_INFER_FRAMES_PER_SPLIT = max(0, int(os.getenv("MAX_INFER_FRAMES_PER_SPLIT", "12")))
-MIN_MAIN_MATCH_ABS = max(1, int(os.getenv("MIN_MAIN_MATCH_ABS", "3")))
-MIN_MAIN_MATCH_RATIO = float(os.getenv("MIN_MAIN_MATCH_RATIO", "0.20"))
-MAIN_DET_MATCH_IOU_THRESHOLD = float(os.getenv("MAIN_DET_MATCH_IOU_THRESHOLD", "0.12"))
-MIN_COMBINED_MEDIAN_CONF = float(os.getenv("MIN_COMBINED_MEDIAN_CONF", "0.12"))
-DISPLAY_DET_MATCH_IOU_THRESHOLD = float(os.getenv("DISPLAY_DET_MATCH_IOU_THRESHOLD", "0.0"))
-
-
-def _log_timing_summary(label, stats, wall_time=None, max_items=12):
-    if not stats:
-        LOGGER.info("%s timing | no data", label)
-        return
-
-    entries = sorted(
-        ((name, float(value)) for name, value in stats.items() if value is not None),
-        key=lambda item: item[1],
-        reverse=True,
-    )
-    if wall_time is None:
-        wall_time = stats.get("wall")
-
-    step_entries = [(name, sec) for name, sec in entries if name != "wall"]
-    parts = []
-    if wall_time is not None:
-        parts.append(f"wall={float(wall_time):.3f}s")
-    for name, sec in step_entries[:max_items]:
-        if wall_time and wall_time > 0:
-            parts.append(f"{name}={sec:.3f}s ({(100.0 * sec / float(wall_time)):.1f}%)")
-        else:
-            parts.append(f"{name}={sec:.3f}s")
-    remaining = max(0, len(step_entries) - max_items)
-    if remaining:
-        parts.append(f"+{remaining} more")
-
-    LOGGER.info("%s timing | %s", label, " | ".join(parts))
-
-
-def _sample_indices(total, n):
-    if total <= 0:
-        return []
-    if total <= n:
-        return list(range(total))
-    return np.linspace(0, total - 1, n).astype(int).tolist()
-
-
-def _format_idx_list(indices, max_items=40):
-    if not indices:
-        return "[]"
-    values = [int(i) for i in indices]
-    if len(values) <= max_items:
-        return str(values)
-    head = values[: max_items // 2]
-    tail = values[-(max_items // 2) :]
-    return f"{head} ... {tail} (len={len(values)})"
-
-
-def _sample_uniform_items(items, n):
-    n = max(1, int(n))
-    if len(items) <= n:
-        return items
-    indices = np.linspace(0, len(items) - 1, n).astype(int).tolist()
-    return [items[i] for i in indices]
-
-
-def _parse_fraction(value):
-    if not value:
-        return None
-    txt = str(value).strip()
-    if not txt or txt == "0/0":
-        return None
-    if "/" in txt:
-        num, den = txt.split("/", 1)
-        try:
-            den_f = float(den)
-            if den_f == 0:
-                return None
-            return float(num) / den_f
-        except Exception:
-            return None
-    try:
-        return float(txt)
-    except Exception:
-        return None
-
-
-def _probe_total_frames_ffprobe(video_path):
-    ffprobe = shutil.which("ffprobe")
-    if ffprobe is None:
-        return None
-
-    timing = {}
-    wall_t0 = time.perf_counter()
-    video_name = os.path.basename(video_path)
-
-    # Try direct frame count first.
-    cmd = [
-        ffprobe,
-        "-v",
-        "error",
-        "-select_streams",
-        "v:0",
-        "-show_entries",
-        "stream=nb_frames",
-        "-of",
-        "default=noprint_wrappers=1:nokey=1",
-        video_path,
-    ]
-    with timer("ffprobe_nb_frames", timing):
-        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
-    if proc.returncode == 0:
-        raw = proc.stdout.strip()
-        if raw.isdigit():
-            val = int(raw)
-            if val > 0:
-                timing["wall"] = time.perf_counter() - wall_t0
-                _log_timing_summary(f"ffprobe ({video_name})", timing, wall_time=timing["wall"])
-                return val
-
-    # Fallback: estimate from duration * avg frame rate.
-    cmd = [
-        ffprobe,
-        "-v",
-        "error",
-        "-select_streams",
-        "v:0",
-        "-show_entries",
-        "stream=avg_frame_rate,duration",
-        "-of",
-        "default=noprint_wrappers=1:nokey=1",
-        video_path,
-    ]
-    with timer("ffprobe_fps_duration", timing):
-        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
-    if proc.returncode != 0:
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"ffprobe ({video_name})", timing, wall_time=timing["wall"])
-        return None
-
-    lines = [line.strip() for line in proc.stdout.splitlines() if line.strip()]
-    if len(lines) < 2:
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"ffprobe ({video_name})", timing, wall_time=timing["wall"])
-        return None
+"""Gradio demo for the two Pyronear smoke models.
 
-    fps = _parse_fraction(lines[0])
-    duration = _parse_fraction(lines[1])
-    if fps is None or duration is None:
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"ffprobe ({video_name})", timing, wall_time=timing["wall"])
-        return None
-
-    estimate = int(round(fps * duration))
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary(f"ffprobe ({video_name})", timing, wall_time=timing["wall"])
-    return estimate if estimate > 0 else None
+Tab 1 — the single-frame detector published at ``pyronear/yolov11s``: one image
+in, boxes out.
 
+Tab 2 — the temporal classifier published at ``pyronear/temporal-model``: an
+ordered sequence in (frames or a video), one smoke/no-smoke decision out. It
+links its detector's boxes into tubes across frames and scores each tube with a
+ViT. Its ``model.zip`` bundles its own YOLO, so tab 2's boxes come from that
+pinned detector, not from tab 1's.
+"""
 
-def _probe_duration_ffprobe(video_path):
-    ffprobe = shutil.which("ffprobe")
-    if ffprobe is None:
-        return None
-
-    cmd = [
-        ffprobe,
-        "-v",
-        "error",
-        "-show_entries",
-        "format=duration",
-        "-of",
-        "default=noprint_wrappers=1:nokey=1",
-        video_path,
-    ]
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
-    if proc.returncode != 0:
-        return None
-
-    lines = [line.strip() for line in proc.stdout.splitlines() if line.strip()]
-    if not lines:
-        return None
-    duration = _parse_fraction(lines[0])
-    if duration is None or duration <= 0:
-        return None
-    return float(duration)
-
-
-def _probe_video_size_ffprobe(video_path):
-    ffprobe = shutil.which("ffprobe")
-    if ffprobe is None:
-        return None
-
-    cmd = [
-        ffprobe,
-        "-v",
-        "error",
-        "-select_streams",
-        "v:0",
-        "-show_entries",
-        "stream=width,height",
-        "-of",
-        "csv=p=0:s=x",
-        video_path,
-    ]
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
-    if proc.returncode != 0:
-        return None
-
-    line = next((txt.strip() for txt in proc.stdout.splitlines() if txt.strip()), "")
-    if "x" not in line:
-        return None
-    left, right = line.split("x", 1)
-    if not left.isdigit() or not right.isdigit():
-        return None
-
-    width, height = int(left), int(right)
-    if width <= 0 or height <= 0:
-        return None
-    return width, height
-
-
-def _extract_bgr_with_ffmpeg_disk(video_path, n):
-    ffmpeg = shutil.which("ffmpeg")
-    if ffmpeg is None:
-        raise RuntimeError("ffmpeg is not available")
-
-    timing = {}
-    wall_t0 = time.perf_counter()
-    video_name = os.path.basename(video_path)
-
-    with timer("probe_total_frames", timing):
-        total = _probe_total_frames_ffprobe(video_path)
-    if total is None or total <= 0:
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        raise RuntimeError("ffprobe could not determine total frame count")
-
-    with timer("sample_indices", timing):
-        indices = _sample_indices(total, int(n))
-    if not indices:
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return []
-
-    LOGGER.info(
-        "Frame extraction | video=%s total_frames=%d n_samples=%d sampled_indices=%s",
-        os.path.basename(video_path),
-        total,
-        len(indices),
-        _format_idx_list(indices),
-    )
-
-    select_expr = "+".join(f"eq(n\\,{int(i)})" for i in indices)
-    vf = f"select={select_expr}"
-
-    with tempfile.TemporaryDirectory(prefix="ffmpeg_frames_") as tmpdir:
-        pattern = os.path.join(tmpdir, "frame_%06d.jpg")
-        cmd = [
-            ffmpeg,
-            "-hide_banner",
-            "-loglevel",
-            "error",
-            "-i",
-            video_path,
-            "-vf",
-            vf,
-            "-vsync",
-            "vfr",
-            "-q:v",
-            "2",
-            pattern,
-        ]
-        with timer("ffmpeg_extract", timing):
-            proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
-        if proc.returncode != 0:
-            timing["wall"] = time.perf_counter() - wall_t0
-            _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-            raise RuntimeError(proc.stderr.strip() or "ffmpeg extraction failed")
-
-        frames = []
-        with timer("read_extracted_images", timing):
-            for name in sorted(os.listdir(tmpdir)):
-                if not name.lower().endswith(".jpg"):
-                    continue
-                frame = cv2.imread(os.path.join(tmpdir, name), cv2.IMREAD_COLOR)
-                if frame is not None:
-                    frames.append(frame)
-        LOGGER.info(
-            "Frame extraction done | video=%s extracted=%d requested=%d",
-            os.path.basename(video_path),
-            len(frames),
-            len(indices),
-        )
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return frames
-
-
-def _extract_bgr_with_ffmpeg(video_path, n):
-    ffmpeg = shutil.which("ffmpeg")
-    if ffmpeg is None:
-        raise RuntimeError("ffmpeg is not available")
-
-    n = max(1, int(n))
-    timing = {}
-    wall_t0 = time.perf_counter()
-    video_name = os.path.basename(video_path)
-
-    with timer("probe_duration", timing):
-        duration = _probe_duration_ffprobe(video_path)
-    if duration is None or duration <= 0:
-        LOGGER.warning("Frame extraction | ffprobe duration unavailable, fallback to disk extraction")
-        with timer("fallback_disk_extract", timing):
-            frames = _extract_bgr_with_ffmpeg_disk(video_path, n)
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return frames
-
-    with timer("probe_video_size", timing):
-        video_size = _probe_video_size_ffprobe(video_path)
-    if video_size is None:
-        LOGGER.warning("Frame extraction | ffprobe size unavailable, fallback to disk extraction")
-        with timer("fallback_disk_extract", timing):
-            frames = _extract_bgr_with_ffmpeg_disk(video_path, n)
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return frames
-
-    width, height = video_size
-    frame_size = int(width) * int(height) * 3
-    if frame_size <= 0:
-        LOGGER.warning("Frame extraction | invalid frame size, fallback to disk extraction")
-        with timer("fallback_disk_extract", timing):
-            frames = _extract_bgr_with_ffmpeg_disk(video_path, n)
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return frames
+import os
+import tempfile
+from functools import lru_cache
+from pathlib import Path
 
-    sample_fps = max(1e-6, float(n) / float(duration))
+import gradio as gr
+from huggingface_hub import hf_hub_download
+from PIL import Image, ImageDraw, ImageFont
 
-    LOGGER.info(
-        "Frame extraction (single ffmpeg/rawvideo) | video=%s duration=%.3fs n_samples=%d fps=%.6f size=%dx%d",
-        video_name,
-        duration,
-        n,
-        sample_fps,
-        width,
-        height,
-    )
+DETECTOR_REPO = os.getenv("DETECTOR_REPO", "pyronear/yolov11s")
+DETECTOR_REVISION = os.getenv("DETECTOR_REVISION", "v8.2.0")
+TEMPORAL_REPO = os.getenv("TEMPORAL_REPO", "pyronear/temporal-model")
+TEMPORAL_REVISION = os.getenv("TEMPORAL_REVISION", "v0.4.0")
 
-    cmd = [
-        ffmpeg,
-        "-hide_banner",
-        "-loglevel",
-        "error",
-        "-i",
-        video_path,
-        "-vf",
-        f"fps={sample_fps:.8f}",
-        "-frames:v",
-        str(n),
-        "-f",
-        "rawvideo",
-        "-pix_fmt",
-        "bgr24",
-        "-",
-    ]
-    with timer("ffmpeg_extract_rawvideo", timing):
-        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
-    if proc.returncode != 0 or not proc.stdout:
-        LOGGER.warning(
-            "Frame extraction rawvideo failed | video=%s err=%s",
-            video_name,
-            (proc.stderr.decode("utf-8", errors="ignore").strip() if proc.stderr else "no stderr"),
-        )
-        with timer("fallback_disk_extract", timing):
-            frames = _extract_bgr_with_ffmpeg_disk(video_path, n)
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-        return frames
+# The detector settings the temporal pipeline runs with (train/params.yaml).
+CONF_THRESHOLD = 0.1
+IOU_NMS = 0.2
+IMAGE_SIZE = 1024
 
-    with timer("decode_rawvideo", timing):
-        raw = proc.stdout
-        frame_count = len(raw) // frame_size
-        usable_bytes = frame_count * frame_size
-        if frame_count > 0 and usable_bytes:
-            arr = np.frombuffer(raw[:usable_bytes], dtype=np.uint8).reshape(frame_count, height, width, 3)
-            frames = [arr[idx].copy() for idx in range(frame_count)]
-        else:
-            frames = []
-    if len(frames) > n:
-        frames = _sample_uniform_items(frames, n)
+EXAMPLES_DIR = Path(__file__).parent / "examples"
+IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp"}
 
-    if not frames:
-        LOGGER.warning("Frame extraction | rawvideo mode returned 0 frame, fallback to disk extraction")
-        with timer("fallback_disk_extract", timing):
-            frames = _extract_bgr_with_ffmpeg_disk(video_path, n)
+# Same palette as the eval viewer, so tube colours read the same across tools.
+TUBE_PALETTE = [
+    "#1f77b4",
+    "#ff7f0e",
+    "#2ca02c",
+    "#d62728",
+    "#9467bd",
+    "#8c564b",
+    "#e377c2",
+    "#7f7f7f",
+    "#bcbd22",
+    "#17becf",
+]
 
-    LOGGER.info(
-        "Frame extraction done | video=%s extracted=%d requested=%d",
-        video_name,
-        len(frames),
-        n,
-    )
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary(f"Frame extraction ({video_name})", timing, wall_time=timing["wall"])
-    return frames
+try:
+    FONT = ImageFont.load_default(size=18)
+except TypeError:  # older Pillow without the size kwarg
+    FONT = ImageFont.load_default()
 
 
-def _extract_with_ffmpeg(video_path, n):
-    timing = {}
-    wall_t0 = time.perf_counter()
-    with timer("extract_bgr", timing):
-        frames = _extract_bgr_with_ffmpeg(video_path, n)
-    with timer("bgr_to_pil", timing):
-        pil_frames = [Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) for frame in frames]
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("Extract with ffmpeg", timing, wall_time=timing["wall"])
-    return pil_frames
+@lru_cache(maxsize=1)
+def detector():
+    """The single-frame YOLO detector (downloaded once, then cached)."""
+    from ultralytics import YOLO  # noqa: PLC0415  # keep app startup fast
 
-
-def split_video(video_path, n=8):
-    if not video_path or not os.path.exists(video_path):
-        return []
-    timing = {}
-    wall_t0 = time.perf_counter()
-    with timer("extract_with_ffmpeg", timing):
-        frames = _extract_with_ffmpeg(video_path, n)
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("split_video", timing, wall_time=timing["wall"])
-    return frames
+    return YOLO(hf_hub_download(DETECTOR_REPO, "best.pt", revision=DETECTOR_REVISION))
 
 
-@contextmanager
-def timer(name, stats):
-    t0 = time.perf_counter()
-    yield
-    stats[name] = stats.get(name, 0.0) + (time.perf_counter() - t0)
+@lru_cache(maxsize=1)
+def temporal_model():
+    """The packaged temporal classifier (downloaded once, then cached)."""
+    from temporal_model.core.model import BboxTubeTemporalModel  # noqa: PLC0415
 
+    package = hf_hub_download(TEMPORAL_REPO, "model.zip", revision=TEMPORAL_REVISION)
+    return BboxTubeTemporalModel.from_package(Path(package))
 
-def _iter_sampled_frames(video_path, n_samples, sampled_frames=None):
-    timing = {}
-    wall_t0 = time.perf_counter()
-    if sampled_frames is None:
-        with timer("extract_bgr_with_ffmpeg", timing):
-            frames = _extract_bgr_with_ffmpeg(video_path, int(n_samples))
-    else:
-        with timer("reuse_sampled_frames", timing):
-            frames = sampled_frames
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("Iter sampled frames", timing, wall_time=timing["wall"])
-    for out_idx, frame in enumerate(frames):
-        yield out_idx, frame
 
+def video_to_frames(video_path: str, n_frames: int, out_dir: Path) -> list[Path]:
+    """Extract ``n_frames`` evenly spaced frames from a video, in time order."""
+    import cv2  # noqa: PLC0415  # only the video path needs OpenCV
 
-def iter_frames(video_path, n_samples, max_w, crop_y, sampled_frames=None):
-    timing = {"resize": 0.0, "crop": 0.0}
-    wall_t0 = time.perf_counter()
-    frame_count = 0
+    capture = cv2.VideoCapture(video_path)
     try:
-        for out_idx, frame in _iter_sampled_frames(video_path, n_samples, sampled_frames=sampled_frames):
-            frame_count += 1
-            proc = frame
-            if max_w > 0 and proc.shape[1] != max_w:
-                t_resize = time.perf_counter()
-                scale = max_w / float(proc.shape[1])
-                proc = cv2.resize(
-                    proc,
-                    (max_w, int(proc.shape[0] * scale)),
-                    interpolation=cv2.INTER_AREA,
-                )
-                timing["resize"] += time.perf_counter() - t_resize
-
-            if crop_y is not None:
-                t_crop = time.perf_counter()
-                h = proc.shape[0]
-                y0 = int(max(0.0, min(1.0, float(crop_y[0]))) * h)
-                y1 = int(max(0.0, min(1.0, float(crop_y[1]))) * h)
-                if y1 > y0:
-                    proc = proc[y0:y1, :]
-                timing["crop"] += time.perf_counter() - t_crop
-
-            yield out_idx, proc
+        total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        if total <= 0:
+            raise gr.Error(f"Could not read any frame from {Path(video_path).name}")
+        step = max(1, total // n_frames)
+        indices = list(range(0, total, step))[:n_frames]
+        paths = []
+        for i, index in enumerate(indices):
+            capture.set(cv2.CAP_PROP_POS_FRAMES, index)
+            ok, frame = capture.read()
+            if not ok:
+                continue
+            path = out_dir / f"frame_{i:03d}.jpg"
+            cv2.imwrite(str(path), frame)
+            paths.append(path)
     finally:
-        timing["wall"] = time.perf_counter() - wall_t0
-        LOGGER.info(
-            "iter_frames summary | n_samples=%d yielded=%d max_w=%d crop_y=%s",
-            int(n_samples),
-            frame_count,
-            int(max_w),
-            crop_y,
+        capture.release()
+    if not paths:
+        raise gr.Error(f"Could not decode {Path(video_path).name}")
+    return paths
+
+
+def collect_frames(
+    images: list[str] | None, video: str | None, n_frames: int, work_dir: Path
+) -> list[Path]:
+    """Resolve the sequence input into ordered frame paths.
+
+    Uploaded images are ordered by filename — the Pyronear convention is that
+    filename order is time order. A video is sampled into ``n_frames``.
+    """
+    if video:
+        return video_to_frames(video, n_frames, work_dir)
+    if not images:
+        raise gr.Error("Upload a sequence of frames, or a video.")
+    paths = sorted((Path(p) for p in images), key=lambda p: p.name)
+    bad = [p.name for p in paths if p.suffix.lower() not in IMAGE_SUFFIXES]
+    if bad:
+        raise gr.Error(f"Not image files: {', '.join(bad)}")
+    return paths
+
+
+def tube_color(tube_id: int) -> str:
+    return TUBE_PALETTE[tube_id % len(TUBE_PALETTE)]
+
+
+def draw_boxes(image_path: Path, boxes: list[tuple]) -> Image.Image:
+    """Draw ``[(bbox_cxcywh_normalized, confidence, colour, label)]`` on a frame."""
+    image = Image.open(image_path).convert("RGB")
+    width, height = image.size
+    draw = ImageDraw.Draw(image)
+    for (cx, cy, w, h), confidence, colour, label in boxes:
+        x0, y0 = (cx - w / 2) * width, (cy - h / 2) * height
+        x1, y1 = (cx + w / 2) * width, (cy + h / 2) * height
+        draw.rectangle([x0, y0, x1, y1], outline=colour, width=4)
+        caption = " ".join(
+            part
+            for part in (label, None if confidence is None else f"{confidence:.2f}")
+            if part
         )
-        _log_timing_summary("iter_frames", timing, wall_time=timing["wall"])
-
-
-def quick_jump_score(prev_gray, gray, small_w=160):
-    h, w = prev_gray.shape[:2]
-    if w > small_w:
-        scale = small_w / float(w)
-        prev_s = cv2.resize(prev_gray, (small_w, int(h * scale)), interpolation=cv2.INTER_AREA)
-        gray_s = cv2.resize(gray, (small_w, int(h * scale)), interpolation=cv2.INTER_AREA)
-    else:
-        prev_s = prev_gray
-        gray_s = gray
-
-    diff = cv2.absdiff(prev_s, gray_s)
-    return float(np.mean(diff))
+        if caption:
+            draw.text((x0, max(0, y0 - 20)), caption, fill=colour, font=FONT)
+    return image
 
 
-def estimate_dx_orb_affine(prev_gray, gray, orb, bf, min_matches, keep_ratio, timing_pair):
-    with timer("orb_detect_compute", timing_pair):
-        kp1, des1 = orb.detectAndCompute(prev_gray, None)
-        kp2, des2 = orb.detectAndCompute(gray, None)
+def input_index(frame_idx: int, padded_frame_indices: list[int]) -> int | None:
+    """Map a model-processed frame index back to the uploaded-frame index.
 
-    if des1 is None or des2 is None or len(kp1) < 8 or len(kp2) < 8:
+    The model pads short sequences with duplicate frames; ``None`` marks such a
+    synthetic slot, which has no uploaded frame to draw on.
+    """
+    if frame_idx in padded_frame_indices:
         return None
-
-    with timer("bf_match", timing_pair):
-        matches = bf.match(des1, des2)
-
-    if len(matches) < min_matches:
-        return None
-
-    with timer("match_sort_filter", timing_pair):
-        matches = sorted(matches, key=lambda m: m.distance)
-        keep_n = max(8, int(len(matches) * keep_ratio))
-        matches = matches[:keep_n]
-
-        pts1 = np.float32([kp1[m.queryIdx].pt for m in matches])
-        pts2 = np.float32([kp2[m.trainIdx].pt for m in matches])
-
-    with timer("ransac_affine", timing_pair):
-        M, inliers = cv2.estimateAffinePartial2D(
-            pts1,
-            pts2,
-            method=cv2.RANSAC,
-            ransacReprojThreshold=3.0,
-            maxIters=1500,
-            confidence=0.99,
+    return frame_idx - sum(1 for p in padded_frame_indices if p < frame_idx)
+
+
+def detect_one_frame(image_path: str | None):
+    """Tab 1: run the single-frame detector and return the annotated image."""
+    if not image_path:
+        raise gr.Error("Upload an image.")
+    results = detector().predict(
+        image_path, imgsz=IMAGE_SIZE, conf=CONF_THRESHOLD, iou=IOU_NMS, verbose=False
+    )[0]
+    rows = [
+        [round(c, 3), *(round(v, 1) for v in xyxy)]
+        for c, xyxy in zip(
+            results.boxes.conf.tolist(), results.boxes.xyxy.tolist(), strict=True
         )
-
-    if M is None:
-        return None
-
-    dx = float(M[0, 2])
-    dy = float(M[1, 2])
-    inlier_ratio = float(np.mean(inliers)) if inliers is not None else 0.0
-
-    return {
-        "dx": dx,
-        "dy": dy,
-        "score_dx": float(abs(dx)),
-        "score_px": float(np.hypot(dx, dy)),
-        "inlier_ratio": inlier_ratio,
-        "matches": len(matches),
-        "M": M,
-    }
+    ]
+    annotated = Image.fromarray(results.plot()[:, :, ::-1])
+    summary = f"**{len(rows)} detection(s)**" if rows else "**No detection**"
+    return annotated, summary, rows
 
 
-def split_video_into_stable_segments_fast(
-    video_path,
-    n_samples=16,
-    max_w=400,
-    crop_y=(0.25, 0.90),
-    dx_threshold_px=1.5,
-    min_inlier_ratio=0.20,
-    min_stable_frames=2,
-    smooth_window=2,
-    orb_nfeatures=800,
-    orb_fast_threshold=12,
-    min_matches=25,
-    keep_ratio=0.4,
-    jump_meanabs_threshold=18.0,
-    progress_every=200,
-    sampled_frames=None,
+def classify_sequence(
+    images: list[str] | None, video: str | None, n_frames: int, compute_trigger: bool
 ):
-    wall_t0 = time.perf_counter()
-    timing_total = {}
-    timing_pair = {}
-
-    with timer("setup", timing_total):
-        orb = cv2.ORB_create(nfeatures=orb_nfeatures, fastThreshold=orb_fast_threshold)
-        bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True)
-
-    metrics = []
-    prev_gray = None
-    frame_count = 0
-
-    with timer("loop_total", timing_total):
-        for _, frame in iter_frames(
-            video_path,
-            n_samples=n_samples,
-            max_w=max_w,
-            crop_y=crop_y,
-            sampled_frames=sampled_frames,
-        ):
-            frame_count += 1
-
-            with timer("to_gray", timing_total):
-                gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-
-            if prev_gray is not None:
-                with timer("quick_jump", timing_total):
-                    q = quick_jump_score(prev_gray, gray)
-
-                if q >= jump_meanabs_threshold:
-                    metrics.append(
-                        {
-                            "dx": np.nan,
-                            "dy": np.nan,
-                            "score_dx": 1e9,
-                            "score_px": 1e9,
-                            "inlier_ratio": 0.0,
-                            "matches": 0,
-                            "M": None,
-                            "quick_jump": q,
-                        }
-                    )
-                else:
-                    m = estimate_dx_orb_affine(
-                        prev_gray,
-                        gray,
-                        orb=orb,
-                        bf=bf,
-                        min_matches=min_matches,
-                        keep_ratio=keep_ratio,
-                        timing_pair=timing_pair,
-                    )
-                    if m is None:
-                        metrics.append(
-                            {
-                                "dx": np.nan,
-                                "dy": np.nan,
-                                "score_dx": 1e9,
-                                "score_px": 1e9,
-                                "inlier_ratio": 0.0,
-                                "matches": 0,
-                                "M": None,
-                                "quick_jump": q,
-                            }
-                        )
-                    else:
-                        m["quick_jump"] = q
-                        metrics.append(m)
-
-                if progress_every and (len(metrics) % progress_every == 0):
-                    print(f"processed pairs: {len(metrics)}")
-
-            prev_gray = gray
-
-    if frame_count < 2:
-        timing_total["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary("Segmentation total", timing_total, wall_time=timing_total["wall"])
-        if timing_pair:
-            _log_timing_summary(
-                "Segmentation pair internals",
-                timing_pair,
-                wall_time=max(timing_total.get("loop_total", 0.0), 1e-9),
-            )
-        return [], metrics, [], {"total": timing_total, "per_pair": timing_pair}
-
-    with timer("post_smooth", timing_total):
-        raw_dx = [m["score_dx"] for m in metrics]
-        raw_inlier = [m["inlier_ratio"] for m in metrics]
-
-        smoothed_dx = []
-        q = deque(maxlen=max(1, int(smooth_window)))
-        for v in raw_dx:
-            if not np.isfinite(v):
-                q.clear()
-                smoothed_dx.append(np.nan)
-            else:
-                q.append(v)
-                smoothed_dx.append(float(np.mean(q)))
-
-    with timer("post_segments", timing_total):
-        min_len = max(1, int(min_stable_frames))
-
-        stable_flags = []
-        for dx_s, r in zip(smoothed_dx, raw_inlier):
-            if not np.isfinite(dx_s):
-                stable_flags.append(False)
-            else:
-                stable_flags.append((dx_s < dx_threshold_px) and (r >= min_inlier_ratio))
-
-        segments = []
-        start = None
-        for i, is_stable in enumerate(stable_flags):
-            if is_stable and start is None:
-                start = i
-            if (not is_stable) and start is not None:
-                end = i
-                if (end - start) >= min_len:
-                    segments.append((start, end))
-                start = None
-
-        if start is not None:
-            end = len(stable_flags)
-            if (end - start) >= min_len:
-                segments.append((start, end))
-
-    LOGGER.info(
-        "Segmentation summary | sampled_frames=%d pair_metrics=%d stable_segments=%d",
-        frame_count,
-        len(metrics),
-        len(segments),
-    )
-    if segments:
-        LOGGER.info("Segment ranges (sample indices) | %s", segments)
-    timing_total["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("Segmentation total", timing_total, wall_time=timing_total["wall"])
-    if timing_pair:
-        _log_timing_summary(
-            "Segmentation pair internals",
-            timing_pair,
-            wall_time=max(timing_total.get("loop_total", 0.0), 1e-9),
-        )
-
-    return segments, metrics, smoothed_dx, {"total": timing_total, "per_pair": timing_pair}
-
-
-def _bgr_to_pil(frame):
-    return Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
-
-
-def extract_segment_frames(video_path, segments, n_samples, sampled_frames=None):
-    timing = {}
-    wall_t0 = time.perf_counter()
-
-    if not segments:
-        LOGGER.info("Segment frame extraction | no segments found")
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary("Segment frame extraction", timing, wall_time=timing["wall"])
-        return []
-
-    with timer("normalize_segments", timing):
-        normalized_segments = []
-        for start, end in segments:
-            s = max(0, int(start))
-            e = max(s, int(end))
-            normalized_segments.append((s, e))
-
-    with timer("prepare_groups", timing):
-        normalized_segments.sort(key=lambda x: x[0])
-        grouped_frames = [[] for _ in normalized_segments]
-        grouped_indices = [[] for _ in normalized_segments]
-    segment_idx = 0
-
-    # Detection runs on original sampled frames (no resize / no crop).
-    to_pil_time = 0.0
-    with timer("assign_frames_to_segments", timing):
-        for frame_idx, frame in _iter_sampled_frames(video_path, n_samples=n_samples, sampled_frames=sampled_frames):
-            while segment_idx < len(normalized_segments) and frame_idx > normalized_segments[segment_idx][1]:
-                segment_idx += 1
-
-            if segment_idx >= len(normalized_segments):
-                break
-
-            seg_start, seg_end = normalized_segments[segment_idx]
-            if seg_start <= frame_idx <= seg_end:
-                t_pil = time.perf_counter()
-                grouped_frames[segment_idx].append(_bgr_to_pil(frame))
-                to_pil_time += time.perf_counter() - t_pil
-                grouped_indices[segment_idx].append(frame_idx)
-    timing["to_pil"] = to_pil_time
-
-    LOGGER.info(
-        "Segment frame extraction summary | segments=%d n_samples=%d",
-        len(normalized_segments),
-        n_samples,
-    )
-    for seg_i, ((seg_start, seg_end), idx_list, frames) in enumerate(
-        zip(normalized_segments, grouped_indices, grouped_frames),
-        start=1,
-    ):
-        LOGGER.info(
-            "Segment %d | requested_range=[%d,%d] matched_frames=%d matched_indices=%s",
-            seg_i,
-            seg_start,
-            seg_end,
-            len(frames),
-            _format_idx_list(idx_list),
+    """Tab 2: run the temporal model and return the decision, frames and tubes."""
+    with tempfile.TemporaryDirectory() as tmp:
+        paths = collect_frames(images, video, int(n_frames), Path(tmp))
+        model = temporal_model()
+        output = model.predict(
+            model.load_sequence(paths), compute_trigger=compute_trigger
         )
+        details = output.details
+        kept = details["tubes"]["kept"]
+        padded = details["preprocessing"]["padded_frame_indices"]
+
+        boxes_per_frame: dict[int, list[tuple]] = {}
+        for tube in kept:
+            colour = tube_color(tube["tube_id"])
+            for entry in tube["entries"]:
+                index = (
+                    None
+                    if entry["bbox"] is None
+                    else input_index(entry["frame_idx"], padded)
+                )
+                if index is None:
+                    continue
+                label = f"#{tube['tube_id']}"
+                boxes_per_frame.setdefault(index, []).append(
+                    (entry["bbox"], entry["confidence"], colour, label)
+                )
 
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("Segment frame extraction", timing, wall_time=timing["wall"])
-    return [frames for frames in grouped_frames if frames]
-
-
-def split_video_stable(video_path, split_cfg=None, fallback_n=16):
-    if not video_path or not os.path.exists(video_path):
-        return []
-
-    timing = {}
-    wall_t0 = time.perf_counter()
-    cfg = DEFAULT_SPLIT_CFG.copy()
-    if split_cfg:
-        cfg.update(split_cfg)
-
-    LOGGER.info("Split config | %s", cfg)
-
-    with timer("extract_sampled_frames", timing):
-        sampled_frames = _extract_bgr_with_ffmpeg(video_path, int(cfg["n_samples"]))
-
-    with timer("split_video_into_stable_segments_fast", timing):
-        segments, _, _, _ = split_video_into_stable_segments_fast(video_path, sampled_frames=sampled_frames, **cfg)
-    with timer("extract_segment_frames", timing):
-        frame_groups = extract_segment_frames(
-            video_path,
-            segments,
-            n_samples=cfg["n_samples"],
-            sampled_frames=sampled_frames,
+        trigger = (
+            None
+            if output.trigger_frame_index is None
+            else input_index(output.trigger_frame_index, padded)
         )
-
-    if frame_groups:
-        LOGGER.info(
-            "Split result | stable_splits=%d split_frame_counts=%s",
-            len(frame_groups),
-            [len(group) for group in frame_groups],
+        gallery = []
+        for i, path in enumerate(paths):
+            caption = f"{i}: {path.name}"
+            if trigger == i:
+                caption += " ⚡ trigger"
+            gallery.append((draw_boxes(path, boxes_per_frame.get(i, [])), caption))
+
+    probabilities = [t["probability"] for t in kept if t["probability"] is not None]
+    verdict = "🔥 **Smoke**" if output.is_positive else "✅ **No smoke**"
+    lines = [
+        verdict,
+        f"- probability: **{max(probabilities):.3f}**"
+        if probabilities
+        else "- probability: n/a (uncalibrated)",
+        f"- tubes: {len(kept)} kept / {details['tubes']['num_candidates']} candidates",
+        f"- frames: {len(paths)} uploaded, {len(padded)} padded",
+    ]
+    if compute_trigger:
+        lines.append(
+            f"- trigger frame: **{trigger}**"
+            if trigger is not None
+            else "- trigger frame: none"
         )
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary("split_video_stable", timing, wall_time=timing["wall"])
-        return frame_groups
-
-    LOGGER.info("Split result | no stable segment, using fallback sampling n=%d", fallback_n)
-    if int(fallback_n) == int(cfg["n_samples"]):
-        with timer("fallback_reuse_sampled_frames", timing):
-            fallback_frames = [_bgr_to_pil(frame) for frame in sampled_frames]
-    else:
-        with timer("fallback_split_video", timing):
-            fallback_frames = split_video(video_path, n=fallback_n)
-    LOGGER.info("Fallback frame count | %d", len(fallback_frames))
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("split_video_stable", timing, wall_time=timing["wall"])
-    return [fallback_frames] if fallback_frames else []
-
-
-@st.cache_resource(show_spinner=False)
-def _load_model():
-    model_t0 = time.perf_counter()
-    clf = Classifier(format="onnx", conf=0.05, imgsz=MODEL_IMGSZ)
-    LOGGER.info("Model init timing | wall=%.3fs", time.perf_counter() - model_t0)
-    LOGGER.info("Model config | imgsz=%d", MODEL_IMGSZ)
-    return clf
-
-
-model = _load_model()
-
-
-def _resolve_video_path(video_input):
-    if not video_input:
-        return None
-    if isinstance(video_input, str):
-        return video_input
-    if isinstance(video_input, dict):
-        for key in ("name", "path", "data", "video"):
-            value = video_input.get(key)
-            if isinstance(value, str) and os.path.exists(value):
-                return value
-    if isinstance(video_input, (list, tuple)):
-        for value in video_input:
-            if isinstance(value, str) and os.path.exists(value):
-                return value
-    return None
-
-
-def _draw_detections(pil_img, preds, subtitle=None):
-    img = pil_img.copy()
-    draw = ImageDraw.Draw(img)
-    width, height = img.size
-    color = (255, 80, 0)
-    preds = np.asarray(preds)
-
-    for x1, y1, x2, y2, conf in preds:
-        x1 = int(max(0.0, min(1.0, float(x1))) * width)
-        y1 = int(max(0.0, min(1.0, float(y1))) * height)
-        x2 = int(max(0.0, min(1.0, float(x2))) * width)
-        y2 = int(max(0.0, min(1.0, float(y2))) * height)
-        draw.rectangle([x1, y1, x2, y2], outline=color, width=3)
-        draw.text((x1 + 4, y1 + 4), f"{conf:.2f}", fill=color)
-
-    draw.text((6, 6), f"detections : {len(preds)}", fill=color)
-    if subtitle:
-        draw.text((6, 26), subtitle, fill=color)
-    return img
-
-
-def _combine_predictions_per_split(frame_preds):
-    n_frames = len(frame_preds)
-    if n_frames == 0:
-        return []
-
-    boxes = np.zeros((0, 5), dtype=np.float64)
-    for bbox in frame_preds:
-        if bbox.size > 0:
-            boxes = np.vstack([boxes, bbox])
+    rows = [
+        [
+            t["tube_id"],
+            f"{t['start_frame']}–{t['end_frame']}",
+            round(t["logit"], 3),
+            None if t["probability"] is None else round(t["probability"], 3),
+            t["first_crossing_frame"],
+        ]
+        for t in kept
+    ]
+    return "\n".join(lines), gallery, rows, details
 
-    if boxes.size == 0:
-        return []
 
-    main_bboxes = np.asarray(nms(boxes), dtype=np.float64)
-    if main_bboxes.size == 0:
+def example_sequences() -> list[list]:
+    """Any ``examples/<name>/*.jpg`` folder becomes a one-click example."""
+    if not EXAMPLES_DIR.is_dir():
         return []
+    return [
+        [sorted(str(p) for p in d.iterdir() if p.suffix.lower() in IMAGE_SUFFIXES)]
+        for d in sorted(EXAMPLES_DIR.iterdir())
+        if d.is_dir()
+    ]
 
-    n_main = len(main_bboxes)
-    matches_per_main = np.zeros(n_main, dtype=int)
-    conf_max_per_main = np.zeros(n_main, dtype=np.float64)
-    matched_conf_values_per_main = [[] for _ in range(n_main)]
-    matched_frame_indices_per_main = [[] for _ in range(n_main)]
-    first_match_frame_idx_per_main = [None for _ in range(n_main)]
-    first_match_bbox_per_main = [None for _ in range(n_main)]
-    for frame_idx, bbox in enumerate(frame_preds):
-        if bbox.size == 0:
-            continue
-        ious = box_iou(bbox[:, :4], main_bboxes[:, :4])
-        match_mask = ious >= MAIN_DET_MATCH_IOU_THRESHOLD
-        has_match = match_mask.any(axis=1)
-        matches_per_main += has_match.astype(int)
-        if np.any(has_match):
-            # Keep only one bbox per frame for each main bbox (best IoU among matches).
-            masked_ious = np.where(match_mask, ious, -1.0)
-            best_idx_per_main = np.argmax(masked_ious, axis=1)
-            best_conf_per_main = bbox[best_idx_per_main, 4].astype(np.float64)
-            matched_conf = np.where(has_match, best_conf_per_main, 0.0)
-            conf_max_per_main = np.maximum(conf_max_per_main, matched_conf)
-            for main_idx in np.flatnonzero(has_match):
-                matched_conf_values_per_main[main_idx].append(float(best_conf_per_main[main_idx]))
-                matched_frame_indices_per_main[main_idx].append(int(frame_idx))
-                if first_match_frame_idx_per_main[main_idx] is None:
-                    first_match_frame_idx_per_main[main_idx] = int(frame_idx)
-                    first_match_bbox_per_main[main_idx] = np.asarray(
-                        bbox[int(best_idx_per_main[main_idx])], dtype=np.float64
-                    ).copy()
-
-    required_matches = max(MIN_MAIN_MATCH_ABS, int(np.ceil(float(MIN_MAIN_MATCH_RATIO) * n_frames)))
-    keep_main = matches_per_main >= required_matches
-    if not np.any(keep_main):
-        return []
 
-    kept = []
-    for idx in np.flatnonzero(keep_main):
-        match_count = int(matches_per_main[idx])
-        matched_conf_values = matched_conf_values_per_main[idx]
-        median_conf = (
-            float(np.median(np.asarray(matched_conf_values, dtype=np.float64))) if matched_conf_values else 0.0
-        )
-        if median_conf < MIN_COMBINED_MEDIAN_CONF:
-            LOGGER.info(
-                (
-                    "Combine drop candidate | matches=%d/%d (required=%d) | "
-                    "median_conf=%.2f < min_combined_median_conf=%.2f"
-                ),
-                match_count,
-                n_frames,
-                required_matches,
-                median_conf,
-                MIN_COMBINED_MEDIAN_CONF,
-            )
-            continue
-        kept.append(
-            {
-                "box": main_bboxes[idx],
-                "match_count": match_count,
-                "n_frames": int(n_frames),
-                "required_matches": int(required_matches),
-                "match_ratio": float(match_count / max(n_frames, 1)),
-                "median_conf": median_conf,
-                "max_conf": float(conf_max_per_main[idx]),
-                "matched_conf_values": matched_conf_values,
-                "matched_frame_indices": matched_frame_indices_per_main[idx],
-                "first_match_frame_idx": first_match_frame_idx_per_main[idx],
-                "first_match_bbox": first_match_bbox_per_main[idx],
-            }
+def build_demo() -> gr.Blocks:
+    with gr.Blocks(title="Pyronear — smoke detection") as demo:
+        gr.Markdown(
+            "# 🔥 Pyronear — smoke detection\n"
+            "Two models, two tabs: the **single-frame detector** "
+            f"([{DETECTOR_REPO}](https://huggingface.co/{DETECTOR_REPO}) "
+            f"`{DETECTOR_REVISION}`) and the **temporal classifier** "
+            f"([{TEMPORAL_REPO}](https://huggingface.co/{TEMPORAL_REPO}) "
+            f"`{TEMPORAL_REVISION}`), which decides on a whole sequence.\n\n"
+            "Running on free CPU — a sequence takes a minute or so."
         )
-    return kept
-
 
-def infer(video_file):
-    timing = {}
-    wall_t0 = time.perf_counter()
-
-    with timer("resolve_video_path", timing):
-        video_path = _resolve_video_path(video_file)
-    LOGGER.info("Inference start | video=%s", video_path)
-    LOGGER.info(
-        (
-            "Inference config | batch_size=%d motion_segmentation=%s fast_n_samples=%d "
-            "max_infer_frames_per_split=%d min_main_match_abs=%d min_main_match_ratio=%.2f "
-            "main_det_match_iou_threshold=%.2f min_combined_median_conf=%.2f "
-            "display_det_match_iou_threshold=%.2f"
-        ),
-        INFER_BATCH_SIZE,
-        ENABLE_MOTION_SEGMENTATION,
-        FAST_N_SAMPLES,
-        MAX_INFER_FRAMES_PER_SPLIT,
-        MIN_MAIN_MATCH_ABS,
-        MIN_MAIN_MATCH_RATIO,
-        MAIN_DET_MATCH_IOU_THRESHOLD,
-        MIN_COMBINED_MEDIAN_CONF,
-        DISPLAY_DET_MATCH_IOU_THRESHOLD,
-    )
-    with timer("prepare_splits", timing):
-        if ENABLE_MOTION_SEGMENTATION:
-            split_frames = split_video_stable(video_path)
-        else:
-            fast_frames = split_video(video_path, n=FAST_N_SAMPLES)
-            split_frames = [fast_frames] if fast_frames else []
-    total_frames = sum(len(frames) for frames in split_frames)
-    LOGGER.info("Inference workload | splits=%d total_frames=%d", len(split_frames), total_frames)
-    if not split_frames:
-        LOGGER.info("Inference stop | no frames available")
-        timing["wall"] = time.perf_counter() - wall_t0
-        _log_timing_summary("Inference", timing, wall_time=timing["wall"])
-        return {"detections": [], "all_frame_predictions": []}
-
-    outputs = []
-    all_frame_predictions = []
-    infer_model = 0.0
-    combine_time = 0.0
-    iou_time = 0.0
-    draw_time = 0.0
-    draw_all_frames_time = 0.0
-    split_loop_time = 0.0
-    for split_idx, frames in enumerate(split_frames):
-        split_t0 = time.perf_counter()
-        original_len = len(frames)
-        if MAX_INFER_FRAMES_PER_SPLIT > 0 and original_len > MAX_INFER_FRAMES_PER_SPLIT:
-            frames_for_infer = _sample_uniform_items(frames, MAX_INFER_FRAMES_PER_SPLIT)
-        else:
-            frames_for_infer = frames
-        LOGGER.info(
-            "Inference split %d | frames=%d used_for_infer=%d",
-            split_idx + 1,
-            original_len,
-            len(frames_for_infer),
-        )
-        t_model = time.perf_counter()
-        if hasattr(model, "infer_batch"):
-            frame_preds = model.infer_batch(frames_for_infer, batch_size=INFER_BATCH_SIZE)
-        else:
-            frame_preds = [model(frame) for frame in frames_for_infer]
-        frame_preds = [np.asarray(bbox, dtype=np.float64).reshape(-1, 5) for bbox in frame_preds]
-        for frame_idx, bbox in enumerate(frame_preds):
-            if bbox.size == 0:
-                LOGGER.info(
-                    "Inference split %d frame %d | detections=0",
-                    split_idx + 1,
-                    frame_idx + 1,
-                )
-                continue
-            confs = bbox[:, 4].astype(np.float64)
-            conf_list_txt = ", ".join(f"{float(c):.2f}" for c in confs.tolist())
-            LOGGER.info(
-                (
-                    "Inference split %d frame %d | detections=%d | confs=[%s] | "
-                    "frame_max_conf=%.2f | frame_mean_conf_all_bboxes=%.2f"
-                ),
-                split_idx + 1,
-                frame_idx + 1,
-                len(bbox),
-                conf_list_txt,
-                float(np.max(confs)),
-                float(np.mean(confs)),
+        with gr.Tab("Single frame — detection"):
+            with gr.Row():
+                with gr.Column():
+                    image_in = gr.Image(type="filepath", label="Frame")
+                    detect_button = gr.Button("Detect", variant="primary")
+                with gr.Column():
+                    detect_summary = gr.Markdown()
+                    image_out = gr.Image(label="Detections")
+            detect_table = gr.Dataframe(
+                headers=["confidence", "x0", "y0", "x1", "y1"], label="Boxes (pixels)"
             )
-        for frame_idx, (frame, bbox) in enumerate(zip(frames_for_infer, frame_preds)):
-            subtitle = f"segment {split_idx + 1} / frame {frame_idx + 1}"
-            t_draw_all = time.perf_counter()
-            all_frame_predictions.append(
-                {
-                    "image": _draw_detections(frame, bbox, subtitle=subtitle),
-                    "caption": f"Segment {split_idx + 1} - Frame {frame_idx + 1}",
-                }
+            detect_button.click(
+                detect_one_frame,
+                inputs=image_in,
+                outputs=[image_out, detect_summary, detect_table],
             )
-            draw_all_frames_time += time.perf_counter() - t_draw_all
-
-        split_model = time.perf_counter() - t_model
-        infer_model += split_model
-        split_iou = 0.0
-        split_draw = 0.0
 
-        t_combine = time.perf_counter()
-        kept_main = _combine_predictions_per_split(frame_preds)
-        dt_combine = time.perf_counter() - t_combine
-        combine_time += dt_combine
-        LOGGER.info(
-            "Inference split %d | combined_detections=%d",
-            split_idx + 1,
-            len(kept_main),
-        )
-        for det_idx, det_info in enumerate(kept_main):
-            conf_values_txt = ", ".join(f"{float(c):.2f}" for c in det_info["matched_conf_values"])
-            frame_indices_txt = ", ".join(str(int(i) + 1) for i in det_info["matched_frame_indices"])
-            LOGGER.info(
-                (
-                    "Inference split %d combined detection %d | matches=%d/%d "
-                    "(required=%d, ratio=%.2f) | combine_median_conf=%.2f | combine_max_conf=%.2f | "
-                    "matched_frames=[%s] | matched_confs=[%s]"
-                ),
-                split_idx + 1,
-                det_idx + 1,
-                det_info["match_count"],
-                det_info["n_frames"],
-                det_info["required_matches"],
-                det_info["match_ratio"],
-                det_info["median_conf"],
-                det_info["max_conf"],
-                frame_indices_txt,
-                conf_values_txt,
-            )
-        if not kept_main:
-            split_elapsed = time.perf_counter() - split_t0
-            split_loop_time += split_elapsed
-            LOGGER.info(
-                (
-                    "Inference split %d timing | total=%.3fs | model=%.3fs | combine=%.3fs | "
-                    "iou=%.3fs | draw=%.3fs | avg_model_ms=%.1f"
-                ),
-                split_idx + 1,
-                split_elapsed,
-                split_model,
-                dt_combine,
-                split_iou,
-                split_draw,
-                (1000.0 * split_model / max(len(frames_for_infer), 1)),
+        with gr.Tab("Sequence — temporal model"):
+            gr.Markdown(
+                "Upload the frames of one sequence (ordered by filename, as in "
+                "production) **or** a video, which is sampled into frames. "
+                "Boxes are coloured per tube."
             )
-            continue
-
-        for det_idx, det_info in enumerate(kept_main):
-            main_box = det_info["box"]
-            selected_frame_idx = None
-            selected_bbox = None
-            selection_source = None
-
-            # Prefer the earliest frame that overlaps the combined detection, using a relaxed
-            # threshold for display (so we show the first visible appearance of the event).
-            for frame_idx, bbox in enumerate(frame_preds):
-                if bbox.size == 0:
-                    continue
-                t_iou = time.perf_counter()
-                ious = box_iou(bbox[:, :4], main_box[:4].reshape(1, 4))
-                dt_iou = time.perf_counter() - t_iou
-                split_iou += dt_iou
-                iou_time += dt_iou
-                if (ious > DISPLAY_DET_MATCH_IOU_THRESHOLD).any():
-                    match_idx = int(np.argmax(ious[0]))
-                    selected_frame_idx = int(frame_idx)
-                    selected_bbox = np.asarray(bbox[match_idx], dtype=np.float64).reshape(1, 5)
-                    selection_source = "display_first_overlap"
-                    break
-
-            first_match_frame_idx = det_info.get("first_match_frame_idx")
-            first_match_bbox = det_info.get("first_match_bbox")
-            if selected_frame_idx is None or selected_bbox is None:
-                if (
-                    first_match_frame_idx is None
-                    or first_match_bbox is None
-                    or int(first_match_frame_idx) < 0
-                    or int(first_match_frame_idx) >= len(frames_for_infer)
-                ):
-                    LOGGER.warning(
-                        "Inference split %d detection %d | missing display frame and first matched frame/bbox",
-                        split_idx + 1,
-                        det_idx + 1,
+            with gr.Row():
+                with gr.Column():
+                    frames_in = gr.File(
+                        file_count="multiple", file_types=["image"], label="Frames"
                     )
-                    continue
-                selected_frame_idx = int(first_match_frame_idx)
-                selected_bbox = np.asarray(first_match_bbox, dtype=np.float64).reshape(1, 5)
-                selection_source = "combine_first_match_fallback"
-
-            frame = frames_for_infer[selected_frame_idx]
-            LOGGER.info(
-                (
-                    "Inference split %d detection %d | selected_frame=%d | source=%s | "
-                    "selected frame_conf=%.2f | combine_median_conf=%.2f | combine_max_conf=%.2f"
-                ),
-                split_idx + 1,
-                det_idx + 1,
-                selected_frame_idx + 1,
-                selection_source,
-                float(selected_bbox[0, 4]),
-                det_info["median_conf"],
-                det_info["max_conf"],
-            )
-            subtitle = (
-                f"segment {split_idx + 1} / detection {det_idx + 1} | "
-                f"frame {selected_frame_idx + 1} | "
-                f"matchs {det_info['match_count']}/{det_info['n_frames']} | "
-                f"conf_med {det_info['median_conf']:.2f}"
+                    video_in = gr.Video(label="…or a video")
+                    n_frames_in = gr.Slider(
+                        3, 30, value=12, step=1, label="Frames sampled from the video"
+                    )
+                    trigger_in = gr.Checkbox(
+                        label="Compute the trigger frame (slower)", value=False
+                    )
+                    classify_button = gr.Button("Classify sequence", variant="primary")
+                with gr.Column():
+                    verdict_out = gr.Markdown()
+                    tubes_out = gr.Dataframe(
+                        headers=["tube", "frames", "logit", "probability", "crossing"],
+                        label="Tubes",
+                    )
+            gallery_out = gr.Gallery(label="Frames", columns=4, height=420)
+            with gr.Accordion("Raw details", open=False):
+                details_out = gr.JSON()
+            examples = example_sequences()
+            if examples:
+                gr.Examples(examples=examples, inputs=frames_in)
+            classify_button.click(
+                classify_sequence,
+                inputs=[frames_in, video_in, n_frames_in, trigger_in],
+                outputs=[verdict_out, gallery_out, tubes_out, details_out],
             )
-            t_draw = time.perf_counter()
-            outputs.append(_draw_detections(frame, selected_bbox, subtitle=subtitle))
-            dt_draw = time.perf_counter() - t_draw
-            split_draw += dt_draw
-            draw_time += dt_draw
-
-        split_elapsed = time.perf_counter() - split_t0
-        split_loop_time += split_elapsed
-        LOGGER.info(
-            (
-                "Inference split %d timing | total=%.3fs | model=%.3fs | combine=%.3fs | "
-                "iou=%.3fs | draw=%.3fs | avg_model_ms=%.1f"
-            ),
-            split_idx + 1,
-            split_elapsed,
-            split_model,
-            dt_combine,
-            split_iou,
-            split_draw,
-            (1000.0 * split_model / max(len(frames_for_infer), 1)),
-        )
-
-    timing["split_loop"] = split_loop_time
-    timing["model_infer"] = infer_model
-    timing["combine_predictions"] = combine_time
-    timing["iou_matching"] = iou_time
-    timing["draw_detections"] = draw_time
-    timing["draw_all_frame_predictions"] = draw_all_frames_time
-    timing["wall"] = time.perf_counter() - wall_t0
-    _log_timing_summary("Inference", timing, wall_time=timing["wall"])
-    LOGGER.info(
-        "Inference done | output_images=%d all_frame_prediction_images=%d",
-        len(outputs),
-        len(all_frame_predictions),
-    )
-    return {"detections": outputs, "all_frame_predictions": all_frame_predictions}
-
-
-def _upload_signature(uploaded_file):
-    buffer = uploaded_file.getbuffer()
-    size = uploaded_file.size if uploaded_file.size is not None else len(buffer)
-    digest = sha1(buffer).hexdigest()
-    return (uploaded_file.name or "uploaded.mp4", int(size), digest)
-
-
-def _write_uploaded_video(uploaded_file):
-    ext = os.path.splitext(uploaded_file.name or "")[1] or ".mp4"
-    with tempfile.NamedTemporaryFile(prefix="upload_", suffix=ext, delete=False) as tmp:
-        tmp.write(uploaded_file.getbuffer())
-        return tmp.name
-
-
-def _render_outputs(outputs):
-    detections = outputs
-    all_frame_predictions = []
-    if isinstance(outputs, dict):
-        detections = outputs.get("detections", [])
-        all_frame_predictions = outputs.get("all_frame_predictions", [])
-
-    if not detections:
-        st.warning("Aucune detection d'incendie trouvee dans cette video.")
-    else:
-        st.subheader("Incendies detectes")
-        columns = st.columns(2)
-        for idx, image in enumerate(detections):
-            columns[idx % 2].image(image, caption=f"Detection {idx + 1}", use_container_width=True)
-
-    # if all_frame_predictions:
-    #     with st.expander(
-    #         f"Predictions sur toutes les frames echantillonnees ({len(all_frame_predictions)})",
-    #         expanded=False,
-    #     ):
-    #         columns = st.columns(2)
-    #         for idx, item in enumerate(all_frame_predictions):
-    #             image = item["image"] if isinstance(item, dict) else item
-    #             caption = (
-    #                 item.get("caption", f"Frame {idx + 1}")
-    #                 if isinstance(item, dict)
-    #                 else f"Frame {idx + 1}"
-    #             )
-    #             columns[idx % 2].image(image, caption=caption, use_container_width=True)
-
-
-def main():
-    st.set_page_config(page_title="Detection d'incendies Pyronear", layout="wide")
-    st.image(PYRONEAR_LOGO_URL, width=220)
-    st.title("Detection d'incendies Pyronear")
-    st.write("Televersez un MP4 pour lancer la detection automatiquement.")
-
-    uploaded = st.file_uploader("Televerser un MP4", type=["mp4"])
-    if uploaded is None:
-        st.info("En attente du televersement d'une video.")
-        return
-
-    signature = _upload_signature(uploaded)
-    previous_signature = st.session_state.get("upload_signature")
-    if signature != previous_signature:
-        temp_path = None
-        st.session_state["upload_signature"] = signature
-        with st.spinner("Detection d'incendies en cours..."):
-            try:
-                temp_path = _write_uploaded_video(uploaded)
-                st.session_state["output_images"] = infer(temp_path)
-                st.session_state["inference_error"] = None
-            except Exception as exc:
-                LOGGER.exception("Inference failed")
-                st.session_state["output_images"] = []
-                st.session_state["inference_error"] = str(exc)
-            finally:
-                if temp_path and os.path.exists(temp_path):
-                    os.remove(temp_path)
-
-    if st.session_state.get("inference_error"):
-        st.error(f"Echec de la detection : {st.session_state['inference_error']}")
-        return
 
-    _render_outputs(st.session_state.get("output_images", []))
+    return demo
 
 
 if __name__ == "__main__":
-    main()
+    build_demo().launch()

docker-compose.yml

@@ -1,60 +0,0 @@
-services:
-  traefik:
-    image: traefik:v3.6.8
-    container_name: traefik
-    environment:
-      DOCKER_API_VERSION: "${TRAEFIK_DOCKER_API_VERSION:-1.44}"
-    command:
-      - "--log.level=INFO"
-      - "--providers.docker=true"
-      - "--providers.docker.exposedbydefault=false"
-      - "--entrypoints.web.address=:80"
-      - "--entrypoints.websecure.address=:443"
-      - "--entrypoints.web.http.redirections.entrypoint.to=websecure"
-      - "--entrypoints.web.http.redirections.entrypoint.scheme=https"
-      - "--certificatesresolvers.pyroresolver.acme.tlschallenge=true"
-      - "--certificatesresolvers.pyroresolver.acme.email=${TRAEFIK_ACME_EMAIL:-contact@pyronear.org}"
-      - "--certificatesresolvers.pyroresolver.acme.storage=/acme.json"
-    ports:
-      - "80:80"
-      - "443:443"
-    volumes:
-      - "/var/run/docker.sock:/var/run/docker.sock:ro"
-      - "./acme.json:/acme.json"
-    restart: unless-stopped
-    depends_on:
-      - app
-
-  app:
-    build:
-      context: .
-      dockerfile: Dockerfile
-    container_name: pyronear-wildfire-detection
-    expose:
-      - "7860"
-    environment:
-      STREAMLIT_SERVER_ADDRESS: "0.0.0.0"
-      STREAMLIT_SERVER_PORT: "7860"
-      STREAMLIT_BROWSER_GATHER_USAGE_STATS: "false"
-      ENABLE_MOTION_SEGMENTATION: "0"
-      FAST_N_SAMPLES: "12"
-      INFER_BATCH_SIZE: "16"
-      MODEL_IMGSZ: "1024"
-      MAX_INFER_FRAMES_PER_SPLIT: "12"
-      MIN_MAIN_MATCH_ABS: "3"
-      MIN_MAIN_MATCH_RATIO: "0.20"
-      ORT_PROVIDERS: "CPUExecutionProvider"
-      ORT_INTRA_OP_NUM_THREADS: "8"
-      ORT_INTER_OP_NUM_THREADS: "1"
-    labels:
-      - "traefik.enable=true"
-      - "traefik.http.routers.pyronear-http.rule=Host(`demo-pyronear-egm.pyronear.org`)"
-      - "traefik.http.routers.pyronear-http.entrypoints=web"
-      - "traefik.http.routers.pyronear-http.middlewares=redirect-to-https"
-      - "traefik.http.middlewares.redirect-to-https.redirectscheme.scheme=https"
-      - "traefik.http.routers.pyronear.rule=Host(`demo-pyronear-egm.pyronear.org`)"
-      - "traefik.http.routers.pyronear.entrypoints=websecure"
-      - "traefik.http.routers.pyronear.tls=true"
-      - "traefik.http.routers.pyronear.tls.certresolver=pyroresolver"
-      - "traefik.http.services.pyronear.loadbalancer.server.port=7860"
-    restart: unless-stopped

images_to_video.py

@@ -1,100 +0,0 @@
-import argparse
-import re
-from pathlib import Path
-
-import imageio.v2 as imageio
-import numpy as np
-from PIL import Image
-
-
-def _natural_key(text):
-    parts = re.split(r"(\d+)", text)
-    return [int(p) if p.isdigit() else p.lower() for p in parts]
-
-
-def _iter_images(folder, extensions, natural_sort):
-    paths = [
-        p
-        for p in Path(folder).iterdir()
-        if p.is_file() and p.suffix.lower() in extensions
-    ]
-    if natural_sort:
-        return sorted(paths, key=lambda p: _natural_key(p.name))
-    return sorted(paths, key=lambda p: p.name.lower())
-
-
-def images_to_mp4(
-    input_dir,
-    output_path,
-    fps,
-    extensions,
-    natural_sort,
-    resize_to_first,
-    codec,
-):
-    images = _iter_images(input_dir, extensions, natural_sort)
-    if not images:
-        raise ValueError(f"No images found in {input_dir}")
-
-    first = Image.open(images[0]).convert("RGB")
-    target_size = first.size
-
-    with imageio.get_writer(output_path, fps=fps, codec=codec) as writer:
-        writer.append_data(np.array(first))
-        for path in images[1:]:
-            img = Image.open(path).convert("RGB")
-            if img.size != target_size:
-                if resize_to_first:
-                    img = img.resize(target_size, Image.Resampling.LANCZOS)
-                else:
-                    raise ValueError(
-                        f"Size mismatch: {path.name} is {img.size}, expected {target_size}"
-                    )
-            writer.append_data(np.array(img))
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Create an MP4 video from images in a folder."
-    )
-    parser.add_argument("input_dir", help="Folder containing images.")
-    parser.add_argument("output", help="Output MP4 path, e.g. out.mp4.")
-    parser.add_argument(
-        "--fps", type=float, default=24, help="Frames per second (default: 24)."
-    )
-    parser.add_argument(
-        "--ext",
-        action="append",
-        default=[".jpg", ".jpeg", ".png"],
-        help="Allowed extensions (repeatable). Default: .jpg .jpeg .png",
-    )
-    parser.add_argument(
-        "--natural",
-        action="store_true",
-        help="Use natural sort for filenames (e.g. img2 before img10).",
-    )
-    parser.add_argument(
-        "--no-resize",
-        action="store_true",
-        help="Fail if image sizes differ instead of resizing to the first image.",
-    )
-    parser.add_argument(
-        "--codec",
-        default="libx264",
-        help="Video codec for MP4 (default: libx264).",
-    )
-    args = parser.parse_args()
-
-    images_to_mp4(
-        args.input_dir,
-        args.output,
-        fps=args.fps,
-        extensions={e.lower() for e in args.ext},
-        natural_sort=args.natural,
-        resize_to_first=not args.no_resize,
-        codec=args.codec,
-    )
-
-
-if __name__ == "__main__":
-    main()

packages.txt

@@ -1 +1,2 @@
-ffmpeg
+libgl1
+libglib2.0-0

requirements.txt

@@ -1,6 +1,2 @@
-streamlit==1.43.2
-numpy>=1.26,<3
-Pillow>=10.0.0
-opencv-python-headless>=4.10.0.84
-onnxruntime>=1.20.0
-tqdm>=4.66.0
+gradio>=5.0
+temporal-model-core[torch] @ git+https://github.com/pyronear/temporal-model.git@main#subdirectory=core

test_app.py

@@ -0,0 +1,66 @@
+"""Self-check for the demo's pure helpers: `python test_app.py`.
+
+The model-running paths need torch and the Hub; these are the bits that can be
+silently wrong (frame ordering, padded-index mapping, box geometry).
+"""
+
+import tempfile
+from pathlib import Path
+
+from app import collect_frames, draw_boxes, input_index, tube_color
+from PIL import Image
+
+
+def test_input_index():
+    # No padding: identity.
+    assert [input_index(i, []) for i in range(3)] == [0, 1, 2]
+    # Frames 0 and 3 are synthetic duplicates; the rest shift down past them.
+    assert input_index(0, [0, 3]) is None
+    assert input_index(1, [0, 3]) == 0
+    assert input_index(2, [0, 3]) == 1
+    assert input_index(3, [0, 3]) is None
+    assert input_index(4, [0, 3]) == 2
+
+
+def test_collect_frames_orders_by_filename():
+    with tempfile.TemporaryDirectory() as tmp:
+        work = Path(tmp)
+        unordered = [str(work / "b_02.jpg"), str(work / "a_01.jpg")]
+        assert [p.name for p in collect_frames(unordered, None, 8, work)] == [
+            "a_01.jpg",
+            "b_02.jpg",
+        ]
+
+
+def test_collect_frames_rejects_non_images():
+    with tempfile.TemporaryDirectory() as tmp:
+        work = Path(tmp)
+        try:
+            collect_frames([str(work / "notes.txt")], None, 8, work)
+        except Exception as e:
+            assert "notes.txt" in str(e)
+        else:
+            raise AssertionError("expected a rejection")
+
+
+def test_draw_boxes_marks_the_box_pixels():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "frame.jpg"
+        Image.new("RGB", (200, 100), "black").save(path)
+        # Centred box, half the frame: outline lands on the mid-height edges.
+        out = draw_boxes(path, [((0.5, 0.5, 0.5, 0.5), 0.42, "#ff0000", "#0")])
+        assert out.size == (200, 100)
+        assert out.getpixel((50, 50))[0] > 100, "left edge should be drawn"
+        assert out.getpixel((150, 50))[0] > 100, "right edge should be drawn"
+        assert out.getpixel((100, 50)) == (0, 0, 0), "box interior stays untouched"
+
+
+def test_tube_color_cycles():
+    assert tube_color(0) == tube_color(10) != tube_color(1)
+
+
+if __name__ == "__main__":
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_"):
+            fn()
+            print(f"ok {name}")

utils.py

@@ -1,116 +0,0 @@
-# Copyright (C) 2022-2025, 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.
-
-
-import cv2
-import numpy as np
-from tqdm import tqdm
-
-__all__ = ["DownloadProgressBar", "letterbox", "nms", "xywh2xyxy"]
-
-
-def xywh2xyxy(x: np.ndarray):
-    y = np.copy(x)
-    y[..., 0] = x[..., 0] - x[..., 2] / 2  # top left x
-    y[..., 1] = x[..., 1] - x[..., 3] / 2  # top left y
-    y[..., 2] = x[..., 0] + x[..., 2] / 2  # bottom right x
-    y[..., 3] = x[..., 1] + x[..., 3] / 2  # bottom right y
-    return y
-
-
-def letterbox(
-    im: np.ndarray,
-    new_shape: tuple = (1024, 1024),
-    color: tuple = (114, 114, 114),
-    auto: bool = False,
-    stride: int = 32,
-):
-    """Letterbox image transform for yolo models
-    Args:
-        im (np.ndarray): Input image
-        new_shape (tuple, optional): Image size. Defaults to (1024, 1024).
-        color (tuple, optional): Pixel fill value for the area outside the transformed image.
-        Defaults to (114, 114, 114).
-        auto (bool, optional): auto padding. Defaults to False.
-        stride (int, optional): padding stride. Defaults to 32.
-    Returns:
-        np.ndarray: Output image
-    """
-    # Resize and pad image while meeting stride-multiple constraints
-    im = np.array(im)
-    shape = im.shape[:2]  # current shape [height, width]
-    if isinstance(new_shape, int):
-        new_shape = (new_shape, new_shape)
-    # Scale ratio (new / old)
-    r = min(new_shape[0] / shape[0], new_shape[1] / shape[1])
-    # Compute padding
-    new_unpad = int(round(shape[1] * r)), int(round(shape[0] * r))
-    dw, dh = new_shape[1] - new_unpad[0], new_shape[0] - new_unpad[1]  # wh padding
-    if auto:  # minimum rectangle
-        dw, dh = np.mod(dw, stride), np.mod(dh, stride)  # wh padding
-    dw /= 2  # divide padding into 2 sides
-    dh /= 2
-    if shape[::-1] != new_unpad:  # resize
-        im = cv2.resize(im, new_unpad, interpolation=cv2.INTER_LINEAR)
-    top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1))
-    left, right = int(round(dw - 0.1)), int(round(dw + 0.1))
-    # add border
-    h, w = im.shape[:2]
-    im_b = np.zeros((h + top + bottom, w + left + right, 3)) + color
-    im_b[top : top + h, left : left + w, :] = im
-    return im_b.astype("uint8"), (left, top)
-
-
-def box_iou(box1: np.ndarray, box2: np.ndarray, eps: float = 1e-7):
-    """
-    Calculate intersection-over-union (IoU) of boxes.
-    Both sets of boxes are expected to be in (x1, y1, x2, y2) format.
-    Based on https://github.com/pytorch/vision/blob/master/torchvision/ops/boxes.py
-
-    Args:
-        box1 (np.ndarray): A numpy array of shape (N, 4) representing N bounding boxes.
-        box2 (np.ndarray): A numpy array of shape (M, 4) representing M bounding boxes.
-        eps (float, optional): A small value to avoid division by zero. Defaults to 1e-7.
-
-    Returns:
-        (np.ndarray): An NxM numpy array containing the pairwise IoU values for every element in box1 and box2.
-    """
-    (a1, a2), (b1, b2) = np.split(box1, 2, 1), np.split(box2, 2, 1)
-    inter = (np.minimum(a2, b2[:, None, :]) - np.maximum(a1, b1[:, None, :])).clip(0).prod(2)
-
-    # IoU = inter / (area1 + area2 - inter)
-    return inter / ((a2 - a1).prod(1) + (b2 - b1).prod(1)[:, None] - inter + eps)
-
-
-def nms(boxes: np.ndarray, overlapThresh: int = 0):
-    """Non maximum suppression
-
-    Args:
-        boxes (np.ndarray): A numpy array of shape (N, 4) representing N bounding boxes in (x1, y1, x2, y2, conf) format
-        overlapThresh (int, optional): iou threshold. Defaults to 0.
-
-    Returns:
-        boxes: Boxes after NMS
-    """
-    # Return an empty list, if no boxes given
-    boxes = boxes[boxes[:, -1].argsort()]
-    if len(boxes) == 0:
-        return []
-
-    indices = np.arange(len(boxes))
-    rr = box_iou(boxes[:, :4], boxes[:, :4])
-    for i, box in enumerate(boxes):
-        temp_indices = indices[indices != i]
-        if np.any(rr[i, temp_indices] > overlapThresh):
-            indices = indices[indices != i]
-
-    return boxes[indices]
-
-
-class DownloadProgressBar(tqdm):
-    def update_to(self, b=1, bsize=1, tsize=None):
-        if tsize is not None:
-            self.total = tsize
-        self.update(b * bsize - self.n)

vision.py

@@ -1,306 +0,0 @@
-# Copyright (C) 2022-2025, 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.
-
-import logging
-import os
-import platform
-import tarfile
-from typing import Sequence, Tuple
-from urllib.request import urlretrieve
-
-import numpy as np
-from PIL import Image
-
-try:
-    import ncnn
-except ImportError:
-    ncnn = None
-
-try:
-    import onnxruntime
-except ImportError:
-    onnxruntime = None
-
-try:
-    from .utils import DownloadProgressBar, box_iou, letterbox, nms, xywh2xyxy
-except ImportError:
-    from utils import DownloadProgressBar, box_iou, letterbox, nms, xywh2xyxy
-
-__all__ = ["Classifier"]
-
-MODEL_URL_FOLDER = "https://huggingface.co/pyronear/yolo11s_mighty-mongoose_v5.1.0/resolve/main/"
-MODEL_NAME = "ncnn_cpu_yolo11s_mighty-mongoose_v5.1.0.tar.gz"
-
-logging.basicConfig(format="%(asctime)s | %(levelname)s: %(message)s", level=logging.INFO, force=True)
-
-
-def _env_int(name: str, default: int) -> int:
-    try:
-        return int(os.getenv(name, str(default)))
-    except Exception:
-        return int(default)
-
-
-class Classifier:
-    """Implements an image classification model using YOLO backend.
-
-    Examples:
-        >>> from pyroengine.vision import Classifier
-        >>> model = Classifier()
-
-    Args:
-        model_path: model path
-    """
-
-    def __init__(
-        self,
-        model_folder="data",
-        imgsz=1024,
-        conf=0.15,
-        iou=0,
-        format="ncnn",
-        model_path=None,
-        max_bbox_size=0.4,
-    ) -> None:
-        if model_path:
-            if not os.path.isfile(model_path):
-                raise ValueError(f"Model file not found: {model_path}")
-            if os.path.splitext(model_path)[-1].lower() != ".onnx":
-                raise ValueError(f"Input model_path should point to an ONNX export but currently is {model_path}")
-            self.format = "onnx"
-        else:
-            if format == "ncnn":
-                if ncnn is None:
-                    raise ImportError("ncnn is required for format='ncnn'. Install ncnn or use format='onnx'.")
-                if not self.is_arm_architecture():
-                    logging.info("NCNN format is optimized for arm architecture only, switching to onnx is recommended")
-                model = MODEL_NAME
-                self.format = "ncnn"
-            elif format == "onnx":
-                if onnxruntime is None:
-                    raise ImportError("onnxruntime is required for format='onnx'. Install onnxruntime.")
-                model = MODEL_NAME.replace("ncnn", "onnx")
-                self.format = "onnx"
-            else:
-                raise ValueError("Unsupported format: should be 'ncnn' or 'onnx'")
-
-            model_path = os.path.join(model_folder, model)
-            model_url = MODEL_URL_FOLDER + model
-
-            if not os.path.isfile(model_path):
-                logging.info(f"Downloading model from {model_url} ...")
-                os.makedirs(model_folder, exist_ok=True)
-                with DownloadProgressBar(unit="B", unit_scale=True, miniters=1, desc=model_path) as t:
-                    urlretrieve(model_url, model_path, reporthook=t.update_to)
-                logging.info("Model downloaded!")
-
-            # Extract .tar.gz archive
-            if model_path.endswith(".tar.gz"):
-                base_name = os.path.basename(model_path).replace(".tar.gz", "")
-                extract_path = os.path.join(model_folder, base_name)
-                if not os.path.isdir(extract_path):
-                    with tarfile.open(model_path, "r:gz") as tar:
-                        tar.extractall(model_folder)
-                    logging.info(f"Extracted model to: {extract_path}")
-                model_path = extract_path
-
-        if self.format == "ncnn":
-            if ncnn is None:
-                raise RuntimeError("ncnn is not available; cannot load NCNN model.")
-            self.model = ncnn.Net()
-            self.model.load_param(os.path.join(model_path, "best_ncnn_model", "model.ncnn.param"))
-            self.model.load_model(os.path.join(model_path, "best_ncnn_model", "model.ncnn.bin"))
-
-        else:
-            if onnxruntime is None:
-                raise RuntimeError("onnxruntime is not available; cannot load ONNX model.")
-            try:
-                onnx_file = model_path if model_path.endswith(".onnx") else os.path.join(model_path, "best.onnx")
-                sess_options = onnxruntime.SessionOptions()
-                sess_options.graph_optimization_level = onnxruntime.GraphOptimizationLevel.ORT_ENABLE_ALL
-                sess_options.execution_mode = onnxruntime.ExecutionMode.ORT_SEQUENTIAL
-
-                default_intra_threads = max(1, int(os.cpu_count() or 1))
-                intra_threads = max(1, _env_int("ORT_INTRA_OP_NUM_THREADS", default_intra_threads))
-                inter_threads = max(1, _env_int("ORT_INTER_OP_NUM_THREADS", 1))
-                sess_options.intra_op_num_threads = intra_threads
-                sess_options.inter_op_num_threads = inter_threads
-
-                providers_env = os.getenv("ORT_PROVIDERS", "CPUExecutionProvider")
-                requested_providers = [p.strip() for p in providers_env.split(",") if p.strip()]
-                available_providers = set(onnxruntime.get_available_providers())
-                providers = [p for p in requested_providers if p in available_providers]
-                if not providers:
-                    providers = ["CPUExecutionProvider"]
-
-                self.ort_session = onnxruntime.InferenceSession(
-                    onnx_file,
-                    sess_options=sess_options,
-                    providers=providers,
-                )
-                logging.info(
-                    "ONNX Runtime config | providers=%s intra_op_threads=%d inter_op_threads=%d",
-                    providers,
-                    intra_threads,
-                    inter_threads,
-                )
-
-            except Exception as e:
-                raise RuntimeError(f"Failed to load the ONNX model from {model_path}: {e!s}") from e
-
-            logging.info(f"ONNX model loaded successfully from {model_path}")
-
-        self.imgsz = imgsz
-        self.conf = conf
-        self.iou = iou
-        self.max_bbox_size = max_bbox_size
-
-    def is_arm_architecture(self):
-        # Check for ARM architecture
-        return platform.machine().startswith("arm") or platform.machine().startswith("aarch")
-
-    def prep_process(self, pil_img: Image.Image) -> Tuple[np.ndarray, Tuple[int, int]]:
-        """Preprocess an image for inference
-
-        Args:
-            pil_img: A valid PIL image.
-
-        Returns:
-            A tuple containing:
-            - The resized and normalized image of shape (1, C, H, W).
-            - Padding information as a tuple of integers (pad_height, pad_width).
-        """
-        np_img, pad = letterbox(np.array(pil_img), self.imgsz)  # Applies letterbox resize with padding
-
-        if self.format == "ncnn":
-            np_img = ncnn.Mat.from_pixels(np_img, ncnn.Mat.PixelType.PIXEL_BGR, np_img.shape[1], np_img.shape[0])
-            mean = [0, 0, 0]
-            std = [1 / 255, 1 / 255, 1 / 255]
-            np_img.substract_mean_normalize(mean=mean, norm=std)
-        else:
-            np_img = np.expand_dims(np_img.astype("float32"), axis=0)  # Add batch dimension
-            np_img = np.ascontiguousarray(np_img.transpose((0, 3, 1, 2)))  # Convert from BHWC to BCHW format
-            np_img /= 255.0  # Normalize to [0, 1]
-
-        return np_img, pad
-
-    def post_process(self, pred: np.ndarray, pad: Tuple[int, int]) -> np.ndarray:
-        """Post-process model predictions.
-
-        Args:
-            pred: Raw predictions from the model.
-            pad: Padding information as (left_pad, top_pad).
-
-        Returns:
-            Processed predictions as a numpy array.
-        """
-        pred = pred[:, pred[-1, :] > self.conf]  # Drop low-confidence predictions
-        pred = np.transpose(pred)
-        pred = xywh2xyxy(pred)
-        pred = pred[pred[:, 4].argsort()]  # Sort by confidence
-        pred = nms(pred)
-        pred = pred[::-1]  # Reverse for highest confidence first
-
-        if len(pred) > 0:
-            left_pad, top_pad = pad  # Unpack the tuple
-            pred[:, :4:2] -= left_pad
-            pred[:, 1:4:2] -= top_pad
-            pred[:, :4:2] /= self.imgsz - 2 * left_pad
-            pred[:, 1:4:2] /= self.imgsz - 2 * top_pad
-            pred = np.clip(pred, 0, 1)
-        else:
-            pred = np.zeros((0, 5))  # Return empty prediction array
-
-        return pred
-
-    def _finalize_prediction(self, pred: np.ndarray, pad: Tuple[int, int], occlusion_bboxes: dict) -> np.ndarray:
-        # Convert pad to a tuple if required
-        if isinstance(pad, list):
-            pad = tuple(pad)
-
-        pred = self.post_process(pred, pad)  # Ensure pad is passed as a tuple
-
-        # drop big detections
-        pred = np.clip(pred, 0, 1)
-        pred = pred[(pred[:, 2] - pred[:, 0]) < self.max_bbox_size, :]
-        pred = np.reshape(pred, (-1, 5))
-
-        logging.debug("Model original pred : %s", pred)
-
-        # Remove prediction in bbox occlusion mask
-        if len(occlusion_bboxes):
-            all_boxes = np.array([b[:4] for b in occlusion_bboxes.values()], dtype=pred.dtype)
-
-            pred_boxes = pred[:, :4].astype(pred.dtype)
-            ious = box_iou(pred_boxes, all_boxes)
-            max_ious = ious.max(axis=0)
-            keep = max_ious <= 0.1
-            pred = pred[keep]
-
-        return pred
-
-    def infer_batch(self, pil_imgs: Sequence[Image.Image], occlusion_bboxes: dict = None, batch_size: int = 8):
-        if not pil_imgs:
-            return []
-
-        if occlusion_bboxes is None:
-            occlusion_bboxes = {}
-
-        # NCNN path stays single-image.
-        if self.format != "onnx":
-            return [self(pil_img, occlusion_bboxes=occlusion_bboxes) for pil_img in pil_imgs]
-
-        batch_size = max(1, int(batch_size))
-        outputs = []
-
-        for start in range(0, len(pil_imgs), batch_size):
-            chunk = pil_imgs[start : start + batch_size]
-            batch_imgs = []
-            pads = []
-            for pil_img in chunk:
-                np_img, pad = self.prep_process(pil_img)
-                batch_imgs.append(np_img)
-                pads.append(pad)
-
-            np_batch = np.concatenate(batch_imgs, axis=0)
-            raw = self.ort_session.run(["output0"], {"images": np_batch})[0]
-
-            if raw.ndim >= 3 and raw.shape[0] == len(chunk):
-                raw_preds = [raw[i] for i in range(len(chunk))]
-            elif len(chunk) == 1 and raw.ndim >= 3:
-                raw_preds = [raw[0]]
-            elif len(chunk) == 1:
-                raw_preds = [raw]
-            else:
-                # Fallback for unexpected output shapes.
-                raw_preds = [self.ort_session.run(["output0"], {"images": arr})[0][0] for arr in batch_imgs]
-
-            for raw_pred, pad in zip(raw_preds, pads):
-                outputs.append(self._finalize_prediction(raw_pred, pad, occlusion_bboxes))
-
-        return outputs
-
-    def __call__(self, pil_img: Image.Image, occlusion_bboxes: dict = {}) -> np.ndarray:
-        """Run the classifier on an input image.
-
-        Args:
-            pil_img: The input PIL image.
-            occlusion_mask: Optional occlusion mask to exclude certain areas.
-
-        Returns:
-            Processed predictions.
-        """
-        np_img, pad = self.prep_process(pil_img)
-
-        if self.format == "ncnn":
-            extractor = self.model.create_extractor()
-            extractor.set_light_mode(True)
-            extractor.input("in0", np_img)
-            pred = ncnn.Mat()
-            extractor.extract("out0", pred)
-            pred = np.asarray(pred)
-        else:
-            pred = self.ort_session.run(["output0"], {"images": np_img})[0][0]
-        return self._finalize_prediction(pred, pad, occlusion_bboxes)