pyronear
diff --git a/‎.env.example‎
Lines changed: 9 additions & 0 deletions b/‎.env.example‎
Lines changed: 9 additions & 0 deletions
diff --git a/‎docker-compose.yml‎
Lines changed: 7 additions & 0 deletions b/‎docker-compose.yml‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎src/app/api/api_v1/endpoints/detections.py‎
Lines changed: 16 additions & 50 deletions b/‎src/app/api/api_v1/endpoints/detections.py‎
Lines changed: 16 additions & 50 deletions
diff --git a/‎src/app/core/config.py‎
Lines changed: 18 additions & 0 deletions b/‎src/app/core/config.py‎
Lines changed: 18 additions & 0 deletions
diff --git a/‎src/app/crud/crud_sequence.py‎
Lines changed: 167 additions & 2 deletions b/‎src/app/crud/crud_sequence.py‎
Lines changed: 167 additions & 2 deletions
@@ -29,6 +29,15 @@ RISK_API_LOGIN=
 RISK_API_PWD=
 RISK_REFRESH_HOUR_UTC=4
 
+# Temporal model API (validates sequences before triangulation)
+TEMPORAL_API_URL=
+TEMPORAL_API_TOKEN=
+TEMPORAL_MODEL_THRESHOLD=0.45
+TEMPORAL_API_TIMEOUT=30
+TEMPORAL_VALIDATION_POLL_SECONDS=2
+TEMPORAL_VALIDATION_MAX_AGE=300
+TEMPORAL_VALIDATION_LEASE_SECONDS=120
+
 # Production-only
 ACME_EMAIL=
 BACKEND_HOST=
 
@@ -71,6 +71,13 @@ services:
       - RISK_API_LOGIN=${RISK_API_LOGIN}
       - RISK_API_PWD=${RISK_API_PWD}
       - RISK_REFRESH_HOUR_UTC=${RISK_REFRESH_HOUR_UTC:-4}
+      - TEMPORAL_API_URL=${TEMPORAL_API_URL}
+      - TEMPORAL_API_TOKEN=${TEMPORAL_API_TOKEN}
+      - TEMPORAL_MODEL_THRESHOLD=${TEMPORAL_MODEL_THRESHOLD:-0.45}
+      - TEMPORAL_API_TIMEOUT=${TEMPORAL_API_TIMEOUT:-30}
+      - TEMPORAL_VALIDATION_POLL_SECONDS=${TEMPORAL_VALIDATION_POLL_SECONDS:-2}
+      - TEMPORAL_VALIDATION_MAX_AGE=${TEMPORAL_VALIDATION_MAX_AGE:-300}
+      - TEMPORAL_VALIDATION_LEASE_SECONDS=${TEMPORAL_VALIDATION_LEASE_SECONDS:-120}
     volumes:
       - ./src/:/app/
     command: "sh -c 'alembic upgrade head && python app/db.py && uvicorn app.main:app --reload --host 0.0.0.0 --port 5050 --proxy-headers'"
 
@@ -4,7 +4,6 @@
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
 
-import json
 import logging
 import re
 from ast import literal_eval
@@ -14,7 +13,6 @@
 import pandas as pd
 from fastapi import (
     APIRouter,
-    BackgroundTasks,
     Depends,
     File,
     Form,
@@ -24,25 +22,20 @@
     UploadFile,
     status,
 )
-from fastapi.encoders import jsonable_encoder
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.api.dependencies import (
-    dispatch_webhook,
-    get_alert_crud,
     get_camera_crud,
     get_detection_crud,
     get_jwt,
-    get_organization_crud,
     get_pose_crud,
     get_sequence_crud,
-    get_webhook_crud,
 )
 from app.core.config import settings
 from app.core.time import utcnow
