Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions src/routers/analytics.py
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,25 @@ def _ensure_detection_visible(db: Session, detection: OccupancyObservation, user
)


def _available_artifact_url(
metadata: dict[str, Any],
variant: str,
) -> str | None:
snapshots = metadata.get("snapshots")
if not isinstance(snapshots, dict):
return None

artifact = snapshots.get(variant)
if not isinstance(artifact, dict):
return None

url = artifact.get("url")
if not isinstance(url, str) or not url.strip():
return None

return url


# ---------------------------------------------------------------------------
# Time and aggregation helpers
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -800,6 +819,9 @@ def _serialize_detection_run(
error_code=_to_str_or_none(_metadata_value(metadata, "error_code")),
error_message=_to_str_or_none(_metadata_value(metadata, "error_message", "error")),
has_feedback=observation.observation_id in feedback_ids,
raw_snapshot_url=_available_artifact_url(metadata, "raw"),
annotated_snapshot_url=_available_artifact_url(metadata, "annotated"),
yolo_labels_url=_available_artifact_url(metadata, "labels"),
)


Expand Down
7 changes: 1 addition & 6 deletions src/routers/cameras.py
Original file line number Diff line number Diff line change
Expand Up @@ -533,12 +533,7 @@ def _stored_snapshot_response(
# GET /cameras/{camera_id}/snapshot
# ---------------------------------------------------------------------------

@router.get(
"/{camera_id}/snapshot",
response_class=Response,
responses=SNAPSHOT_IMAGE_RESPONSES,
)
def get_camera_snapshot(
def _legacy_get_snapshot(
camera_id: int,
current_user: Annotated[User, require("cameras.view")],
db: Annotated[Session, Depends(get_db)],
Expand Down
3 changes: 3 additions & 0 deletions src/schemas/analytics.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,9 @@ class DetectionRun(BaseModel):
error_code: str | None
error_message: str | None
has_feedback: bool
raw_snapshot_url: str | None
annotated_snapshot_url: str | None
yolo_labels_url: str | None


class DetectionRunListResponse(BaseModel):
Expand Down
53 changes: 51 additions & 2 deletions tests/test_snapshot_routes.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,17 @@
from __future__ import annotations

from datetime import datetime, timezone
import os
import unittest

os.environ.setdefault("DATABASE_URL", "sqlite://")

from src.db_models import OccupancyObservation # noqa: E402
from src.routers import analytics, cameras # noqa: E402


class SnapshotRouteContractTests(unittest.TestCase):
def test_snapshot_is_exposed_by_cameras_only(self):
def test_separate_detection_artifact_endpoints_remain_removed(self):
analytics_paths = {route.path for route in analytics.router.routes}
camera_paths = {route.path for route in cameras.router.routes}

Expand All @@ -21,7 +23,54 @@ def test_snapshot_is_exposed_by_cameras_only(self):
"/admin/analytics/detections/{detection_run_id}/labels",
analytics_paths,
)
self.assertIn("/cameras/{camera_id}/snapshot", camera_paths)
self.assertNotIn("/cameras/{camera_id}/snapshot", camera_paths)

def test_detection_response_keeps_public_s3_artifact_urls(self):
raw_url = "https://s3.example.test/detections/42/raw.jpg"
annotated_url = "https://s3.example.test/detections/42/annotated.jpg"
labels_url = "https://s3.example.test/detections/42/labels.txt"
observed_at = datetime(2026, 7, 23, 12, 0, tzinfo=timezone.utc)
detection = OccupancyObservation(
observation_id=42,
zone_id=9,
camera_id=17,
source_type="detector",
capacity=8,
occupied=3,
confidence=0.9,
observed_at=observed_at,
ingested_at=observed_at,
metadata_json={
"snapshots": {
"raw": {"url": raw_url},
"annotated": {"url": annotated_url},
"labels": {"url": labels_url},
}
},
)

result = analytics._serialize_detection_run(
db=None, # type: ignore[arg-type]
observation=detection,
feedback_ids=set(),
)

self.assertEqual(result.raw_snapshot_url, raw_url)
self.assertEqual(result.annotated_snapshot_url, annotated_url)
self.assertEqual(result.yolo_labels_url, labels_url)

def test_invalid_or_missing_s3_artifact_urls_are_null(self):
metadata = {
"snapshots": {
"raw": {"url": " "},
"annotated": {"object_key": "annotated.jpg"},
"labels": "https://s3.example.test/labels.txt",
}
}

self.assertIsNone(analytics._available_artifact_url(metadata, "raw"))
self.assertIsNone(analytics._available_artifact_url(metadata, "annotated"))
self.assertIsNone(analytics._available_artifact_url(metadata, "labels"))


if __name__ == "__main__":
Expand Down
Loading