diff --git a/client/pyroclient/client.py b/client/pyroclient/client.py index c5db9be6..569492fb 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_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: diff --git a/client/tests/test_client.py b/client/tests/test_client.py index 75068f34..83854cbb 100644 --- a/client/tests/test_client.py +++ b/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): diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index 9d5c2d40..da749ce7 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/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) 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/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/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/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_detections.py b/src/tests/endpoints/test_detections.py index dc413c30..2bafdeb1 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_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): """ 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