From c8fe393efc95d298bddd87f36dc7e5908402028d Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 22 Oct 2026 15:41:38 +0200 Subject: [PATCH 1/5] feat(detections): accept the temporal score computed on the device The engine may run the temporal smoke model itself and send its score with the detection. Store it on the touched sequences with 'engine' as provenance so the validation worker knows the sequence was already scored. --- src/app/api/api_v1/endpoints/detections.py | 25 ++++++++- src/app/models.py | 8 ++- src/tests/endpoints/test_detections.py | 64 ++++++++++++++++++++++ 3 files changed, 95 insertions(+), 2 deletions(-) diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index 9d5c2d40..6870d6a2 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/src/app/api/api_v1/endpoints/detections.py @@ -37,7 +37,17 @@ 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, @@ -522,6 +532,15 @@ async def create_detection( "converted to UTC; naive values are assumed UTC. Defaults to server now if omitted." ), ), + temporal_model_score: Optional[float] = Form( + None, + ge=0, + le=1, + description=( + "Smoke probability from the temporal model run on the device for the event this frame belongs to. " + "When given, the API applies its own threshold to it instead of scoring the sequence itself." + ), + ), file: UploadFile = File(..., alias="file"), crop_files: Optional[List[UploadFile]] = File(None, alias="crop"), detections: DetectionCRUD = Depends(get_detection_crud), @@ -711,7 +730,11 @@ async def create_detection( # 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. + # A score computed on the device is stored first, so the worker finds it when it claims + # the job and skips its own model call. for seq_id in affected_sequences: + if temporal_model_score is not None: + await sequences.set_temporal_score(seq_id, temporal_model_score, api_version=TEMPORAL_SCORE_SOURCE_ENGINE) await sequences.enqueue_validation(seq_id) first_det = cast(Detection, await detections.get(created[0].id, strict=True)) diff --git a/src/app/models.py b/src/app/models.py index 599531a8..8622d968 100644 --- a/src/app/models.py +++ b/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, diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py index dc413c30..a65ef65a 100644 --- a/src/tests/endpoints/test_detections.py +++ b/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_score=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_score=None, file=upload_again, crop_files=None, detections=detections, @@ -1918,6 +1921,67 @@ 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.""" + 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_score": 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 + + +@pytest.mark.parametrize("score", ["-0.1", "1.5", "nan"]) +@pytest.mark.asyncio +async def test_create_detection_rejects_out_of_range_temporal_score( + async_client: AsyncClient, mock_img: bytes, score: str +): + 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_score": score, + }, + 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): """ From 517839a71f0466b6b3bfa2090ed9d8728f04a6b1 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 22 Oct 2026 15:41:38 +0200 Subject: [PATCH 2/5] feat(validation): skip the model for sequences scored on the device When the stored score came from the engine, apply the threshold to it instead of calling the temporal API. No frame window: the engine scores a sliding window and sends a fresh score with every frame, so a sequence below threshold waits for the next one. The risk gate still applies. --- src/app/services/validation.py | 14 +++- .../endpoints/test_temporal_validation.py | 77 ++++++++++++++++++- 2 files changed, 89 insertions(+), 2 deletions(-) diff --git a/src/app/services/validation.py b/src/app/services/validation.py index 50ad22ca..c9205239 100644 --- a/src/app/services/validation.py +++ b/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,6 +246,18 @@ 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 @@ -252,7 +265,6 @@ async def _process_claimed_sequence(sequence_id: int) -> None: # 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) diff --git a/src/tests/endpoints/test_temporal_validation.py b/src/tests/endpoints/test_temporal_validation.py index 0cd99579..79b9ada8 100644 --- a/src/tests/endpoints/test_temporal_validation.py +++ b/src/tests/endpoints/test_temporal_validation.py @@ -16,7 +16,7 @@ from app.core.time import utcnow from app.crud import DetectionCRUD, SequenceCRUD from app.db import session_factory -from app.models import AlertSequence, Detection, Sequence +from app.models import TEMPORAL_SCORE_SOURCE_ENGINE, AlertSequence, Detection, Sequence from app.services import validation as validation_service from app.services.risk import risk_service from app.services.temporal import TemporalPrediction, temporal_service @@ -477,6 +477,81 @@ async def test_process_validates_from_prior_high_score_past_window(detection_ses assert await _has_alert_link(detection_session, cast(int, seq.id)) is True +@pytest.mark.asyncio +async def test_process_engine_score_above_threshold_validates_without_model( + detection_session: AsyncSession, monkeypatch +): + """A score sent by the device is judged on the threshold alone: no model call, no + MIN_FRAMES rule (the engine scored its own window), reachable service or not.""" + seq = await _seed_sequence(detection_session, 2, max_conf=0.30) # below MIN_FRAMES on purpose + crud = SequenceCRUD(detection_session) + await crud.set_temporal_score(cast(int, seq.id), 0.60, api_version=TEMPORAL_SCORE_SOURCE_ENGINE) + await _enqueue(detection_session, cast(int, seq.id)) + predict = AsyncMock(return_value=_prediction(0.9)) + monkeypatch.setattr(risk_service, "_scores", {}) + monkeypatch.setattr(temporal_service, "is_available", lambda: True) + monkeypatch.setattr(temporal_service, "predict", predict) + + assert await _process_one_due() is True + + predict.assert_not_awaited() + await detection_session.refresh(seq) + assert seq.is_validated is True + assert seq.validation_status == VALIDATED_BY_MODEL + assert seq.temporal_model_score == pytest.approx(0.60) # the device's score is kept as is + assert seq.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE + assert await _has_alert_link(detection_session, cast(int, seq.id)) is True + + +@pytest.mark.asyncio +async def test_process_engine_score_below_threshold_waits_for_next_frame(detection_session: AsyncSession, monkeypatch): + """Below threshold the sequence stays in the DB unvalidated and is never window-exhausted: + the device keeps scoring a sliding window, so the next frame brings a fresh verdict.""" + seq = await _seed_sequence(detection_session, 12, max_conf=0.30) # past MAX_FRAMES on purpose + crud = SequenceCRUD(detection_session) + await crud.set_temporal_score(cast(int, seq.id), 0.20, api_version=TEMPORAL_SCORE_SOURCE_ENGINE) + await _enqueue(detection_session, cast(int, seq.id)) + predict = AsyncMock(return_value=_prediction(0.9)) + monkeypatch.setattr(risk_service, "_scores", {}) + monkeypatch.setattr(temporal_service, "is_available", lambda: True) + monkeypatch.setattr(temporal_service, "predict", predict) + + assert await _process_one_due() is True + + predict.assert_not_awaited() + await detection_session.refresh(seq) + assert seq.is_validated is False + assert seq.validation_status is None # not window_exhausted: still eligible to re-enqueue + assert seq.validation_due_at is None # retried when the next detection re-enqueues + assert await _has_alert_link(detection_session, cast(int, seq.id)) is False + + # The next frame carries a higher device score: validated on re-enqueue, still no model call. + await crud.set_temporal_score(cast(int, seq.id), 0.70, api_version=TEMPORAL_SCORE_SOURCE_ENGINE) + await _enqueue(detection_session, cast(int, seq.id)) + assert await _process_one_due() is True + predict.assert_not_awaited() + await detection_session.refresh(seq) + assert seq.is_validated is True + assert seq.validation_status == VALIDATED_BY_MODEL + + +@pytest.mark.asyncio +async def test_process_engine_score_still_behind_risk_gate(detection_session: AsyncSession, monkeypatch): + seq = await _seed_sequence(detection_session, 5, max_conf=0.30) + await SequenceCRUD(detection_session).set_temporal_score( + cast(int, seq.id), 0.90, api_version=TEMPORAL_SCORE_SOURCE_ENGINE + ) + await _enqueue(detection_session, cast(int, seq.id)) + monkeypatch.setattr(risk_service, "_scores", {1: "very_low"}) # 0.6 threshold > 0.30 + monkeypatch.setattr(temporal_service, "is_available", lambda: True) + + assert await _process_one_due() is True + + await detection_session.refresh(seq) + assert seq.is_validated is False + assert seq.validation_due_at is None + + @pytest.mark.asyncio async def test_attach_failure_past_window_does_not_window_exhaust(detection_session: AsyncSession, monkeypatch): """End-to-end regression: high score persisted, attach fails, retry past MAX_FRAMES From fb14ad4f7d2cf43fdf87636f7efa455f6c97d9e8 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 22 Oct 2026 15:41:38 +0200 Subject: [PATCH 3/5] feat(client): forward temporal_model_score on create_detection --- client/pyroclient/client.py | 6 ++++++ client/tests/test_client.py | 5 +++++ 2 files changed, 11 insertions(+) diff --git a/client/pyroclient/client.py b/client/pyroclient/client.py index c5db9be6..b004ed8e 100644 --- a/client/pyroclient/client.py +++ b/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_score: 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_score: optional smoke probability (0 to 1) from the temporal model run on the + device for the event this frame belongs to. When given, the API applies its own threshold + to it instead of scoring the sequence itself. Returns: HTTP response @@ -391,6 +395,8 @@ def create_detection( data["pose_id"] = str(pose_id) if recorded_at is not None: data["recorded_at"] = recorded_at + if temporal_model_score is not None: + data["temporal_model_score"] = str(temporal_model_score) # 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: diff --git a/client/tests/test_client.py b/client/tests/test_client.py index 75068f34..f66e0167 100644 --- a/client/tests/test_client.py +++ b/client/tests/test_client.py @@ -63,6 +63,11 @@ 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_score" not in captured["data"] + + captured.clear() + api_client.create_detection(b"img", [(0.1, 0.1, 0.5, 0.8, 0.5)], pose_id=1, temporal_model_score=0.62) + assert captured["data"]["temporal_model_score"] == "0.62" def test_get_current_poses_camera(cam_token, cam_pose_id): From 4db70566218ab3fad2b4e153be5d11492865188c Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 22 Oct 2026 15:49:45 +0200 Subject: [PATCH 4/5] fix(detections): reset the device score when a frame arrives without one A scored sequence that receives an unscored frame means the device's temporal service failed open. Drop the stale device verdict so the worker falls back to scoring the sequence itself. --- src/app/api/api_v1/endpoints/detections.py | 4 ++++ src/app/crud/crud_sequence.py | 4 ++-- src/tests/endpoints/test_detections.py | 15 ++++++++++++++- 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index 6870d6a2..ff172f53 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/src/app/api/api_v1/endpoints/detections.py @@ -658,6 +658,10 @@ 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) + # The device stopped sending scores (its temporal service failed open): drop its + # stale verdict so the worker scores the sequence itself again. + if temporal_model_score is None and matched_sequence.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE: + await sequences.set_temporal_score(matched_sequence.id, None) affected_sequences.add(matched_sequence.id) else: det_filters: List[tuple[str, Any]] = [ diff --git a/src/app/crud/crud_sequence.py b/src/app/crud/crud_sequence.py index 8e8a70b6..d13de575 100644 --- a/src/app/crud/crud_sequence.py +++ b/src/app/crud/crud_sequence.py @@ -43,7 +43,7 @@ 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: @@ -51,7 +51,7 @@ async def set_temporal_score( 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) diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py index a65ef65a..466b7963 100644 --- a/src/tests/endpoints/test_detections.py +++ b/src/tests/endpoints/test_detections.py @@ -1926,7 +1926,8 @@ 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.""" + 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"], @@ -1958,6 +1959,18 @@ async def test_create_detection_stores_engine_temporal_score( 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.parametrize("score", ["-0.1", "1.5", "nan"]) @pytest.mark.asyncio From 658cc6a2a28d9b52fd33167b7affc6ddae4e45b1 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 22 Oct 2026 19:22:08 +0200 Subject: [PATCH 5/5] feat(detections): take one temporal score per bbox The device scores each tracked object separately, so the detection route now takes temporal_model_scores, a list aligned with bboxes like the crops, and each matched or created sequence receives the score of its own bbox instead of one score shared by every sequence of the request. --- client/pyroclient/client.py | 17 +++++--- client/tests/test_client.py | 11 +++-- src/app/api/api_v1/endpoints/detections.py | 44 +++++++++++++------ src/app/schemas/detections.py | 12 +++++- src/tests/endpoints/test_detections.py | 49 ++++++++++++++++++---- 5 files changed, 102 insertions(+), 31 deletions(-) diff --git a/client/pyroclient/client.py b/client/pyroclient/client.py index b004ed8e..569492fb 100644 --- a/client/pyroclient/client.py +++ b/client/pyroclient/client.py @@ -358,7 +358,7 @@ def create_detection( pose_id: int, crops: List[bytes] | None = None, recorded_at: str | None = None, - temporal_model_score: float | None = None, + temporal_model_scores: List[float] | None = None, ) -> Response: """Notify the detection of a wildfire on the picture taken by a camera. @@ -378,9 +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_score: optional smoke probability (0 to 1) from the temporal model run on the - device for the event this frame belongs to. When given, the API applies its own threshold - to it instead of scoring the sequence itself. + 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 @@ -389,14 +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_score is not None: - data["temporal_model_score"] = str(temporal_model_score) + 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: diff --git a/client/tests/test_client.py b/client/tests/test_client.py index f66e0167..83854cbb 100644 --- a/client/tests/test_client.py +++ b/client/tests/test_client.py @@ -63,11 +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_score" 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)], pose_id=1, temporal_model_score=0.62) - assert captured["data"]["temporal_model_score"] == "0.62" + 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): diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index ff172f53..da749ce7 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/src/app/api/api_v1/endpoints/detections.py @@ -53,6 +53,8 @@ BOX_PATTERN, BOXES_PATTERN, EMPTY_BBOXES, + FLOAT_PATTERN, + SCORES_PATTERN, DetectionCreate, DetectionRead, DetectionSequence, @@ -532,13 +534,14 @@ async def create_detection( "converted to UTC; naive values are assumed UTC. Defaults to server now if omitted." ), ), - temporal_model_score: Optional[float] = Form( + temporal_model_scores: Optional[str] = Form( None, - ge=0, - le=1, + pattern=SCORES_PATTERN, + max_length=settings.MAX_BBOX_STR_LENGTH, description=( - "Smoke probability from the temporal model run on the device for the event this frame belongs to. " - "When given, the API applies its own threshold to it instead of scoring the sequence itself." + "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"), @@ -575,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 @@ -628,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( @@ -658,9 +671,13 @@ 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) - # The device stopped sending scores (its temporal service failed open): drop its - # stale verdict so the worker scores the sequence itself again. - if temporal_model_score is None and matched_sequence.temporal_api_version == TEMPORAL_SCORE_SOURCE_ENGINE: + 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: @@ -711,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) @@ -733,12 +754,9 @@ 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. - # A score computed on the device is stored first, so the worker finds it when it claims - # the job and skips its own model call. + # 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: - if temporal_model_score is not None: - await sequences.set_temporal_score(seq_id, temporal_model_score, api_version=TEMPORAL_SCORE_SOURCE_ENGINE) await sequences.enqueue_validation(seq_id) first_det = cast(Detection, await detections.get(created[0].id, strict=True)) diff --git a/src/app/schemas/detections.py b/src/app/schemas/detections.py index 21fb5b93..c6895b11 100644 --- a/src/app/schemas/detections.py +++ b/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 = "[]" diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py index 466b7963..2bafdeb1 100644 --- a/src/tests/endpoints/test_detections.py +++ b/src/tests/endpoints/test_detections.py @@ -1165,7 +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_score=None, + temporal_model_scores=None, file=upload, crop_files=None, detections=detections, @@ -1201,7 +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_score=None, + temporal_model_scores=None, file=upload_again, crop_files=None, detections=detections, @@ -1939,13 +1939,13 @@ async def test_create_detection_stores_engine_temporal_score( await detection_session.commit() await detection_session.refresh(pose) - for idx, score in enumerate(("0.15", "0.62")): + 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_score": score, + "temporal_model_scores": score, }, files={"file": ("logo.png", mock_img, "image/png")}, headers=auth, @@ -1972,11 +1972,44 @@ async def test_create_detection_stores_engine_temporal_score( assert seq.temporal_api_version is None -@pytest.mark.parametrize("score", ["-0.1", "1.5", "nan"]) @pytest.mark.asyncio -async def test_create_detection_rejects_out_of_range_temporal_score( - async_client: AsyncClient, mock_img: bytes, score: str +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"], @@ -1987,7 +2020,7 @@ async def test_create_detection_rejects_out_of_range_temporal_score( data={ "pose_id": pytest.pose_table[0]["id"], "bboxes": "[(0.1,0.1,0.2,0.2,0.9)]", - "temporal_model_score": score, + "temporal_model_scores": scores, }, files={"file": ("logo.png", mock_img, "image/png")}, headers=auth,