feat(robot-server): Add GET runs/:runId/currentState

api/src/opentrons/protocol_runner/run_orchestrator.py

@@ -14,6 +14,7 @@
 from . import protocol_runner, RunResult, JsonRunner, PythonAndLegacyRunner
 from ..hardware_control import HardwareControlAPI
 from ..hardware_control.modules import AbstractModule as HardwareModuleAPI
+from ..hardware_control.nozzle_manager import NozzleMap
 from ..protocol_engine import (
     ProtocolEngine,
     CommandCreate,
@@ -397,6 +398,12 @@ def get_deck_type(self) -> DeckType:
         """Get engine deck type."""
         return self._protocol_engine.state_view.config.deck_type
 
+    def get_nozzle_map(self, pipette_id: str) -> NozzleMap:
+        """Get nozzle map for a pipette."""
+        return self._protocol_engine.state_view.tips.get_pipette_nozzle_map(
+            pipette_id=pipette_id
+        )
+
     def set_error_recovery_policy(self, policy: ErrorRecoveryPolicy) -> None:
         """Create error recovery policy for the run."""
         self._protocol_engine.set_error_recovery_policy(policy)

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

@@ -46,11 +46,15 @@
 )
 from robot_server.protocols.router import ProtocolNotFound
 
-from ..run_models import RunNotFoundError
+from ..run_models import RunNotFoundError, ActiveNozzleLayout
 from ..run_auto_deleter import RunAutoDeleter
 from ..run_models import Run, BadRun, RunCreate, RunUpdate
 from ..run_orchestrator_store import RunConflictError
-from ..run_data_manager import RunDataManager, RunNotCurrentError
+from ..run_data_manager import (
+    RunDataManager,
+    RunNotCurrentError,
+    NozzleMapNotFoundError,
+)
 from ..dependencies import (
     get_run_data_manager,
     get_run_auto_deleter,
@@ -106,6 +110,14 @@ class RunStopped(ErrorDetails):
     errorCode: str = ErrorCodes.GENERAL_ERROR.value.code
 
 
+class NozzleMapNotFound(ErrorDetails):
+    """An error if one tries to modify a stopped run."""
+
+    id: Literal["NozzleMapNotFound"] = "NozzleMapNotFound"
+    title: str = "Nozzle Map Not Found"
+    errorCode: str = ErrorCodes.GENERAL_ERROR.value.code
+
+
 class AllRunsLinks(BaseModel):
     """Links returned along with a collection of runs."""
 
@@ -523,3 +535,53 @@ async def get_run_commands_error(
         ),
         status_code=status.HTTP_200_OK,
     )
+
+
+@PydanticResponse.wrap_route(
+    base_router.get,
+    path="/runs/{runId}/activeNozzleLayout/{pipetteId}",
+    summary="Get the current run's active nozzle layout for a specific pipette.",
+    description=dedent(
+        """
+        Get the active nozzle layout for a specific pipette.
+        """
+    ),
+    responses={
+        status.HTTP_200_OK: {"model": SimpleBody[ActiveNozzleLayout]},
+        # status.HTTP_404_NOT_FOUND: {
+        #     "model": ErrorBody[Union[RunNotFound, PipetteNotFound]]
+        # },
+        status.HTTP_409_CONFLICT: {"model": ErrorBody[RunStopped]},
+    },
+)
+async def get_active_nozzle_layout(
+    runId: str,
+    pipetteId: str,
+    run_data_manager: Annotated[RunDataManager, Depends(get_run_data_manager)],
+) -> PydanticResponse[SimpleBody[ActiveNozzleLayout]]:
+    """Get the active nozzle layout for a specific pipette in a run.
+
+    Arguments:
+        runId: Run ID pulled from URL.
+        pipetteId: Pipette ID pulled from URL.
+        run_data_manager: Run data retrieval interface.
+    """
+    try:
+        active_nozzle_map = run_data_manager.get_nozzle_map(
+            run_id=runId, pipette_id=pipetteId
+        )
+    except NozzleMapNotFoundError as e:
+        raise NozzleMapNotFound(detail=str(e)).as_error(status.HTTP_404_NOT_FOUND)
+    except RunNotCurrentError as e:
+        raise RunStopped(detail=str(e)).as_error(status.HTTP_409_CONFLICT)
+
+    return await PydanticResponse.create(
+        content=SimpleBody.construct(
+            data=ActiveNozzleLayout.construct(
+                configuration=active_nozzle_map.configuration,
+                columns=active_nozzle_map.columns,
+                rows=active_nozzle_map.rows,
+            )
+        ),
+        status_code=status.HTTP_200_OK,
+    )

