feat(inspect): show dataset snapshot (#3168)

* add snapshot download endpoint

Signed-off-by: Ma, Xiangxiang <[email protected]>

* fix snapshot timestamp and test

Signed-off-by: Ma, Xiangxiang <[email protected]>

* remove camera macos workaround

Signed-off-by: Ma, Xiangxiang <[email protected]>

* address comments

Signed-off-by: Ma, Xiangxiang <[email protected]>

* Update application/backend/src/api/dependencies.py

Co-authored-by: Copilot <[email protected]>
Signed-off-by: Max Xiang <[email protected]>

* rename to item

Signed-off-by: Ma, Xiangxiang <[email protected]>

* fix test

---------

Signed-off-by: Ma, Xiangxiang <[email protected]>
Signed-off-by: Max Xiang <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: maxxgx

application/backend/pyproject.toml

@@ -23,7 +23,8 @@ dependencies = [
     "uvloop==0.21.0",
     "loguru==0.7.3",
     "trackio~=0.6.0",
-    "cv2-enumerate-cameras==1.3.0",
+    "sse-starlette~=3.0",
+    "cv2-enumerate-cameras==1.3.1",
     "pyarrow~=21.0.0",
 ]
 

application/backend/src/alembic/versions/7a213a27d666_initial_schema.py

@@ -103,6 +103,7 @@ def upgrade() -> None:
         sa.Column("id", sa.Text(), nullable=False),
         sa.Column("project_id", sa.String(), nullable=False),
         sa.Column("filename", sa.Text(), nullable=False),
+        sa.Column("count", sa.Integer(), nullable=False),
         sa.Column("created_at", sa.DateTime(), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
         sa.PrimaryKeyConstraint("id"),
     )

application/backend/src/api/dependencies.py

@@ -154,6 +154,11 @@ async def get_job_id(job_id: str) -> UUID:
     return get_uuid(job_id, "job")
 
 
+async def get_snapshot_id(snapshot_id: str) -> UUID:
+    """Initializes and validates a Snapshot ID"""
+    return get_uuid(snapshot_id, "snapshot")
+
+
 async def get_webrtc_manager(request: Request) -> WebRTCManager:
     """Provides the global WebRTCManager instance from FastAPI application's state."""
     return request.app.state.webrtc_manager

application/backend/src/api/endpoints/snapshot_endpoints.py

@@ -0,0 +1,69 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import FileResponse
+
+from api.dependencies import get_project_id, get_snapshot_id
+from api.endpoints import API_PREFIX
+from pydantic_models.dataset_snapshot import DatasetSnapshot, DatasetSnapshotList
+from repositories.binary_repo import DatasetSnapshotBinaryRepository
+from services.dataset_snapshot_service import DatasetSnapshotService
+
+router = APIRouter(
+    prefix=API_PREFIX + "/projects/{project_id}",
+    tags=["snapshots"],
+)
+
+
+@router.get(
+    "/snapshots",
+    responses={
+        status.HTTP_200_OK: {"description": "List of available dataset snapshots"},
+    },
+    response_model_exclude_none=True,
+)
+async def get_snapshot_list(
+    project_id: Annotated[UUID, Depends(get_project_id)],
+) -> DatasetSnapshotList:
+    """Endpoint to get list of all snapshots"""
+    snapshots = await DatasetSnapshotService.list_snapshots(project_id=project_id)
+    return DatasetSnapshotList(snapshots=snapshots)
+
+
+@router.get(
+    "/snapshots/{snapshot_id}",
+    response_model_exclude_none=True,
+)
+async def get_snapshot(
+    project_id: Annotated[UUID, Depends(get_project_id)],
+    snapshot_id: UUID,
+) -> DatasetSnapshot:
+    """Endpoint to get snapshot file by ID"""
+    return await DatasetSnapshotService.get_snapshot(project_id=project_id, snapshot_id=snapshot_id)
+
+
+@router.get(
+    "/snapshots/{snapshot_id}/parquet",
+    response_model_exclude_none=True,
+    responses={
+        status.HTTP_200_OK: {"description": "Snapshot Parquet file retrieved successfully"},
+        status.HTTP_404_NOT_FOUND: {"description": "Snapshot not found"},
+    },
+)
+async def get_snapshot_file(
+    project_id: Annotated[UUID, Depends(get_project_id)],
+    snapshot_id: Annotated[UUID, Depends(get_snapshot_id)],
+) -> FileResponse:
+    """Endpoint to get snapshot file by ID"""
+    try:
+        snapshot_bin_repo = DatasetSnapshotBinaryRepository(project_id=project_id)
+        snapshot_path = snapshot_bin_repo.get_snapshot_path(snapshot_id)
+        return FileResponse(snapshot_path)
+    except FileNotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Snapshot with ID {snapshot_id} not found",
+        ) from e

application/backend/src/db/schema.py

@@ -45,6 +45,7 @@ class DatasetSnapshotDB(Base):
     id: Mapped[str] = mapped_column(primary_key=True, default=lambda: str(uuid4()))
     project_id: Mapped[str] = mapped_column(ForeignKey("projects.id"))
     filename: Mapped[str] = mapped_column(Text, nullable=False)