-from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, OrganizationCRUD, PoseCRUD, SequenceCRUD, WebhookCRUD
-from app.models import Alert, AlertSequence, Camera, Detection, Organization, Pose, Role, Sequence, UserRole
+from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, PoseCRUD, SequenceCRUD
+from app.models import Alert, AlertSequence, Camera, Detection, Pose, Role, Sequence, UserRole
 from app.schemas.alerts import AlertCreate, AlertUpdate
 from app.schemas.detections import (
     BOX_PATTERN,
@@ -57,11 +50,8 @@
 from app.schemas.sequences import SequenceUpdate
 from app.services.cones import resolve_cone
 from app.services.overlap import compute_overlap, haversine_km
-from app.services.risk import risk_service
 from app.services.sequence_confidence import max_conf_from_bboxes
-from app.services.slack import slack_client
 from app.services.storage import s3_service, upload_file
-from app.services.telegram import telegram_client
 from app.services.telemetry import telemetry_client
 
 logger = logging.getLogger("uvicorn.error")
@@ -155,6 +145,9 @@ def _build_overlap_records(
         cam = camera_by_id.get(seq.camera_id)
         if cam is None or seq.sequence_azimuth is None or seq.cone_angle is None:
             continue
+        # Only validated sequences are eligible for triangulation (as target or partner).
+        if not seq.is_validated:
+            continue
         records.append({
             "id": int(seq.id),
             "pose_id": seq.pose_id,
@@ -349,7 +342,6 @@ async def _attach_sequence_to_alert(
 
 @router.post("/", status_code=status.HTTP_201_CREATED, summary="Register a new wildfire detection")
 async def create_detection(
-    background_tasks: BackgroundTasks,
     bboxes: str = Form(
         ...,
         description="string representation of list of detection localizations, each represented as a tuple of relative coords (max 3 decimals) in order: xmin, ymin, xmax, ymax, conf",
@@ -361,10 +353,7 @@ async def create_detection(
     file: UploadFile = File(..., alias="file"),
     crop_file: Optional[UploadFile] = File(None, alias="crop"),
     detections: DetectionCRUD = Depends(get_detection_crud),
-    webhooks: WebhookCRUD = Depends(get_webhook_crud),
-    organizations: OrganizationCRUD = Depends(get_organization_crud),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
-    alerts: AlertCRUD = Depends(get_alert_crud),
     cameras: CameraCRUD = Depends(get_camera_crud),
     poses: PoseCRUD = Depends(get_pose_crud),
     token_payload: TokenPayload = Security(get_jwt, scopes=[Role.CAMERA]),
@@ -399,6 +388,8 @@ async def create_detection(
 
     created: List[Detection] = []
     camera = cast(Camera, await cameras.get(token_payload.sub, strict=True))
+    # sequences touched by this request, to mark due for validation (DB-backed queue).
+    affected_sequences: Set[int] = set()
 
     for idx, bbox_str in enumerate(bbox_strings):
         single_bboxes = _bbox_list_to_str([bbox_str])
@@ -444,6 +435,7 @@ async def create_detection(
             det_max_conf = max_conf_from_bboxes(det.bbox)
             if det_max_conf is not None:
                 await sequences.bump_max_conf(matched_sequence.id, det_max_conf)
+            affected_sequences.add(matched_sequence.id)
         else:
             det_filters: List[tuple[str, Any]] = [
                 ("camera_id", token_payload.sub),
@@ -489,43 +481,17 @@ async def create_detection(
                     updated = await detections.update(det_.id, DetectionSequence(sequence_id=sequence_.id))
                     if det_.id == det.id:
                         det = updated
-
-                alert_id = await _attach_sequence_to_alert(sequence_, camera, cameras, sequences, alerts)
-
-                # Webhooks
-                whs = await webhooks.fetch_all()
-                if any(whs):
-                    for webhook in await webhooks.fetch_all():
-                        background_tasks.add_task(dispatch_webhook, webhook.url, det)
-
-                org = None
-                # Telegram notifications
-                if telegram_client.is_enabled:
-                    org = cast(Organization, await organizations.get(token_payload.organization_id, strict=True))
-                    if org.telegram_id:
-                        background_tasks.add_task(telegram_client.notify, org.telegram_id, det.model_dump_json())
-
-                if slack_client.is_enabled:
-                    if org is None:
-                        org = cast(Organization, await organizations.get(token_payload.organization_id, strict=True))
-                    if org.slack_hook:
-                        min_conf = risk_service.min_confidence(camera.id)
-                        if min_conf is None or sequence_.max_conf is None or sequence_.max_conf >= min_conf:
-                            slack_payload = jsonable_encoder(det)
-                            slack_payload["sequence_azimuth"] = sequence_.sequence_azimuth
-                            background_tasks.add_task(
-                                slack_client.notify, org.slack_hook, json.dumps(slack_payload), camera.name, alert_id
-                            )
-                        else:
-                            logger.info(
-                                "Skipping Slack notification for camera %s: max conf %.3f < threshold %.3f",
-                                camera.name,
-                                sequence_.max_conf,
-                                min_conf,
-                            )
+                affected_sequences.add(sequence_.id)
 
         created.append(det)
 
+    # Mark touched sequences due for validation (idempotent: one queue entry per sequence,
+    # whichever uvicorn worker received the detection). The per-process validation worker
+    # claims due sequences from the DB and runs the gated pipeline: triangulation and ALL
+    # notification channels (webhooks, Telegram, Slack) fire only once validated.
+    for seq_id in affected_sequences:
+        await sequences.enqueue_validation(seq_id)
+
     first_det = cast(Detection, await detections.get(created[0].id, strict=True))
     return DetectionRead(**first_det.model_dump())
 
 
@@ -77,6 +77,24 @@ def sqlachmey_uri(cls, v: str) -> str:
     TELEGRAM_TOKEN: Union[str, None] = os.environ.get("TELEGRAM_TOKEN")
     PLATFORM_URL: str = os.environ.get("PLATFORM_URL", "")
 
+    # Temporal model API (validates sequences from their frames)
+    TEMPORAL_API_URL: Union[str, None] = os.environ.get("TEMPORAL_API_URL")
+    # Shared bearer token for /predict; empty = server has auth disabled, send no header.
+    TEMPORAL_API_TOKEN: Union[str, None] = os.environ.get("TEMPORAL_API_TOKEN") or None
+    TEMPORAL_MODEL_THRESHOLD: float = float(os.environ.get("TEMPORAL_MODEL_THRESHOLD") or 0.45)
+    # Generous timeout: the temporal API serializes inference server-side, so with N uvicorn
+    # workers a call can wait behind N-1 others; keep N * model latency under this value.
+    TEMPORAL_API_TIMEOUT: float = float(os.environ.get("TEMPORAL_API_TIMEOUT") or 30.0)
+    # Validation worker (one loop per uvicorn process, coordinated through the DB):
+    # idle poll interval for due sequences,
+    TEMPORAL_VALIDATION_POLL_SECONDS: float = float(os.environ.get("TEMPORAL_VALIDATION_POLL_SECONDS") or 2.0)
+    # max time a sequence may wait in the queue before failing open on the risk gate alone
+    # (bounds validation latency under a backlog; traced as validation_status=fail_open_stale),
+    TEMPORAL_VALIDATION_MAX_AGE: float = float(os.environ.get("TEMPORAL_VALIDATION_MAX_AGE") or 300.0)
+    # and how long a claimed job is leased before a sibling worker may retry it (must exceed
+    # TEMPORAL_API_TIMEOUT plus the DB phases).
+    TEMPORAL_VALIDATION_LEASE_SECONDS: float = float(os.environ.get("TEMPORAL_VALIDATION_LEASE_SECONDS") or 120.0)
+
     # Risk API (daily fire-weather index per camera)
     RISK_API_URL: Union[str, None] = os.environ.get("RISK_API_URL")
     RISK_API_LOGIN: Union[str, None] = os.environ.get("RISK_API_LOGIN")
 
@@ -4,17 +4,23 @@
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
 
+import logging
+from datetime import timedelta
 from typing import Any, Union, cast
 
-from sqlalchemy import case, or_, update
+from sqlalchemy import case, distinct, func, null, or_, select, update
+from sqlmodel import select as select_model
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from app.core.time import utcnow
 from app.crud.base import BaseCRUD
-from app.models import Sequence
+from app.models import TERMINAL_VALIDATION_STATUSES, VALIDATION_FAILED, Detection, Sequence
 from app.schemas.sequences import SequenceLabel, SequenceUpdate
 
 __all__ = ["SequenceCRUD"]
 
+logger = logging.getLogger("uvicorn.error")
+
 
 class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLabel]]):
     def __init__(self, session: AsyncSession) -> None:
@@ -33,3 +39,162 @@ async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
         stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
         await self.session.exec(stmt)
         await self.session.commit()
+
+    async def set_temporal_score(
+        self,
+        sequence_id: int,
+        score: float,
+        model_version: Union[str, None] = None,
+        api_version: Union[str, None] = None,
+    ) -> None:
+        """Persist the latest temporal-model score with its provenance.
+
+        The versions are written unconditionally in the same UPDATE: they describe the
+        stored score, so the triple always moves together (a re-score by a newer release
+        overwrites all three).
+        """
+        stmt: Any = (
+            update(Sequence)
+            .where(cast(Any, Sequence.id) == sequence_id)
+            .values(
+                temporal_model_score=score,
+                temporal_model_version=model_version,
+                temporal_api_version=api_version,
+            )
+        )
+        await self.session.exec(stmt)
+        await self.session.commit()
+
+    async def claim_validation(self, sequence_id: int, validation_status: Union[str, None] = None) -> bool:
+        """Atomically flip ``is_validated`` from False to True, recording how it concluded.
+
+        Returns True only for the caller that won the flip, so concurrent workers for the
+        same sequence don't both triangulate and notify. ``validation_status`` is written in
+        the same UPDATE so the verdict and its label are durable together: post-claim work
+        (triangulation, notifications) that fails or dies is resumed without re-deciding.
+        """
+        id_col = cast(Any, Sequence.id)
+        validated_col = cast(Any, Sequence.is_validated)
+        values: dict = {"is_validated": True}
+        if validation_status is not None:
+            values["validation_status"] = validation_status
+        stmt: Any = update(Sequence).where(id_col == sequence_id).where(validated_col.is_(False)).values(**values)
+        result = await self.session.exec(stmt)
+        await self.session.commit()
+        return bool(getattr(result, "rowcount", 0))
+
+    async def enqueue_validation(self, sequence_id: int) -> None:
+        """Mark the sequence as due for temporal validation (the DB-backed queue).
+
+        Idempotent and FIFO-preserving: ``COALESCE`` keeps the oldest due timestamp, so a
+        sequence already queued is NOT re-queued (one entry per sequence, whichever worker
+        receives the detection). No-op for validated sequences and terminal states
+        (window-exhausted, failed).
+        """
+        status_col = cast(Any, Sequence.validation_status)
+        due_col = cast(Any, Sequence.validation_due_at)
+        stmt: Any = (
+            update(Sequence)
+            .where(cast(Any, Sequence.id) == sequence_id)
+            .where(cast(Any, Sequence.is_validated).is_(False))
+            .where(or_(status_col.is_(None), status_col.not_in(TERMINAL_VALIDATION_STATUSES)))
+            .values(validation_due_at=func.coalesce(due_col, utcnow()))
+        )
+        await self.session.exec(stmt)
+        await self.session.commit()
+
+    async def claim_due_validation(self, lease_seconds: float) -> Union[Sequence, None]:
+        """Claim the oldest due sequence for validation, or None when nothing is due.
+
+        ``FOR UPDATE SKIP LOCKED`` keeps concurrent workers (multi-worker uvicorn) off the
+        same row; the lease keeps them off for the duration of the model call, which runs
+        long after this transaction commits. ``validation_due_at`` is intentionally NOT
+        cleared here: a worker dying mid-job leaves a due row whose lease expires, so the
+        job is picked up again instead of being lost. Validated rows are NOT filtered out:
+        a still-due validated row means a worker died after winning the validation claim
+        but before triangulating/notifying, and the job must be resumed.
+        """
+        now = utcnow()
+        due_col = cast(Any, Sequence.validation_due_at)
+        lease_col = cast(Any, Sequence.validation_lease_until)
+        stmt: Any = (
+            select_model(Sequence)
+            .where(due_col.is_not(None))
+            .where(due_col <= now)
+            .where(or_(lease_col.is_(None), lease_col < now))
+            .order_by(due_col)
+            .limit(1)
+            .with_for_update(skip_locked=True)
+        )
+        res = await self.session.exec(stmt)
+        sequence_ = res.first()
+        if sequence_ is None:
+            await self.session.commit()
+            return None
+        sequence_.validation_lease_until = now + timedelta(seconds=lease_seconds)
+        self.session.add(sequence_)
+        await self.session.commit()
+        await self.session.refresh(sequence_)
+        return sequence_
+
+    async def finish_validation_job(
+        self,
+        sequence_id: int,
+        frame_count: Union[int, None] = None,
+        validation_status: Union[str, None] = None,
+    ) -> None:
+        """Release the lease and clear the due marker, completing the job.
+
+        With ``frame_count`` set, the due marker is only cleared if the sequence still has
+        exactly that many distinct frames — frames that arrived while the model was scoring
+        keep the job due, so the worker re-runs it with the fresh frame set instead of
+        waiting for (or losing, if the sequence just ended) the next detection. The
+        comparison runs inside the UPDATE so it can't race a concurrent enqueue.
+        ``validation_status`` records how validation concluded (observability, incl.
+        explicit fail-open reasons) in the same UPDATE.
+
+        Known limitation: the count is a frame-set *proxy*. A detection landing on an
+        already-seen bucket_key (changing the ROI but not the count) is not detected; the
+        next detection re-enqueues within the camera cadence, which is good enough at the
+        expected volume.
+        """
+        due_col = cast(Any, Sequence.validation_due_at)
+        if frame_count is None:
+            new_due: Any = null()
+        else:
+            count_select: Any = select(func.count(distinct(cast(Any, Detection.bucket_key)))).where(
+                cast(Any, Detection.sequence_id) == sequence_id
+            )
+            count_sq = count_select.scalar_subquery()
+            new_due = cast(Any, case)((count_sq == frame_count, null()), else_=due_col)
+        # Completing a job also resets the consecutive-error counter, so the dead-letter
+        # cap counts consecutive failures, not lifetime ones.
+        values: dict = {"validation_due_at": new_due, "validation_lease_until": None, "validation_attempts": 0}
+        if validation_status is not None:
+            values["validation_status"] = validation_status
+        stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(**values)
+        await self.session.exec(stmt)
+        await self.session.commit()
+
+    async def fail_or_retry_validation(self, sequence_id: int, *, max_attempts: int, retry_in_seconds: float) -> None:
+        """Error path: release the lease and either back off the retry or dead-letter.
+
+        Increments the consecutive-error counter; below ``max_attempts`` the job stays due,
+        pushed ``retry_in_seconds`` into the future (no tight retry loop). At the cap the
+        job dead-letters: terminal ``validation_status='failed'``, due cleared, never
+        retried nor re-enqueued — a poison job must not starve the serial worker forever.
+        Note the staleness fail-open can't bound this (each retry refreshes the due time),
+        hence the explicit attempts cap.
+        """
+        sequence_ = cast(Sequence, await self.get(sequence_id, strict=True))
+        attempts = (sequence_.validation_attempts or 0) + 1
+        values: dict = {"validation_attempts": attempts, "validation_lease_until": None}
+        if attempts >= max_attempts:
+            values["validation_status"] = VALIDATION_FAILED
+            values["validation_due_at"] = None
+            logger.error("Sequence %s failed validation %d times; giving up", sequence_id, attempts)
+        else:
+            values["validation_due_at"] = utcnow() + timedelta(seconds=retry_in_seconds)
+        stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(**values)
+        await self.session.exec(stmt)
+        await self.session.commit()