robot-server/robot_server/runs/run_data_manager.py

@@ -4,6 +4,9 @@
 
 from opentrons_shared_data.labware.labware_definition import LabwareDefinition
 from opentrons_shared_data.errors.exceptions import InvalidStoredData, EnumeratedError
+
+from opentrons.hardware_control.nozzle_manager import NozzleMap
+
 from opentrons.protocol_engine import (
     EngineStatus,
     LabwareOffsetCreate,
@@ -123,6 +126,10 @@ class RunNotCurrentError(ValueError):
     """Error raised when a requested run is not the current run."""
 
 
+class NozzleMapNotFoundError(ValueError):
+    """Error raised when a requested nozzle map cannot be found."""
+
+
 class PreSerializedCommandsNotAvailableError(LookupError):
     """Error raised when a run's commands are not available as pre-serialized list of commands."""
 
@@ -473,6 +480,16 @@ def get_command_errors(self, run_id: str) -> list[ErrorOccurrence]:
         # TODO(tz, 8-5-2024): Change this to return the error list from the DB when we implement https://opentrons.atlassian.net/browse/EXEC-655.
         raise RunNotCurrentError()
 
+    def get_nozzle_map(self, run_id: str, pipette_id: str) -> NozzleMap:
+        """Get nozzle map for a pipette."""
+        if run_id == self._run_orchestrator_store.current_run_id:
+            try:
+                return self._run_orchestrator_store.get_nozzle_map(pipette_id)
+            except KeyError:
+                raise NozzleMapNotFoundError()
+
+        raise RunNotCurrentError()
+
     def get_all_commands_as_preserialized_list(
         self, run_id: str, include_fixit_commands: bool
     ) -> List[str]:

robot-server/robot_server/runs/run_models.py

@@ -1,7 +1,7 @@
 """Request and response models for run resources."""
 from datetime import datetime
 from pydantic import BaseModel, Field
-from typing import List, Optional, Literal
+from typing import List, Optional, Literal, Dict
 
 from opentrons.protocol_engine import (
     CommandStatus,
@@ -24,6 +24,7 @@
     CSVRunTimeParamFilesType,
 )
 from opentrons_shared_data.errors import GeneralError
+from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 from robot_server.service.json_api import ResourceModel
 from robot_server.errors.error_responses import ErrorDetails
 from .action_models import RunAction
@@ -280,6 +281,22 @@ class LabwareDefinitionSummary(BaseModel):
     )
 
 
+class ActiveNozzleLayout(BaseModel):
+    """Details about the active nozzle layout for a pipette."""
+
+    configuration: "NozzleConfigurationType" = Field(
+        ..., description="The active nozzle configuration."
+    )
+    columns: Dict[str, List[str]] = Field(
+        ...,
+        description="A map of all the pipette columns active in the current configuration.",
+    )
+    rows: Dict[str, List[str]] = Field(
+        ...,
+        description="A map of all the pipette rows active in the current configuration.",
+    )
+
+
 class RunNotFoundError(GeneralError):
     """Error raised when a given Run ID is not found in the store."""
 

robot-server/robot_server/runs/run_orchestrator_store.py

@@ -16,6 +16,7 @@
 
 from opentrons.config import feature_flags
 from opentrons.hardware_control import HardwareControlAPI
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 from opentrons.hardware_control.types import (
     EstopState,
     HardwareEvent,
@@ -324,6 +325,10 @@ def get_loaded_labware_definitions(self) -> List[LabwareDefinition]:
         """Get loaded labware definitions."""
         return self.run_orchestrator.get_loaded_labware_definitions()
 
+    def get_nozzle_map(self, pipette_id: str) -> NozzleMap:
+        """Get nozzle map for a pipette."""
+        return self.run_orchestrator.get_nozzle_map(pipette_id=pipette_id)
+
     def get_run_time_parameters(self) -> List[RunTimeParameter]:
         """Parameter definitions defined by protocol, if any. Will always be empty before execution."""
         return self.run_orchestrator.get_run_time_parameters()

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

@@ -13,6 +13,8 @@
 )
 from opentrons.protocol_reader import ProtocolSource, JsonProtocolConfig
 
+from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType, NozzleMap
+
 from robot_server.data_files.data_files_store import DataFilesStore, DataFileInfo
 
 from robot_server.errors.error_responses import ApiError
@@ -34,9 +36,13 @@
 
 from robot_server.runs.run_auto_deleter import RunAutoDeleter
 
