feat(robot-server): Implementation for preview image capture (#20368)

Covers EXEC-2129
Implements the `/camera/capturePreviewImage` and `/runs/{runId}/camera/capturePreviewImage` endpoints which captures and returns a singular photo with the embedded camera .

Author: CaseyBatten

api/src/opentrons/system/camera.py

@@ -23,6 +23,9 @@
 # Default System Cameras
 DEFAULT_SYSTEM_CAMERA = "/dev/ot_system_camera"
 
+# Default Preview Image Filename
+PREVIEW_IMAGE = "preview_image.jpeg"
+
 # Stream Globals
 DEFAULT_CONF_FILE = (
     "/lib/systemd/system/opentrons-live-stream/opentrons-live-stream.env"

robot-server/robot_server/data_files/models.py

@@ -70,6 +70,13 @@ class FileIdNotFound(ErrorDetails):
     title: str = "Specified file id not found on the robot"
 
 
+class FileNotFound(ErrorDetails):
+    """An error returned when specified file path was not found on the robot."""
+
+    id: Literal["FileNotFound"] = "FileNotFound"
+    title: str = "Specified file path not found on the robot"
+
+
 class NoImagesFound(ErrorDetails):
     """An error returned when no images are found for the specified run."""
 

robot-server/robot_server/data_files/router.py

@@ -38,6 +38,7 @@
     DataFile,
     FileIdNotFoundError,
     FileIdNotFound,
+    FileNotFound,
     FileInUseError,
     ImageFileMetadata,
     NoImagesFound,
@@ -75,13 +76,6 @@ class NoDataFileSourceProvided(ErrorDetails):
     title: str = "No data file source provided"
 
 
-class FileNotFound(ErrorDetails):
-    """An error returned when specified file path was not found on the robot."""
-
-    id: Literal["FileNotFound"] = "FileNotFound"
-    title: str = "Specified file path not found on the robot"
-
-
 class UnexpectedFileFormat(ErrorDetails):
     """An error returned when specified file is not in expected format."""
 

robot-server/robot_server/runs/router/camera_router.py

@@ -1,9 +1,11 @@
 """Router for /runs endpoints dealing with run specific camera settings and behavior."""
 
 import logging
+import os
 from typing import Annotated, Union
-
-from fastapi import Depends, status
+from pathlib import Path
+from fastapi import Depends, status, HTTPException
+from fastapi.responses import FileResponse
 from server_utils.fastapi_utils.light_router import LightRouter
 
 from opentrons.protocol_engine.resources.camera_provider import CameraSettings
@@ -22,6 +24,9 @@
     get_camera_provider,
 )
 from opentrons.protocol_engine.resources.camera_provider import CameraProvider
+from opentrons.protocol_engine.resources.camera_provider import ImageParameters
+from robot_server.persistence.fastapi_dependencies import get_images_directory
+from robot_server.data_files.models import FileNotFound
 
 from ..run_models import Run
 from ..run_orchestrator_store import RunOrchestratorStore
@@ -33,6 +38,7 @@
     CameraEnable,
     CameraCaptureImageSettings,
 )
+from opentrons.protocol_engine import EngineStatus
 
 log = logging.getLogger(__name__)
 camera_router = LightRouter()
@@ -177,3 +183,92 @@ async def add_camera_capture_image_settings(
         content=SimpleBody.model_construct(data=request_body.data),
         status_code=status.HTTP_201_CREATED,
     )