+    count: Mapped[int] = mapped_column(nullable=False)
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.current_timestamp())
 
 

application/backend/src/main.py

@@ -16,6 +16,7 @@
 from api.endpoints.pipeline_endpoints import router as pipeline_router
 from api.endpoints.project_endpoints import project_router
 from api.endpoints.sink_endpoints import router as sink_router
+from api.endpoints.snapshot_endpoints import router as snapshot_router
 from api.endpoints.source_endpoints import router as source_router
 from api.endpoints.trainable_models_endpoints import router as trainable_model_router
 from api.endpoints.webrtc import router as webrtc_router
@@ -58,6 +59,7 @@
 app.include_router(trainable_model_router)
 app.include_router(device_router)
 app.include_router(capture_router)
+app.include_router(snapshot_router)
 
 
 if __name__ == "__main__":

application/backend/src/pydantic_models/dataset_snapshot.py

@@ -4,7 +4,7 @@
 from datetime import datetime
 from uuid import UUID
 
-from pydantic import Field
+from pydantic import BaseModel, Field
 
 from pydantic_models.base import BaseIDModel
 
@@ -16,15 +16,21 @@ class DatasetSnapshot(BaseIDModel):
 
     project_id: UUID
     filename: str
-    created_at: datetime = Field(default_factory=datetime.now)
+    count: int
+    created_at: datetime | None = Field(default=None)
 
     model_config = {
         "json_schema_extra": {
             "example": {
                 "id": "76e07d18-196e-4e33-bf98-ac1d35dca4cb",
                 "project_id": "16e07d18-196e-4e33-bf98-ac1d35dcaaaa",
                 "filename": "dataset_snapshot_123.parquet",
+                "count": 42,
                 "created_at": "2025-01-01T12:00:00",
             }
         }
     }
+
+
+class DatasetSnapshotList(BaseModel):
+    snapshots: list[DatasetSnapshot]

application/backend/src/repositories/mappers/dataset_snapshot_mapper.py

@@ -16,6 +16,7 @@ def to_schema(model: DatasetSnapshot) -> DatasetSnapshotDB:
             id=str(model.id),
             project_id=str(model.project_id),
             filename=model.filename,
+            count=model.count,
             created_at=model.created_at,
         )
 
@@ -25,5 +26,6 @@ def from_schema(schema: DatasetSnapshotDB) -> DatasetSnapshot:
             id=UUID(schema.id),
             project_id=UUID(schema.project_id),
             filename=schema.filename,
+            count=schema.count,
             created_at=schema.created_at,
         )

application/backend/src/repositories/project_repo.py

@@ -5,6 +5,7 @@
 from uuid import UUID
 
 import sqlalchemy as sa
+from loguru import logger
 from sqlalchemy.ext.asyncio.session import AsyncSession
 
 from db.schema import PipelineDB, ProjectDB
@@ -47,6 +48,7 @@ async def update_dataset_timestamp(self, project_id: str | UUID) -> None:
                 updated_at=sa.func.current_timestamp(),
             )
         )
+        logger.info(f"Updated dataset timestamp for project {project_id} to current time.")
         await self.db.commit()
 
     async def get_dataset_timestamp(self, project_id: str | UUID) -> datetime:

application/backend/src/services/active_pipeline_service.py

@@ -73,15 +73,15 @@ async def _initialize(
         await self._load_app_config()
 
         # For child processes with config_changed_condition, start a daemon to monitor configuration changes
-        if start_daemon is not None and self.config_changed_condition is not None:
+        if start_daemon and self.config_changed_condition is not None:
             # Store the current event loop for the daemon thread to use
             self._event_loop = asyncio.get_running_loop()
 
             self._config_reload_daemon = Thread(
                 target=self._reload_config_daemon_routine, name="Config reloader", daemon=True
             )
             self._config_reload_daemon.start()
-        elif start_daemon is not None and self.config_changed_condition is None:
+        elif self.config_changed_condition is None:
             # This is a child process but no condition provided - this is likely an API process
             # that doesn't need the daemon thread, so we just log and continue
             logger.debug("Child process detected but no config_changed_condition provided - skipping daemon thread")
@@ -102,7 +102,6 @@ async def _load_app_config(self) -> None:
         This method loads the active pipeline configuration and updates the
         internal source and sink configurations accordingly.
         """
-        logger.info("Loading configuration from database")
         async with get_async_db_session_ctx() as db:
             repo = PipelineRepository(db)
 
@@ -141,7 +140,7 @@ def _reload_config_daemon_routine(self) -> None:
                 notified = self.config_changed_condition.wait(timeout=3)
                 if not notified:  # awakened before of timeout
                     continue
-                logger.debug(f"Configuration changes detected. Process: {mp.current_process().name}")
+                logger.info(f"Configuration changes detected. Process: {mp.current_process().name}")
                 # Schedule the async reload in the event loop using the stored loop reference
                 asyncio.run(self.reload())