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,33 @@ 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. Left unset (the default) the API picks
+                10, or with sampling the size of the whole sampled span capped at 500
             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 are picked
+                chronologically and do not depend on desc. Pair it with limit unset to span the
+                sequence, since an explicit limit below ceil(detections_count / sampling) returns
+                only part of the span. The X-Sampled-Total and X-Sampled-Truncated response
+                headers say whether what came back covers the sequence
+            offset: number of detections to skip, within the sampled set
 
         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

@@ -7,7 +7,7 @@
 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 +17,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 +29,11 @@
 
 router = APIRouter()
 
+# Historical default, kept for sampling=1 so unsampled callers see no change.
+DEFAULT_DETECTION_LIMIT = 10
+# Ceiling on one response, mirrored in the limit Query constraint.
+MAX_DETECTION_LIMIT = 500
+
 
 async def verify_org_rights(
     organization_id: int, camera_id: int, cameras: CameraCRUD = Depends(get_camera_crud)
@@ -71,17 +76,46 @@ 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=(
+            "Maximum number of detections to fetch. Defaults to 10, except when `sampling` is set "
+            "and `limit` is omitted: it then defaults to whatever spans the whole sampled set "
+            "(capped at 500), so `?sampling=10` returns a spread instead of a truncated tail."
+        ),
+        ge=1,
+        le=500,
+    ),
+    offset: int = Query(0, description="Number of detections to skip, within the sampled set", 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). The kept frames are picked "
+            "chronologically from the start of the sequence, so the set does not depend on `desc` "
+            "and does not shift as the sequence grows; the first detection is always kept. "
+            "`limit` and `offset` then page that sampled set. Note that with `desc=true` a newly "
+            "recorded detection shifts page boundaries, since it lands at the front. "
+            "`sampling` thins the set and `limit` caps how much of it comes back, so an explicit "
+            "`limit` below `ceil(detections_count / sampling)` returns only part of the span (the "
+            "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 what you "
+            "got covers the sequence. Since `limit` caps at 500, spanning a sequence in one call "
+            "needs `sampling >= detections_count / 500`."
+        ),
+        ge=1,
+        le=10_000,
+    ),
     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 +124,28 @@ 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.")
 
+    effective_limit = DEFAULT_DETECTION_LIMIT if limit is None else limit
+    if sampling > 1:
+        # Sampling thins the set but limit still truncates it, and truncation is invisible in the
+        # response: you get frames that look like a spread but are only its tail. So size the
+        # sampled set once (the same count the endpoint returns, continuity rows included), use it
+        # to fill in an omitted limit, and report it either way so a caller passing an explicit
+        # limit can still tell whether it covered the sequence.
+        counts = await get_detection_counts_by_sequence_ids(session, [sequence_id])
+        sampled_total = -(-counts.get(sequence_id, 0) // sampling)  # ceil division
+        if limit is None:
+            effective_limit = max(1, min(MAX_DETECTION_LIMIT, sampled_total))
+        response.headers["X-Sampled-Total"] = str(sampled_total)
+        response.headers["X-Sampled-Truncated"] = str(offset + effective_limit < sampled_total).lower()
+
     # 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,60 @@ 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``.
+
+        The row number is always computed ascending on ``created_at``, so the sampled frame set
+        is the same whatever ``order_desc`` is: the latter only flips the output order. That
+        also keeps the set stable as the sequence grows, since a new detection lands last and
+        cannot renumber earlier rows, so the player keeps hitting the same frames (and the same
+        cached URLs) across polls. ``(rn - 1) % sampling == 0`` keeps the first detection, so a
+        non-empty sequence always yields at least one row. ``limit``/``offset`` page the
+        *sampled* set, not the raw rows.
+
+        Note that ``limit`` cannot push down: ``row_number()`` has to cover every row of the
+        sequence before the modulo and the limit apply, which is why detections is indexed on
+        ``(sequence_id, created_at)``.
+        """
+        if sampling <= 1:
+            # Unchanged pre-sampling behaviour, on purpose: same query, same ordering.
+            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())
+        )
+        # sqlalchemy's select for the numbering subquery (two entities, and .subquery() on it);
+        # sqlmodel's select for the outer one, since a single-entity SelectOfScalar is what makes
+        # session.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)
+        stmt: Any = (
+            select(sampled)
+            .where((subq.c.rn - 1) % 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)
+            .offset(offset)
+        )
+        result = await self.session.exec(stmt)
+        return list(result.all())