+
+
+@camera_router.post(
+    path="/runs/{runId}/camera/capturePreviewImage",
+    summary="Capture a preview image based on provided settings and the run specific camera enablement.",
+    description="Return a preview image based on provided capture image settings.",
+    responses={
+        status.HTTP_200_OK: {
+            "content": {"image/jpeg": {}},
+            "description": "Preview image taken with specific settings.",
+        },
+        status.HTTP_404_NOT_FOUND: {"model": ErrorBody[FileNotFound]},
+    },
+)
+async def post_camera_preview_image(
+    request_body: RequestModel[CameraCaptureImageSettings],
+    run: Annotated[Run, Depends(get_run_data_from_url)],
+    images_directory: Annotated[Path, Depends(get_images_directory)],
+    robot_type: Annotated[RobotType, Depends(get_robot_type)],
+) -> FileResponse:
+    """Return a preview image based on the provided capture image settings and run specific enablement."""
+    if IS_ROBOT and not camera.camera_exists():
+        # todo(chb): Eventually we'll have mulitple camera ids that can be sent, so this should be able to verify more than just the default
+        raise LegacyErrorResponse(
+            message="Video device is unavailable.",
+            errorCode=ErrorCodes.GENERAL_ERROR.value.code,
+        ).as_error(status.HTTP_503_SERVICE_UNAVAILABLE)
+
+    if run.status not in [
+        EngineStatus.IDLE,
+        EngineStatus.STOPPED,
+        EngineStatus.FAILED,
+        EngineStatus.SUCCEEDED,
+    ]:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str("Cannot capture preview photo, run is not inactive."),
+        )
+
+    image_data = await camera.image_capture(
+        robot_type=robot_type,
+        parameters=ImageParameters(
+            resolution=request_body.data.resolution,
+            zoom=request_body.data.zoom,
+            pan=request_body.data.pan,
+            contrast=(
+                (request_body.data.contrast / 100) * 2.0
+                if request_body.data.contrast is not None
+                else None
+            ),
+            brightness=(
+                int(((request_body.data.brightness * 256) // 100) - 128) * -1
+                if request_body.data.brightness is not None
+                else None
+            ),
+            saturation=(
+                (request_body.data.saturation / 100) * 2.0
+                if request_body.data.saturation is not None
+                else None
+            ),
+        ),
+    )
+
+    file_path = images_directory / camera.PREVIEW_IMAGE
+
+    if IS_ROBOT:
+        if isinstance(image_data, bytes):
+            os.makedirs(os.path.dirname(file_path), exist_ok=True)
+
+            with open(file=file_path, mode="wb") as f:
+                f.write(image_data)
+        else:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=str(
+                    f"Preview image capture failed with the following: {image_data.message}"
+                ),
+            )
+
+        if not file_path.exists():
+            raise FileNotFound(detail="Preview image file not found.").as_error(
+                status.HTTP_404_NOT_FOUND
+            )
+
+    return FileResponse(
+        path=file_path,
+        media_type="image/jpeg",
+        filename=camera.PREVIEW_IMAGE,
+    )

robot-server/robot_server/service/legacy/routers/camera.py

@@ -4,20 +4,23 @@
 import tempfile
 from typing import Annotated
 from pathlib import Path
-from fastapi import APIRouter, HTTPException, Depends
+from fastapi import APIRouter, HTTPException, Depends, Response
+from fastapi.responses import FileResponse
 from starlette import status
 from starlette.background import BackgroundTask
 from starlette.responses import StreamingResponse
 from opentrons.system import camera
-from opentrons.system.camera import StreamConfigurationKeys
+from opentrons.system.camera import StreamConfigurationKeys, PREVIEW_IMAGE
 from robot_server.errors.error_responses import LegacyErrorResponse
 from opentrons_shared_data.errors import ErrorCodes
+from robot_server.errors.error_responses import ErrorBody
 from robot_server.service.legacy.models.settings import (
     CameraEnable,
     LiveStreamData,
     LiveStreamSettings,
     Resolution,
     StreamStatusType,
+    CameraCaptureImageSettings,
 )
 from robot_server.service.json_api import RequestModel
 from opentrons.config import IS_ROBOT
@@ -30,7 +33,9 @@
     CameraSettingStore,
     get_camera_setting_store,
 )
-
+from opentrons.protocol_engine.resources.camera_provider import ImageParameters
+from robot_server.persistence.fastapi_dependencies import get_images_directory
+from robot_server.data_files.models import FileNotFound
 
 log = logging.getLogger(__name__)
 
