feat(sequences): add detection sampling, plus S3 bucket and presigned-URL caching

client/pyroclient/client.py

@@ -496,9 +496,11 @@ def fetch_latest_sequences(self) -> Response:
     def fetch_sequences_detections(
         self,
         sequence_id: int,
-        limit: int = 10,
+        limit: Union[int, None] = None,
         desc: bool = True,
         with_crop: bool = True,
+        sampling: int = 1,
+        offset: int = 0,
     ) -> Response:
         """List the detections of a sequence
 
@@ -508,17 +510,32 @@ def fetch_sequences_detections(
 
         Args:
             sequence_id: ID of the associated sequence entry
-            limit: maximum number of detections to fetch
+            limit: maximum number of detections to fetch. Unset (the default) lets the API pick:
+                10, or the size of the whole sampled span capped at 500 when sampling is set
             desc: whether to order the detections by created_at in descending order
             with_crop: whether to include the crop_url for detections that have a crop
+            sampling: keep one detection every N (1 = all, max 10000). The kept frames do not
+                depend on desc. Leave limit unset to span the sequence, and read the
+                X-Sampled-Total / X-Sampled-Truncated response headers
+            offset: raw detections to skip, from the oldest end when sampling. Page by advancing
+                it in multiples of sampling
 
         Returns:
             HTTP response
         """
+        params: Dict[str, Any] = {
+            "desc": desc,
+            "with_crop": with_crop,
+            "sampling": sampling,
+            "offset": offset,
+        }
+        # Omitted rather than defaulted client-side, so the API can size it from the sampled set.
+        if limit is not None:
+            params["limit"] = limit
         return requests.get(
             urljoin(self._route_prefix, ClientRoute.SEQUENCES_FETCH_DETECTIONS.format(seq_id=sequence_id)),
             headers=self.headers,
-            params={"limit": limit, "desc": desc, "with_crop": with_crop},
+            params=params,
             timeout=self.timeout,
         )
 

client/tests/test_client.py

@@ -191,9 +191,20 @@ def test_user_workflow(test_cam_workflow, user_token):
     assert len(response.json()) == 0  # Sequence was labeled by agent
     response = user_client.fetch_sequences_from_date(datetime.utcnow().date().isoformat())
     assert len(response.json()) == 1
-    response = user_client.fetch_sequences_detections(response.json()[0]["id"])
+    sequence_id = response.json()[0]["id"]
+    response = user_client.fetch_sequences_detections(sequence_id)
     assert response.status_code == 200, response.__dict__
     # 4 real detections + the continuity row added by the empty frame in test_cam_workflow
     detections = response.json()
     assert len(detections) == 5
     assert sum(det["bbox"] == "[]" for det in detections) == 1
+    # An explicit limit is forwarded, an omitted one is left to the API.
+    response = user_client.fetch_sequences_detections(sequence_id, limit=2)
+    assert response.status_code == 200, response.__dict__
+    assert len(response.json()) == 2
+    # With sampling and no limit, the API sizes the response to span the sampled set.
+    response = user_client.fetch_sequences_detections(sequence_id, sampling=2)
+    assert response.status_code == 200, response.__dict__
+    assert len(response.json()) == 3  # ceil(5 / 2)
+    assert response.headers["x-sampled-total"] == "3"
+    assert response.headers["x-sampled-truncated"] == "false"

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

@@ -4,10 +4,11 @@
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
 
+import math
 from datetime import date, timedelta
 from typing import Any, List, Union, cast
 
-from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, Response, Security, status
 from sqlmodel import delete, func, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -17,7 +18,7 @@
 from app.db import get_session
 from app.models import AlertSequence, AnnotationType, Camera, Detection, Sequence, UserRole
 from app.schemas.alerts import AlertCreate
-from app.schemas.detections import DetectionRead, DetectionSequence, DetectionWithUrl
+from app.schemas.detections import DetectionSequence, DetectionWithUrl
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.alerts import refresh_alert_state
@@ -29,6 +30,10 @@
 
 router = APIRouter()
 