-from robot_server.runs.run_models import Run, RunCreate, RunUpdate
+from robot_server.runs.run_models import Run, RunCreate, RunUpdate, ActiveNozzleLayout
 from robot_server.runs.run_orchestrator_store import RunConflictError
-from robot_server.runs.run_data_manager import RunDataManager, RunNotCurrentError
+from robot_server.runs.run_data_manager import (
+    RunDataManager,
+    RunNotCurrentError,
+    NozzleMapNotFoundError,
+)
 from robot_server.runs.run_models import RunNotFoundError
 from robot_server.runs.router.base_router import (
     AllRunsLinks,
@@ -48,6 +54,7 @@
     update_run,
     put_error_recovery_policy,
     get_run_commands_error,
+    get_active_nozzle_layout,
 )
 
 from robot_server.deck_configuration.store import DeckConfigurationStore
@@ -84,6 +91,21 @@ def labware_offset_create() -> LabwareOffsetCreate:
     )
 
 
+@pytest.fixture
+def mock_nozzle_map() -> NozzleMap:
+    """Get a mock NozzleMap."""
+    return NozzleMap(
+        configuration=NozzleConfigurationType.FULL,
+        columns={"1": ["A1"]},
+        rows={"A": ["A1"]},
+        map_store={},
+        starting_nozzle="A1",
+        valid_map_key="mock-key",
+        full_instrument_map_store={},
+        full_instrument_rows={},
+    )
+
+
 async def test_create_run(
     decoy: Decoy,
     mock_run_data_manager: RunDataManager,
@@ -803,3 +825,76 @@ async def test_get_run_commands_errors_defualt_cursor(
         cursor=expected_cursor_result, totalLength=3
     )
     assert result.status_code == 200
+
+
+async def test_get_active_nozzle_layout_success(
+    decoy: Decoy,
+    mock_run_data_manager: RunDataManager,
+    mock_nozzle_map: NozzleMap,
+) -> None:
+    """It should return the active nozzle layout for a specific pipette."""
+    run_id = "test-run-id"
+    pipette_id = "test-pipette-id"
+
+    decoy.when(
+        mock_run_data_manager.get_nozzle_map(run_id=run_id, pipette_id=pipette_id)
+    ).then_return(mock_nozzle_map)
+
+    result = await get_active_nozzle_layout(
+        runId=run_id,
+        pipetteId=pipette_id,
+        run_data_manager=mock_run_data_manager,
+    )
+
+    assert result.status_code == 200
+    assert result.content.data == ActiveNozzleLayout(
+        configuration=NozzleConfigurationType.FULL,
+        columns={"1": ["A1"]},
+        rows={"A": ["A1"]},
+    )
+
+
+async def test_get_active_nozzle_layout_nozzle_map_not_found(
+    decoy: Decoy,
+    mock_run_data_manager: RunDataManager,
+) -> None:
+    """It should raise NozzleMapNotFound when the nozzle map is not found."""
+    run_id = "test-run-id"
+    pipette_id = "non-existent-pipette-id"
+
+    decoy.when(
+        mock_run_data_manager.get_nozzle_map(run_id=run_id, pipette_id=pipette_id)
+    ).then_raise(NozzleMapNotFoundError("Nozzle map not found"))
+
+    with pytest.raises(ApiError) as exc_info:
+        await get_active_nozzle_layout(
+            runId=run_id,
+            pipetteId=pipette_id,
+            run_data_manager=mock_run_data_manager,
+        )
+
+    assert exc_info.value.status_code == 404
+    assert exc_info.value.content["errors"][0]["id"] == "NozzleMapNotFound"
+
+
+async def test_get_active_nozzle_layout_run_not_current(
+    decoy: Decoy,
+    mock_run_data_manager: RunDataManager,
+) -> None:
+    """It should raise RunStopped when the run is not current."""
+    run_id = "non-current-run-id"
+    pipette_id = "test-pipette-id"
+
+    decoy.when(
+        mock_run_data_manager.get_nozzle_map(run_id=run_id, pipette_id=pipette_id)
+    ).then_raise(RunNotCurrentError("Run is not current"))
+
+    with pytest.raises(ApiError) as exc_info:
+        await get_active_nozzle_layout(
+            runId=run_id,
+            pipetteId=pipette_id,
+            run_data_manager=mock_run_data_manager,
+        )
+
+    assert exc_info.value.status_code == 409
+    assert exc_info.value.content["errors"][0]["id"] == "RunStopped"