@@ -152,6 +157,94 @@ async def get_camera(
 # todo(chb, 2025-09-08): Implement POST/GET /camera/picture/settings for picture taking settings when in protocol run
 
 
+@router.post(
+    "/camera/capturePreviewImage",
+    description="Return a preview image based on provided capture image settings.",
+    responses={
+        status.HTTP_404_NOT_FOUND: {"model": ErrorBody[FileNotFound]},
+    },
+)
+async def post_camera_preview_image(
+    request_body: RequestModel[CameraCaptureImageSettings],
+    run_data_manager: Annotated[RunDataManager, Depends(get_run_data_manager)],
+    camera_settings_store: Annotated[
+        CameraSettingStore, Depends(get_camera_setting_store)
+    ],
+    images_directory: Annotated[Path, Depends(get_images_directory)],
+    robot_type: Annotated[RobotType, Depends(get_robot_type)],
+) -> Response:
+    """
+    Return a preview image based on the provided capture image settings.
+    """
+    if run_data_manager.current_run_id is not None and (
+        run_data_manager.get(run_data_manager.current_run_id).status
+        not in [EngineStatus.STOPPED, EngineStatus.FAILED, EngineStatus.SUCCEEDED]
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str("Cannot capture preview photo, run is active."),
+        )
+
+    _validate_camera_present()
+
+    if not camera_settings_store.get_camera_enabled():
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str("Cannot capture preview photo, camera is disabled on robot."),
+        )
+
+    image_data = await camera.image_capture(
+        robot_type=robot_type,
+        parameters=ImageParameters(
+            resolution=request_body.data.resolution,
+            zoom=request_body.data.zoom,
+            pan=request_body.data.pan,
+            contrast=(
+                (request_body.data.contrast / 100) * 2.0
+                if request_body.data.contrast is not None
+                else None
+            ),
+            brightness=(
+                int(((request_body.data.brightness * 256) // 100) - 128) * -1
+                if request_body.data.brightness is not None
+                else None
+            ),
+            saturation=(
+                (request_body.data.saturation / 100) * 2.0
+                if request_body.data.saturation is not None
+                else None
+            ),
+        ),
+    )
+
+    file_path = images_directory / PREVIEW_IMAGE
+
+    if IS_ROBOT:
+        if isinstance(image_data, bytes):
+            os.makedirs(os.path.dirname(file_path), exist_ok=True)
+
+            with open(file=file_path, mode="wb") as f:
+                f.write(image_data)
+        else:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=str(
+                    f"Preview image capture failed with the following: {image_data.message}"
+                ),
+            )
+
+        if not file_path.exists():
+            raise FileNotFound(detail="Preview image file not found.").as_error(
+                status.HTTP_404_NOT_FOUND
+            )
+
+    return FileResponse(
+        path=file_path,
+        media_type="image/jpeg",
+        filename=PREVIEW_IMAGE,
+    )
+
+
 @router.post(
     "/camera/picture",
     description="Capture an image from the OT-2's on-board camera " "and return it",

robot-server/tests/runs/router/test_camera_router.py

@@ -1,19 +1,21 @@
 """Tests for camera /runs/{runId}/camera routes."""
 import pytest
 from datetime import datetime
+from pathlib import Path
+import tempfile
 from decoy import Decoy
 
 from robot_server.service.json_api import RequestModel
 
-from robot_server.runs.router.camera_router import add_camera_capture_image_settings
-
-from robot_server.service.legacy.models.settings import (
-    CameraCaptureImageSettings,
+from robot_server.runs.router.camera_router import (
+    add_camera_capture_image_settings,
+    post_camera_preview_image,
 )
-
 from robot_server.runs.run_models import Run
 from robot_server.runs.run_orchestrator_store import RunOrchestratorStore
 from opentrons.protocol_engine import EngineStatus
+from robot_server.service.legacy.models.settings import CameraCaptureImageSettings
+from fastapi.responses import FileResponse
 
 
 @pytest.fixture()
@@ -76,3 +78,28 @@ async def test_camera_settings(
 
     assert result.content.data == image_settings
     assert result.status_code == 201
+
+
+async def test_camera_preview_image(
+    decoy: Decoy,
+    run: Run,
+) -> None:
+    """Test that we can request a preview image with a collection of image settings based on run specific enablement."""
+    with tempfile.NamedTemporaryFile() as conf:
+        response = await post_camera_preview_image(
+            request_body=RequestModel(
+                data=CameraCaptureImageSettings(
+                    cameraId=None,
+                    resolution=(720, 1280),
+                    zoom=1.5,
+                    pan=(0, 0),
+                    contrast=25.0,
+                    brightness=50.0,
+                    saturation=75.0,
+                )
+            ),
+            run=run,
+            images_directory=Path(conf.name),
+            robot_type="OT-3 Standard",
+        )
+        assert isinstance(response, FileResponse)