+DEFAULT_DETECTION_LIMIT = 10  # historical default, kept for sampling=1
+MAX_DETECTION_LIMIT = 500
+MAX_SAMPLING = 10_000
+
 
 async def verify_org_rights(
     organization_id: int, camera_id: int, cameras: CameraCRUD = Depends(get_camera_crud)
@@ -71,17 +76,50 @@ async def get_sequence(
     ),
 )
 async def fetch_sequence_detections(
+    response: Response,
     sequence_id: int = Path(..., gt=0),
-    limit: int = Query(10, description="Maximum number of detections to fetch", ge=1, le=100),
-    offset: int = Query(0, description="Number of detections to skip", ge=0),
+    limit: Union[int, None] = Query(
+        None,
+        description=(
+            f"Maximum number of detections to fetch. Defaults to {DEFAULT_DETECTION_LIMIT}, except "
+            "when `sampling` is set and `limit` is omitted: it then spans the whole sampled set "
+            f"from `offset` onward, capped at {MAX_DETECTION_LIMIT}."
+        ),
+        ge=1,
+        le=MAX_DETECTION_LIMIT,
+    ),
+    offset: int = Query(
+        0,
+        description=(
+            "Number of detections to skip, counted in raw detections whatever `sampling` is. When "
+            "sampling, it counts from the oldest end regardless of `desc`, so "
+            "`offset=20&sampling=48` starts at detection 21. Page by advancing `offset` in "
+            "multiples of `sampling` to keep the grid on the same detections."
+        ),
+        ge=0,
+    ),
     desc: bool = Query(True, description="Whether to order the detections by created_at in descending order"),
+    sampling: int = Query(
+        1,
+        description=(
+            "Keep one detection every N (1 = every detection). Frames are picked chronologically, "
+            "so the set does not depend on `desc`. An explicit `limit` below "
+            "`ceil((detections_count - offset) / sampling)` returns only part of the span (its "
+            "most recent part when `desc=true`); omit `limit` to get the whole span, and read the "
+            "`X-Sampled-Total` and `X-Sampled-Truncated` response headers to tell whether it "
+            "covered the rest of the sequence."
+        ),
+        ge=1,
+        le=MAX_SAMPLING,
+    ),
     with_crop: bool = Query(
         False,
         description="If true, presign and include crop_url for detections that have a crop. Defaults to false to skip the extra S3 head requests when crops are not needed.",
     ),
     cameras: CameraCRUD = Depends(get_camera_crud),
     detections: DetectionCRUD = Depends(get_detection_crud),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
+    session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[DetectionWithUrl]:
     telemetry_client.capture(token_payload.sub, event="sequences-get", properties={"sequence_id": sequence_id})
@@ -90,18 +128,25 @@ async def fetch_sequence_detections(
     if not token_payload.is_admin and token_payload.organization_id != camera.organization_id:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
 
+    if sampling > 1:
+        # limit still truncates the thinned set, invisibly: you get what looks like a spread but is
+        # only its tail. Size the set so an omitted limit spans it, and report it either way.
+        counts = await get_detection_counts_by_sequence_ids(session, [sequence_id])
+        sampled_total = math.ceil(max(0, counts.get(sequence_id, 0) - offset) / sampling)
+        effective_limit = limit if limit is not None else min(MAX_DETECTION_LIMIT, sampled_total)
+        response.headers["X-Sampled-Total"] = str(sampled_total)
+        response.headers["X-Sampled-Truncated"] = str(effective_limit < sampled_total).lower()
+    else:
+        effective_limit = limit if limit is not None else DEFAULT_DETECTION_LIMIT
+
     # Get the bucket of the camera's organization
     bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(camera.organization_id))
-    fetched = await detections.fetch_all(
-        filters=("sequence_id", sequence_id),
-        order_by="created_at",
-        order_desc=desc,
-        limit=limit,
-        offset=offset,
+    fetched = await detections.fetch_by_sequence(
+        sequence_id, sampling=sampling, order_desc=desc, limit=effective_limit, offset=offset
     )
     return [
         DetectionWithUrl(
-            **DetectionRead(**elt.model_dump()).model_dump(),
+            **elt.model_dump(),
             url=bucket.get_public_url(elt.bucket_key, verify_exists=False),
             crop_url=(
                 bucket.get_public_url(elt.crop_bucket_key, verify_exists=False)

src/app/crud/crud_detection.py

@@ -3,9 +3,11 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
-from typing import Any, Union, cast
+from typing import Any, List, Union, cast
 
-from sqlalchemy import desc
+from sqlalchemy import desc, func
+from sqlalchemy import select as select_sa
+from sqlalchemy.orm import aliased
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -31,3 +33,56 @@ async def get_latest_with_bbox(self, sequence_id: int) -> Union[Detection, None]
         )
         results = await self.session.exec(statement)
         return results.first()
+
+    async def fetch_by_sequence(
+        self,
+        sequence_id: int,
+        sampling: int = 1,
+        order_desc: bool = True,
+        limit: int = 10,
+        offset: int = 0,
+    ) -> List[Detection]:
+        """Fetch the detections of a sequence, keeping one every ``sampling``.
+
+        ``offset`` counts raw detections, never sampled frames. When sampling, the row number is
+        computed ascending on ``created_at``, so neither the offset nor the kept set depends on
+        ``order_desc``: it only flips the output order. Page by advancing ``offset`` in multiples
+        of ``sampling`` to keep the grid on the same detections. Unsampled calls delegate to
+        ``fetch_all``, where a SQL ``OFFSET`` applies after the sort and so counts from whichever
+        end ``order_desc`` selects.
+        """
+        if sampling <= 1:
+            return await self.fetch_all(
+                filters=("sequence_id", sequence_id),
+                order_by="created_at",
+                order_desc=order_desc,
+                limit=limit,
+                offset=offset,
+            )
+
+        # id breaks created_at ties so the sampled set is deterministic run to run.
+        row_num = func.row_number().over(
+            order_by=(cast(Any, Detection.created_at).asc(), cast(Any, Detection.id).asc())
+        )
+        # sqlmodel's select on the outer query: a single-entity SelectOfScalar is what makes
+        # exec return Detection instances rather than Row tuples.
+        numbered: Any = select_sa(Detection, row_num.label("rn")).where(cast(Any, Detection.sequence_id) == sequence_id)
+        subq = numbered.subquery()
+        sampled = aliased(Detection, subq)
+        created_at_col = cast(Any, sampled.created_at)
+        id_col = cast(Any, sampled.id)
+        # offset in the WHERE, not a SQL OFFSET: it counts raw detections on the ascending
+        # numbering, and a SQL OFFSET would instead apply after ORDER BY.
+        position = subq.c.rn - 1
+        stmt: Any = (
+            select(sampled)
+            .where(position >= offset)
+            .where((position - offset) % sampling == 0)
+            .order_by(
+                created_at_col.desc() if order_desc else created_at_col.asc(),
+                id_col.desc() if order_desc else id_col.asc(),
+            )
+            .limit(limit)
+        )
+        result = await self.session.exec(stmt)
+        return list(result.all())

src/app/main.py

@@ -126,6 +126,9 @@ async def add_process_time_header(request: Request, call_next):
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
+    # allow_headers governs request headers; response headers stay hidden from browser JS unless
+    # they are listed here, and the sampling signals exist for the frontend player.
+    expose_headers=["X-Sampled-Total", "X-Sampled-Truncated"],
 )
 
 if isinstance(settings.SENTRY_DSN, str):