feat: accept the temporal score computed on the device

client/pyroclient/client.py

@@ -358,6 +358,7 @@ def create_detection(
         pose_id: int,
         crops: List[bytes] | None = None,
         recorded_at: str | None = None,
+        temporal_model_scores: List[float] | None = None,
     ) -> Response:
         """Notify the detection of a wildfire on the picture taken by a camera.
 
@@ -377,6 +378,9 @@ def create_detection(
                 Each crop frames a single object, so its length must equal that of `bboxes`.
             recorded_at: optional ISO 8601 timestamp of when the image was captured on-device.
                 Timezone-aware values are converted to UTC server-side; defaults to server now if omitted.
+            temporal_model_scores: optional smoke probabilities (0 to 1) from the temporal model run on the
+                device, one per bbox (must align with `bboxes`). When given, the API applies its own
+                threshold to them instead of scoring the sequences itself.
 
         Returns:
             HTTP response
@@ -385,12 +389,19 @@ def create_detection(
             raise ValueError("bboxes must be a list of tuples with a maximum of 5 boxes")
         if crops is not None and len(crops) != len(bboxes):
             raise ValueError("crops must have the same length as bboxes")
+        if temporal_model_scores is not None:
+            if len(temporal_model_scores) != len(bboxes):
+                raise ValueError("temporal_model_scores must have the same length as bboxes")
+            if any(score > 1 or score < 0 for score in temporal_model_scores):
+                raise ValueError("temporal_model_scores are expected to be probabilities")
         data: Dict[str, str] = {
             "bboxes": _dump_bbox_to_json(bboxes),
         }
         data["pose_id"] = str(pose_id)
         if recorded_at is not None:
             data["recorded_at"] = recorded_at
+        if temporal_model_scores is not None:
+            data["temporal_model_scores"] = f"[{','.join(_to_str(score) for score in temporal_model_scores)}]"
         # Use a list of tuples (not a dict) so the repeated "crop" key is preserved per bbox.
         files: List[Tuple[str, Tuple[str, bytes, str]]] = [("file", ("frame.jpg", media, "image/jpeg"))]
         if crops is not None:

client/tests/test_client.py

@@ -63,6 +63,16 @@ def fake_post(url, headers=None, data=None, timeout=None, files=None):
     captured.clear()
     api_client.create_detection(b"img", [(0.1, 0.1, 0.5, 0.8, 0.5)], pose_id=1)
     assert "recorded_at" not in captured["data"]
+    assert "temporal_model_scores" not in captured["data"]
+
+    captured.clear()
+    api_client.create_detection(
+        b"img", [(0.1, 0.1, 0.5, 0.8, 0.5), (0.6, 0.6, 0.7, 0.7, 0.4)], pose_id=1, temporal_model_scores=[0.6234, 0.15]
+    )
+    assert captured["data"]["temporal_model_scores"] == "[0.623,0.150]"
+
+    with pytest.raises(ValueError, match="same length"):
+        api_client.create_detection(b"img", [(0.1, 0.1, 0.5, 0.8, 0.5)], pose_id=1, temporal_model_scores=[0.6, 0.1])
 
 
 def test_get_current_poses_camera(cam_token, cam_pose_id):

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

@@ -37,12 +37,24 @@
 from app.core.config import settings
 from app.core.time import to_utc_naive, utcnow
 from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, PoseCRUD, SequenceCRUD
-from app.models import Alert, AlertSequence, Camera, Detection, Pose, Role, Sequence, UserRole
+from app.models import (
+    TEMPORAL_SCORE_SOURCE_ENGINE,
+    Alert,
+    AlertSequence,
+    Camera,
+    Detection,
+    Pose,
+    Role,
+    Sequence,
+    UserRole,
+)
 from app.schemas.alerts import AlertCreate, AlertUpdate
 from app.schemas.detections import (
     BOX_PATTERN,
     BOXES_PATTERN,
     EMPTY_BBOXES,
+    FLOAT_PATTERN,
+    SCORES_PATTERN,
     DetectionCreate,
     DetectionRead,
     DetectionSequence,
@@ -522,6 +534,16 @@ async def create_detection(
             "converted to UTC; naive values are assumed UTC. Defaults to server now if omitted."
         ),
     ),
+    temporal_model_scores: Optional[str] = Form(
+        None,
+        pattern=SCORES_PATTERN,
+        max_length=settings.MAX_BBOX_STR_LENGTH,
+        description=(
+            "string representation of the list of smoke probabilities (max 3 decimals) from the temporal model "
+            "run on the device, one per bbox in the same order. When given, the API applies its own threshold "
+            "to them instead of scoring the sequences itself."
+        ),
+    ),
     file: UploadFile = File(..., alias="file"),
     crop_files: Optional[List[UploadFile]] = File(None, alias="crop"),
     detections: DetectionCRUD = Depends(get_detection_crud),
@@ -556,6 +578,15 @@ async def create_detection(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail="Number of crops must match the number of bboxes.",
         )
+    # Same alignment rule for the device's temporal scores: one per bbox.
+    scores: Optional[List[float]] = None
+    if temporal_model_scores is not None:
+        scores = [float(value) for value in re.findall(FLOAT_PATTERN, temporal_model_scores)]
+        if len(scores) != len(bbox_strings):
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="Number of temporal model scores must match the number of bboxes.",
+            )
 
     # The engine may report when the image was actually captured; fall back to now when it doesn't.
     # This is the event time the platform displays, and it sets the sequence's started_at; liveness
@@ -609,6 +640,7 @@ async def create_detection(
         )
 
         det_bbox = _parse_bbox(bbox_str)
+        device_score = scores[idx] if scores is not None else None
         # Sequence handling
         # Check if there is a sequence that was seen recently
         candidate_sequences = await sequences.fetch_all(
@@ -639,6 +671,14 @@ 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)
+            if device_score is not None:
+                await sequences.set_temporal_score(
+                    matched_sequence.id, device_score, api_version=TEMPORAL_SCORE_SOURCE_ENGINE
+                )
+            elif matched_sequence.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE:
+                # The device stopped sending scores (its temporal service failed open): drop its
+                # stale verdict so the worker scores the sequence itself again.
+                await sequences.set_temporal_score(matched_sequence.id, None)
             affected_sequences.add(matched_sequence.id)
         else:
             det_filters: List[tuple[str, Any]] = [
@@ -688,6 +728,10 @@ async def create_detection(
                     updated = await detections.update(det_.id, DetectionSequence(sequence_id=sequence_.id))
                     if det_.id == det.id:
                         det = updated
+                if device_score is not None:
+                    await sequences.set_temporal_score(
+                        sequence_.id, device_score, api_version=TEMPORAL_SCORE_SOURCE_ENGINE
+                    )
                 affected_sequences.add(sequence_.id)
 
         created.append(det)
@@ -710,7 +754,8 @@ async def create_detection(
     # 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.
+    # notification channels (webhooks, Telegram, Slack) fire only once validated. A score
+    # computed on the device was stored above, so the worker skips its own model call.
     for seq_id in affected_sequences:
         await sequences.enqueue_validation(seq_id)
 

src/app/crud/crud_sequence.py

@@ -43,15 +43,15 @@ async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
     async def set_temporal_score(
         self,
         sequence_id: int,
-        score: float,
+        score: Union[float, None],
         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).
+        overwrites all three). A None score resets the sequence to never scored.
         """
         stmt: Any = (
             update(Sequence)

src/app/models.py

@@ -121,6 +121,9 @@ class Detection(SQLModel, table=True):
 VALIDATION_FAILED = "failed"
 # Terminal states: the queue must never resurrect these.
 TERMINAL_VALIDATION_STATUSES = (WINDOW_EXHAUSTED, VALIDATION_FAILED)
+# sequences.temporal_api_version value for a score computed on the device and sent with the
+# detection: the worker applies the threshold to it instead of calling the model.
+TEMPORAL_SCORE_SOURCE_ENGINE = "engine"
 
 
 class Sequence(SQLModel, table=True):
@@ -168,7 +171,10 @@ class Sequence(SQLModel, table=True):
         None,
         nullable=True,
         max_length=32,
-        description="Temporal API serving-code version (Docker image tag) that produced temporal_model_score.",
+        description=(
+            "Temporal API serving-code version (Docker image tag) that produced temporal_model_score, "
+            "or 'engine' when the device scored the sequence itself and sent the score with the detection."
+        ),
     )
     is_validated: bool = Field(
         default=False,

src/app/schemas/detections.py

@@ -11,7 +11,15 @@
 from app.core.config import settings
 from app.models import AnnotationType, Detection
 
-__all__ = ["EMPTY_BBOXES", "DetectionCreate", "DetectionLabel", "DetectionRead", "DetectionUrl", "DetectionWithUrl"]
+__all__ = [
+    "EMPTY_BBOXES",
+    "SCORES_PATTERN",
+    "DetectionCreate",
+    "DetectionLabel",
+    "DetectionRead",
+    "DetectionUrl",
+    "DetectionWithUrl",
+]
 
 
 class DetectionLabel(BaseModel):
@@ -23,6 +31,8 @@ class DetectionLabel(BaseModel):
 BOX_PATTERN = rf"\({FLOAT_PATTERN},{FLOAT_PATTERN},{FLOAT_PATTERN},{FLOAT_PATTERN},{FLOAT_PATTERN}\)"
 # An empty list is valid: a frame with no detection (kept for sequence continuity).
 BOXES_PATTERN = rf"^\[({BOX_PATTERN}(,{BOX_PATTERN})*)?\]$"
+# Temporal-model scores computed on the device, one per bbox in the same order.
+SCORES_PATTERN = rf"^\[({FLOAT_PATTERN}(,{FLOAT_PATTERN})*)?\]$"
 
 # Stored bbox of a continuity detection: a frame attached to a sequence with no detection on it.
 EMPTY_BBOXES = "[]"

src/app/services/validation.py

@@ -34,6 +34,7 @@
 from app.models import (
     FAIL_OPEN_STALE,
     FAIL_OPEN_UNAVAILABLE,
+    TEMPORAL_SCORE_SOURCE_ENGINE,
     VALIDATED_BY_MODEL,
     VALIDATION_FAILED,
     WINDOW_EXHAUSTED,
@@ -245,14 +246,25 @@ async def _process_claimed_sequence(sequence_id: int) -> None:
     prediction: Optional[TemporalPrediction] = None
     validation_status: Optional[str] = None
 
+    # Scored on the device: the engine already ran the temporal model and sent its latest
+    # score with the detection, so only the threshold is applied here. No frame window
+    # either: the engine scores a sliding window and sends a new score with every frame,
+    # so a sequence below threshold just waits for the next one instead of exhausting.
+    prior_score = sequence_.temporal_model_score
+    if prior_score is not None and sequence_.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE:
+        if prior_score > settings.TEMPORAL_MODEL_THRESHOLD:
+            await _complete_validated_sequence(sequence_id, VALIDATED_BY_MODEL, claim=True)
+        else:
+            await _finish_job(sequence_id, frame_count=total_frames)
+        return
+
     # Window rules past MAX_FRAMES, decided on the stored score. Declined (score <=
     # threshold after its in-window chances) is terminal: don't call the model again and
     # never resurrect via a later fail-open. Scored ABOVE threshold means a previous run
     # validated but didn't complete (e.g. a failed alert attachment released the claim):
     # validate from the stored score — a high score must never decay into a drop. Never
     # scored (backlog, model freshly configured): score the last MAX_FRAMES anyway —
     # backlog must delay sequences, never silently drop them.
-    prior_score = sequence_.temporal_model_score
     if total_frames > temporal_service.MAX_FRAMES and prior_score is not None:
         if prior_score > settings.TEMPORAL_MODEL_THRESHOLD:
             await _complete_validated_sequence(sequence_id, VALIDATED_BY_MODEL, claim=True)

src/tests/endpoints/test_detections.py

@@ -35,6 +35,7 @@
 from app.core.time import utcnow
 from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, OrganizationCRUD, PoseCRUD, SequenceCRUD
 from app.models import (
+    TEMPORAL_SCORE_SOURCE_ENGINE,
     Alert,
     AlertSequence,
     AnnotationType,
@@ -1164,6 +1165,7 @@ async def test_create_detection_sequence_flow_direct(detection_session: AsyncSes
         bboxes="[(0.2,0.2,0.3,0.3,0.9)]",
         pose_id=pose_id,
         recorded_at=None,
+        temporal_model_scores=None,
         file=upload,
         crop_files=None,
         detections=detections,
@@ -1199,6 +1201,7 @@ async def fake_fetch_all(*args, **kwargs):
         bboxes="[(0.25,0.25,0.35,0.35,0.9)]",
         pose_id=pose_id,
         recorded_at=None,
+        temporal_model_scores=None,
         file=upload_again,
         crop_files=None,
         detections=detections,
@@ -1918,6 +1921,113 @@ async def test_create_detection_sequence_starts_at_earliest_capture_on_backlog_f
     assert seq.started_at == earliest_capture
 
 
+@pytest.mark.asyncio
+async def test_create_detection_stores_engine_temporal_score(
+    async_client: AsyncClient, detection_session: AsyncSession, mock_img: bytes, monkeypatch
+):
+    """A score computed on the device lands on the sequence with 'engine' provenance; the
+    latest upload wins so the worker always judges the device's most recent verdict, and a
+    frame without one (device failed open) resets the sequence so the API scores it itself."""
+    monkeypatch.setattr(settings, "SEQUENCE_MIN_INTERVAL_DETS", 2)
+    auth = pytest.get_token(
+        pytest.camera_table[0]["id"],
+        ["camera"],
+        pytest.camera_table[0]["organization_id"],
+    )
+    pose = Pose(camera_id=pytest.camera_table[0]["id"], azimuth=130.0)
+    detection_session.add(pose)
+    await detection_session.commit()
+    await detection_session.refresh(pose)
+
+    for idx, score in enumerate(("[0.15]", "[0.62]")):
+        response = await async_client.post(
+            "/detections",
+            data={
+                "pose_id": pose.id,
+                "bboxes": f"[(0.1{idx},0.1{idx},0.2{idx},0.2{idx},0.9)]",
+                "temporal_model_scores": score,
+            },
+            files={"file": ("logo.png", mock_img, "image/png")},
+            headers=auth,
+        )
+        assert response.status_code == 201, response.text
+
+    seq = await detection_session.get(Sequence, response.json()["sequence_id"])
+    assert seq is not None
+    assert seq.temporal_model_score == pytest.approx(0.62)
+    assert seq.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE
+    assert seq.temporal_model_version is None
+    assert seq.validation_due_at is not None  # still queued: the worker applies the threshold
+
+    response = await async_client.post(
+        "/detections",
+        data={"pose_id": pose.id, "bboxes": "[(0.12,0.12,0.22,0.22,0.9)]"},
+        files={"file": ("logo.png", mock_img, "image/png")},
+        headers=auth,
+    )
+    assert response.status_code == 201, response.text
+    assert response.json()["sequence_id"] == seq.id
+    await detection_session.refresh(seq)
+    assert seq.temporal_model_score is None
+    assert seq.temporal_api_version is None
+
+
+@pytest.mark.asyncio
+async def test_create_detection_scores_each_bbox_sequence(
+    async_client: AsyncClient, detection_session: AsyncSession, mock_img: bytes, monkeypatch
+):
+    """Two objects on one frame: each sequence gets the score of its own bbox."""
+    monkeypatch.setattr(settings, "SEQUENCE_MIN_INTERVAL_DETS", 1)
+    auth = pytest.get_token(
+        pytest.camera_table[0]["id"],
+        ["camera"],
+        pytest.camera_table[0]["organization_id"],
+    )
+    pose = Pose(camera_id=pytest.camera_table[0]["id"], azimuth=130.0)
+    detection_session.add(pose)
+    await detection_session.commit()
+    await detection_session.refresh(pose)
+
+    response = await async_client.post(
+        "/detections",
+        data={
+            "pose_id": pose.id,
+            "bboxes": "[(0.1,0.1,0.2,0.2,0.9),(0.6,0.6,0.7,0.7,0.8)]",
+            "temporal_model_scores": "[0.62,0.15]",
+        },
+        files={"file": ("logo.png", mock_img, "image/png")},
+        headers=auth,
+    )
+    assert response.status_code == 201, response.text
+
+    res = await detection_session.exec(select(Sequence).where(Sequence.pose_id == pose.id).order_by(Sequence.id))
+    assert sorted(seq.temporal_model_score for seq in res.all()) == pytest.approx([0.15, 0.62])
+
+
+@pytest.mark.parametrize("scores", ["[-0.1]", "[1.5]", "[nan]", "0.5", "[0.5,0.6]", "[]"])
+@pytest.mark.asyncio
+async def test_create_detection_rejects_invalid_temporal_scores(
+    async_client: AsyncClient, mock_img: bytes, scores: str
+):
+    """Out-of-range or malformed scores fail the pattern; a count mismatch with bboxes is a 422 too."""
+    auth = pytest.get_token(
+        pytest.camera_table[0]["id"],
+        ["camera"],
+        pytest.camera_table[0]["organization_id"],
+    )
+    response = await async_client.post(
+        "/detections",
+        data={
+            "pose_id": pytest.pose_table[0]["id"],
+            "bboxes": "[(0.1,0.1,0.2,0.2,0.9)]",
+            "temporal_model_scores": scores,
+        },
+        files={"file": ("logo.png", mock_img, "image/png")},
+        headers=auth,
+    )
+    assert response.status_code == 422, response.text
+
+
 @pytest.mark.asyncio
 async def test_attach_sequence_merges_same_event_alerts(detection_session: AsyncSession):
     """