feat: filter alerts/sequences/slack by daily fire risk score (#583)

* feat: add risk-api client and sequence confidence helper

* feat: schedule daily risk-api refresh in app lifespan

* feat: filter alerts and sequences by risk-driven confidence threshold

* feat: skip slack alert when sequence max conf below risk threshold

* feat: query risk-api per date for from_date endpoints

* test: cover risk-driven filtering of alerts and sequences

* chore: apply ruff fixes

* fix: clamp risk-refresh hour to 0..23 to avoid retry loop

* feat: scope from_date risk lookup to caller organization, normalize fwi class casing

* chore: rename risk-api env vars to RISK_API_URL/LOGIN/PWD

* feat: add max_conf column to sequences with backfill migration

* feat: maintain sequence max_conf at ingest with atomic update

* refactor: read max_conf from sequence row instead of parsing detections

* test: seed max_conf directly on test sequences

* fix: portable max_conf bump and validate fwi thresholds in [0,1]

* feat: add risk_score query param to override fwi class on alerts endpoints

* refactor: collapse risk filter helpers into one and use literal type for risk_score

* fix: restore refresh() cache-replace on empty list and harden payload parser

* feat: push risk filter into SQL WHERE for exact pagination

* refactor: replace fwi conf settings with FWI_MIN_CONF dict in risk module

* test: pagination on /sequences/all/fromdate keeps page full when filter applies

* fix: drop max_conf clause collapse and route pagination test through override

* fix: compute sequence max_conf from primary bbox only, ignore sibling detections

* chore: apply ruff format

* fix: silence mypy on case() and chained where() over join()

* fix: annotate case() result as Any to satisfy mypy and reformat with ruff

* test: cover _seconds_until_next_utc_hour and risk_score override on /sequences/*

* chore: address risk filter review comments

* fix: satisfy mypy for risk filter queries

* test: cover risk refresh lifecycle

* test: cover sequence risk filter endpoints

* test: cover risk service http paths

* test: cover alerts/fromdate risk filter and mixed-seq alert

* test: parametrize keep-all assertion across moderate/high/very_high/extreme

* test: pin fail-open on null max_conf and unknown cameras

* test: parametrize alerts no-filter override across moderate/high/very_high/extreme

Author: MateoLostanlen

.env.example

@@ -23,6 +23,12 @@ POSTHOG_KEY=
 SUPPORT_EMAIL=
 TELEGRAM_TOKEN=
 
+# Risk API (daily fire-weather index per camera)
+RISK_API_URL=
+RISK_API_LOGIN=
+RISK_API_PWD=
+RISK_REFRESH_HOUR_UTC=4
+
 # Production-only
 ACME_EMAIL=
 BACKEND_HOST=

docker-compose.yml

@@ -65,6 +65,10 @@ services:
       - S3_PROXY_URL=${S3_PROXY_URL}
       - SERVER_NAME=${SERVER_NAME}
       - PLATFORM_URL=${PLATFORM_URL:-https://platform.pyronear.org}
+      - RISK_API_URL=${RISK_API_URL}
+      - RISK_API_LOGIN=${RISK_API_LOGIN}
+      - RISK_API_PWD=${RISK_API_PWD}
+      - RISK_REFRESH_HOUR_UTC=${RISK_REFRESH_HOUR_UTC:-4}
     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'"

src/app/api/api_v1/endpoints/alerts.py

@@ -9,17 +9,20 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status
 from sqlalchemy import asc, desc
+from sqlalchemy.sql import ColumnElement
 from sqlmodel import delete, func, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.api.dependencies import get_alert_crud, get_jwt
 from app.core.time import utcnow
 from app.crud import AlertCRUD
 from app.db import get_session
-from app.models import Alert, AlertSequence, Sequence, UserRole
+from app.models import Alert, AlertSequence, Camera, Sequence, UserRole
 from app.schemas.alerts import AlertReadWithSequences
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceRead
+from app.services.risk import FwiClass, risk_service
+from app.services.sequence_confidence import max_conf_filter_clause
 from app.services.sequence_counts import get_detection_counts_by_sequence_ids
 from app.services.telemetry import telemetry_client
 
@@ -31,22 +34,44 @@ def verify_org_rights(organization_id: int, alert: Alert) -> None:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
 
 
-async def _fetch_sequences_by_alert_ids(session: AsyncSession, alert_ids: List[int]) -> Dict[int, List[Sequence]]:
+async def _fetch_sequences_by_alert_ids(
+    session: AsyncSession,
+    alert_ids: List[int],
+    seq_filter: Union[ColumnElement[bool], None] = None,
+) -> Dict[int, List[Sequence]]:
     mapping: Dict[int, List[Sequence]] = {}
     if not alert_ids:
         return mapping
     seq_stmt: Any = (
         select(AlertSequence.alert_id, Sequence)
         .join(Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id))
         .where(AlertSequence.alert_id.in_(alert_ids))  # type: ignore[attr-defined]
-        .order_by(cast(Any, AlertSequence.alert_id), desc(cast(Any, Sequence.last_seen_at)))
     )
+    if seq_filter is not None:
+        seq_stmt = seq_stmt.where(seq_filter)
+    seq_stmt = seq_stmt.order_by(cast(Any, AlertSequence.alert_id), desc(cast(Any, Sequence.last_seen_at)))
     res = await session.exec(seq_stmt)
     for alert_id, sequence in res.all():
         mapping.setdefault(int(alert_id), []).append(sequence)
     return mapping
 
 
+async def _resolve_fwi_class_per_camera(
+    session: AsyncSession,
+    organization_id: int,
+    target_date: Union[date, None] = None,
+    override_class: Union[str, None] = None,
+) -> Dict[int, Union[str, None]]:
+    """Resolve ``{camera_id: fwi_class}`` for the org, picking override -> per-date -> today's cache."""
+    if override_class is not None:
+        cam_ids = (await session.exec(select(Camera.id).where(Camera.organization_id == organization_id))).all()
+        return dict.fromkeys(cam_ids, override_class)
+    if target_date is not None:
+        scores = await risk_service.get_scores_for_date(target_date, organization_id=organization_id)
+        return {cid: cls for cid, cls in scores.items()}
+    return {cid: cls for cid, cls in risk_service.scores().items()}
+
+
 def _serialize_sequence(sequence: Sequence, detections_count: int = 0) -> SequenceRead:
     return SequenceRead(**sequence.model_dump(), detections_count=detections_count)
 
@@ -113,24 +138,39 @@ async def fetch_alert_sequences(
     summary="Fetch all the alerts with unlabeled sequences from the last 24 hours",
 )
 async def fetch_latest_unlabeled_alerts(
+    risk_score: Union[FwiClass, None] = Query(
+        None, description="Override FWI class applied to every sequence; bypasses risk-api lookup."
+    ),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[AlertReadWithSequences]:
     telemetry_client.capture(token_payload.sub, event="alerts-fetch-latest")
 
-    alerts_stmt: Any = select(Alert).join(AlertSequence, cast(Any, AlertSequence.alert_id == Alert.id))
-    alerts_stmt = alerts_stmt.join(Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id))
-    alerts_stmt = (
-        alerts_stmt.where(Alert.organization_id == token_payload.organization_id)
-        .where(Sequence.last_seen_at > utcnow() - timedelta(hours=24))
-        .where(Sequence.is_wildfire.is_(None))  # type: ignore[union-attr]
+    fwi_classes_by_camera = await _resolve_fwi_class_per_camera(
+        session, token_payload.organization_id, override_class=risk_score
+    )
+    seq_filter = max_conf_filter_clause(fwi_classes_by_camera)
+
+    seq_match: Any = cast(
+        Any,
+        select(AlertSequence.alert_id).join(Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id)),
+    )
+    seq_match = (
+        seq_match.where(Sequence.last_seen_at > utcnow() - timedelta(hours=24)).where(Sequence.is_wildfire.is_(None))  # type: ignore[union-attr]
+    )
+    if seq_filter is not None:
+        seq_match = seq_match.where(seq_filter)
+
+    alerts_stmt: Any = (
+        select(Alert)
+        .where(Alert.organization_id == token_payload.organization_id)
+        .where(cast(Any, Alert.id).in_(seq_match))
         .order_by(Alert.started_at.desc())  # type: ignore[attr-defined]
         .limit(15)
     )
-    alerts_res = await session.exec(alerts_stmt)
-    alerts = alerts_res.unique().all()
+    alerts = list((await session.exec(alerts_stmt)).all())
     alert_ids = [alert.id for alert in alerts]
-    seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids)
+    seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids, seq_filter)
     detection_counts = await get_detection_counts_by_sequence_ids(
         session,
         list({sequence.id for sequences in seq_map.values() for sequence in sequences}),
@@ -143,23 +183,35 @@ async def fetch_alerts_from_date(
     from_date: date = Query(),
     limit: Union[int, None] = Query(15, description="Maximum number of alerts to fetch"),
     offset: Union[int, None] = Query(0, description="Number of alerts to skip before starting to fetch"),
+    risk_score: Union[FwiClass, None] = Query(
+        None, description="Override FWI class applied to every sequence; bypasses risk-api lookup."
+    ),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[AlertReadWithSequences]:
     telemetry_client.capture(token_payload.sub, event="alerts-fetch-from-date")
 
+    fwi_classes_by_camera = await _resolve_fwi_class_per_camera(
+        session, token_payload.organization_id, target_date=from_date, override_class=risk_score
+    )
+    seq_filter = max_conf_filter_clause(fwi_classes_by_camera)
+
     alerts_stmt: Any = (
         select(Alert)
         .where(Alert.organization_id == token_payload.organization_id)
         .where(func.date(Alert.started_at) == from_date)
-        .order_by(Alert.started_at.desc())  # type: ignore[attr-defined]
-        .limit(limit)
-        .offset(offset)
     )
-    alerts_res = await session.exec(alerts_stmt)
-    alerts = alerts_res.all()
+    if seq_filter is not None:
+        seq_match: Any = select(AlertSequence.alert_id).join(
+            Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id)
+        )
+        seq_match = seq_match.where(seq_filter)
+        alerts_stmt = alerts_stmt.where(cast(Any, Alert.id).in_(seq_match))
+    alerts_stmt = alerts_stmt.order_by(Alert.started_at.desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
+
+    alerts = list((await session.exec(alerts_stmt)).all())
     alert_ids = [alert.id for alert in alerts]
-    seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids)
+    seq_map = await _fetch_sequences_by_alert_ids(session, alert_ids, seq_filter)
     detection_counts = await get_detection_counts_by_sequence_ids(
         session,
         list({sequence.id for sequences in seq_map.values() for sequence in sequences}),

src/app/api/api_v1/endpoints/detections.py

@@ -5,6 +5,7 @@
 
 
 import json
+import logging
 import re
 from ast import literal_eval
 from datetime import datetime, timedelta
@@ -56,11 +57,15 @@
 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")
+
 router = APIRouter()
 
 
@@ -427,6 +432,10 @@ async def create_detection(
         if matched_sequence is not None:
             await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at))
             det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id))
+            # Only the primary bbox tracks the sequence; siblings in others_bboxes are unrelated detections.
+            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)
         else:
             det_filters: List[tuple[str, Any]] = [
                 ("camera_id", token_payload.sub),
@@ -455,6 +464,7 @@ async def create_detection(
             if len(overlapping_dets) >= settings.SEQUENCE_MIN_INTERVAL_DETS:
                 first_det = min(overlapping_dets, key=lambda item: item.created_at)
                 cone_azimuth, cone_angle = resolve_cone(pose.azimuth, first_det.bbox, camera.angle_of_view)
+                seq_max_conf = max_conf_from_bboxes(*[d.bbox for d in overlapping_dets])
                 sequence_ = await sequences.create(
                     Sequence(
                         camera_id=token_payload.sub,
@@ -464,6 +474,7 @@ async def create_detection(
                         cone_angle=cone_angle,
                         started_at=first_det.created_at,
                         last_seen_at=det.created_at,
+                        max_conf=seq_max_conf,
                     )
                 )
                 for det_ in overlapping_dets:
@@ -490,11 +501,20 @@ async def create_detection(
                     if org is None:
                         org = cast(Organization, await organizations.get(token_payload.organization_id, strict=True))
                     if org.slack_hook:
-                        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
-                        )
+                        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,
+                            )
 
         created.append(det)
 

src/app/api/api_v1/endpoints/sequences.py

@@ -22,6 +22,8 @@
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.overlap import compute_overlap
+from app.services.risk import FwiClass, risk_service
+from app.services.sequence_confidence import max_conf_filter_clause
 from app.services.sequence_counts import get_detection_counts_by_sequence_ids
 from app.services.storage import s3_service
 from app.services.telemetry import telemetry_client
@@ -146,22 +148,32 @@ async def fetch_sequence_detections(
     summary="Fetch all the unlabeled sequences from the last 24 hours",
 )
 async def fetch_latest_unlabeled_sequences(
+    risk_score: Union[FwiClass, None] = Query(
+        None, description="Override FWI class applied to every sequence; bypasses risk-api lookup."
+    ),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[SequenceRead]:
     telemetry_client.capture(token_payload.sub, event="sequence-fetch-latest")
-    camera_ids = await session.exec(select(Camera.id).where(Camera.organization_id == token_payload.organization_id))
-
-    fetched_sequences = (
-        await session.exec(
-            select(Sequence)
-            .where(Sequence.started_at > utcnow() - timedelta(hours=24))
-            .where(Sequence.camera_id.in_(camera_ids.all()))  # type: ignore[attr-defined]
-            .where(Sequence.is_wildfire.is_(None))  # type: ignore[union-attr]
-            .order_by(Sequence.started_at.desc())  # type: ignore[attr-defined]
-            .limit(15)
-        )
+    camera_ids = (
+        await session.exec(select(Camera.id).where(Camera.organization_id == token_payload.organization_id))
     ).all()
+    classes: dict[int, Union[str, None]] = (
+        dict.fromkeys(camera_ids, risk_score) if risk_score is not None else dict(risk_service.scores())
+    )
+
+    stmt: Any = (
+        select(Sequence)
+        .where(Sequence.started_at > utcnow() - timedelta(hours=24))
+        .where(Sequence.camera_id.in_(camera_ids))  # type: ignore[attr-defined]
+        .where(Sequence.is_wildfire.is_(None))  # type: ignore[union-attr]
+    )
+    seq_filter = max_conf_filter_clause(classes)
+    if seq_filter is not None:
+        stmt = stmt.where(seq_filter)
+    stmt = stmt.order_by(Sequence.started_at.desc()).limit(15)  # type: ignore[attr-defined]
+
+    fetched_sequences = (await session.exec(stmt)).all()
     counts = await get_detection_counts_by_sequence_ids(session, [sequence.id for sequence in fetched_sequences])
     return [_serialize_sequence(sequence, counts.get(sequence.id, 0)) for sequence in fetched_sequences]
 
@@ -171,23 +183,32 @@ async def fetch_sequences_from_date(
     from_date: date = Query(),
     limit: Union[int, None] = Query(15, description="Maximum number of sequences to fetch"),
     offset: Union[int, None] = Query(0, description="Number of sequences to skip before starting to fetch"),
+    risk_score: Union[FwiClass, None] = Query(
+        None, description="Override FWI class applied to every sequence; bypasses risk-api lookup."
+    ),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[SequenceRead]:
     telemetry_client.capture(token_payload.sub, event="sequence-fetch-from-date")
     # Limit to cameras in the same organization
-    camera_ids = await session.exec(select(Camera.id).where(Camera.organization_id == token_payload.organization_id))
-    # Identify the sequences from that day
-    fetched_sequences = (
-        await session.exec(
-            select(Sequence)
-            .where(func.date(Sequence.started_at) == from_date)
-            .where(Sequence.camera_id.in_(camera_ids.all()))  # type: ignore[attr-defined]
-            .order_by(Sequence.started_at.desc())  # type: ignore[attr-defined]
-            .limit(limit)
-            .offset(offset)
-        )
+    camera_ids = (
+        await session.exec(select(Camera.id).where(Camera.organization_id == token_payload.organization_id))
     ).all()
+    if risk_score is not None:
+        classes: dict[int, Union[str, None]] = dict.fromkeys(camera_ids, risk_score)
+    else:
+        scores = await risk_service.get_scores_for_date(from_date, organization_id=token_payload.organization_id)
+        classes = dict(scores)
+
+    stmt: Any = (
+        select(Sequence).where(func.date(Sequence.started_at) == from_date).where(Sequence.camera_id.in_(camera_ids))  # type: ignore[attr-defined]
+    )
+    seq_filter = max_conf_filter_clause(classes)
+    if seq_filter is not None:
+        stmt = stmt.where(seq_filter)
+    stmt = stmt.order_by(Sequence.started_at.desc()).limit(limit).offset(offset)  # type: ignore[attr-defined]
+
+    fetched_sequences = (await session.exec(stmt)).all()
     counts = await get_detection_counts_by_sequence_ids(session, [sequence.id for sequence in fetched_sequences])
     return [_serialize_sequence(sequence, counts.get(sequence.id, 0)) for sequence in fetched_sequences]
 

src/app/core/config.py

@@ -77,6 +77,12 @@ 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", "")
 
+    # 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")
+    RISK_API_PWD: Union[str, None] = os.environ.get("RISK_API_PWD")
+    RISK_REFRESH_HOUR_UTC: int = int(os.environ.get("RISK_REFRESH_HOUR_UTC") or 4)
+
     # Error monitoring
     SENTRY_DSN: Union[str, None] = os.environ.get("SENTRY_DSN")
     SERVER_NAME: str = os.environ.get("SERVER_NAME", socket.gethostname())

src/app/crud/crud_sequence.py

@@ -4,8 +4,9 @@
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
 
-from typing import Union
+from typing import Any, Union, cast
 
+from sqlalchemy import case, or_, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.crud.base import BaseCRUD
@@ -18,3 +19,17 @@
 class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLabel]]):
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Sequence)
+
+    async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
+        """Atomically raise sequences.max_conf to candidate if higher (or set if NULL).
+
+        Uses a portable CASE expression so it runs on SQLite as well as Postgres.
+        """
+        max_conf_col = cast(Any, Sequence.max_conf)
+        bumped: Any = cast(Any, case)(
+            (or_(max_conf_col.is_(None), max_conf_col < candidate), candidate),
+            else_=max_conf_col,
+        )
+        stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
+        await self.session.exec(stmt)
+        await self.session.commit()