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

api/src/opentrons/protocol_engine/state/tips.py

@@ -486,6 +486,10 @@ def get_pipette_nozzle_map(self, pipette_id: str) -> NozzleMap:
         """Get the current nozzle map the given pipette's configuration."""
         return self._state.nozzle_map_by_pipette_id[pipette_id]
 
+    def get_pipette_nozzle_maps(self) -> Dict[str, NozzleMap]:
+        """Get current nozzle maps keyed by pipette id."""
+        return self._state.nozzle_map_by_pipette_id
+
     def has_clean_tip(self, labware_id: str, well_name: str) -> bool:
         """Get whether a well in a labware has a clean tip.
 

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,10 @@ def get_deck_type(self) -> DeckType:
         """Get engine deck type."""
         return self._protocol_engine.state_view.config.deck_type
 
+    def get_nozzle_maps(self) -> Dict[str, NozzleMap]:
+        """Get current nozzle maps keyed by pipette id."""
+        return self._protocol_engine.state_view.tips.get_pipette_nozzle_maps()
+
     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

@@ -37,6 +37,7 @@
     MultiBodyMeta,
     ResourceLink,
     PydanticResponse,
+    Body,
 )
 
 from robot_server.protocols.dependencies import get_protocol_store
@@ -46,11 +47,14 @@
 )
 from robot_server.protocols.router import ProtocolNotFound
 
-from ..run_models import RunNotFoundError
+from ..run_models import RunNotFoundError, ActiveNozzleLayout, RunCurrentState
 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,
+)
 from ..dependencies import (
     get_run_data_manager,
     get_run_auto_deleter,
@@ -115,6 +119,15 @@ class AllRunsLinks(BaseModel):
     )
 
 
+class CurrentStateLinks(BaseModel):
+    """Links returned with the current state of a run."""
+
+    active_command: Optional[ResourceLink] = Field(
+        None,
+        description="Path to the command active when current state was reported.",
+    )
+
+
 async def get_run_data_from_url(
     runId: str,
     run_data_manager: Annotated[RunDataManager, Depends(get_run_data_manager)],
@@ -523,3 +536,60 @@ async def get_run_commands_error(
         ),
         status_code=status.HTTP_200_OK,
     )
+
+
+@PydanticResponse.wrap_route(
+    base_router.get,
+    path="/runs/{runId}/currentState",
+    summary="Get a run's current state.",
+    description=dedent(
+        """
+        Get current state associated with a run if the run is current.
+        """
+    ),
+    responses={
+        status.HTTP_200_OK: {"model": SimpleBody[RunCurrentState]},
+        status.HTTP_409_CONFLICT: {"model": ErrorBody[RunStopped]},
+    },
+)
+async def get_current_state(
+    runId: str,
+    run_data_manager: Annotated[RunDataManager, Depends(get_run_data_manager)],
+) -> PydanticResponse[Body[RunCurrentState, CurrentStateLinks]]:
+    """Get current state associated with a run if the run is current.
+
+    Arguments:
+        runId: Run ID pulled from URL.
+        run_data_manager: Run data retrieval interface.
+    """
+    try:
+        active_nozzle_maps = run_data_manager.get_nozzle_maps(run_id=runId)
+
+        nozzle_layouts = {
+            pipetteId: ActiveNozzleLayout.construct(
+                startingNozzle=nozzle_map.starting_nozzle,
+                activeNozzles=list(nozzle_map.map_store.keys()),
+                config=nozzle_map.configuration.value,
+            )
+            for pipetteId, nozzle_map in active_nozzle_maps.items()
+        }
+
+        current_command = run_data_manager.get_current_command(run_id=runId)
+    except RunNotCurrentError as e:
+        raise RunStopped(detail=str(e)).as_error(status.HTTP_409_CONFLICT)
+
+    links = CurrentStateLinks.construct(
+        active_command=ResourceLink.construct(
+            href=f"/runs/{runId}/commands/{current_command.command_id}"
+        )
+        if current_command is not None
+        else None
+    )
+
+    return await PydanticResponse.create(
+        content=Body.construct(
+            data=RunCurrentState.construct(activeNozzleLayouts=nozzle_layouts),
+            links=links,
+        ),
+        status_code=status.HTTP_200_OK,
+    )

robot-server/robot_server/runs/run_data_manager.py

@@ -1,9 +1,12 @@
 """Manage current and historical run data."""
 from datetime import datetime
-from typing import List, Optional, Callable, Union
+from typing import List, Optional, Callable, Union, Dict
 
 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,
@@ -473,6 +476,13 @@ 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_maps(self, run_id: str) -> Dict[str, NozzleMap]:
+        """Get current nozzle maps keyed by pipette id."""
+        if run_id == self._run_orchestrator_store.current_run_id:
+            return self._run_orchestrator_store.get_nozzle_maps()
+
+        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,
@@ -280,6 +280,25 @@ class LabwareDefinitionSummary(BaseModel):
     )
 
 
+class ActiveNozzleLayout(BaseModel):
+    """Details about the active nozzle layout for a pipette used in the current run."""
+
+    startingNozzle: str = Field(
+        ..., description="The nozzle used when issuing pipette commands."
+    )
+    activeNozzles: List[str] = Field(
+        ...,
+        description="A map of all the pipette nozzles active in the current configuration.",
+    )
+    config: str = Field(..., description="The nozzle configuration type.")
+
+
+class RunCurrentState(BaseModel):
+    """Current details about a run."""
+
+    activeNozzleLayouts: Dict[str, ActiveNozzleLayout] = Field(..., description="")
+
+
 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

@@ -1,7 +1,7 @@
 """In-memory storage of ProtocolEngine instances."""
 import asyncio
 import logging
-from typing import List, Optional, Callable
+from typing import List, Optional, Callable, Dict
 
 from opentrons.protocol_engine.errors.exceptions import EStopActivatedError
 from opentrons.protocol_engine.types import (
@@ -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_maps(self) -> Dict[str, NozzleMap]:
+        """Get the current nozzle map keyed by pipette id."""
+        return self.run_orchestrator.get_nozzle_maps()
+
     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

@@ -1,18 +1,23 @@
 """Tests for base /runs routes."""
+from typing import Dict
+
 import pytest
 from datetime import datetime
 from decoy import Decoy
 from pathlib import Path
 
-from opentrons.types import DeckSlotName
+from opentrons.types import DeckSlotName, Point
 from opentrons.protocol_engine import (
     LabwareOffsetCreate,
     types as pe_types,
     errors as pe_errors,
     CommandErrorSlice,
+    CommandPointer,
 )
 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 +39,18 @@
 
 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,
+    RunCurrentState,
+    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,
+)
 from robot_server.runs.run_models import RunNotFoundError
 from robot_server.runs.router.base_router import (
     AllRunsLinks,
@@ -48,6 +62,8 @@
     update_run,
     put_error_recovery_policy,
     get_run_commands_error,
+    get_current_state,
+    CurrentStateLinks,
 )
 
 from robot_server.deck_configuration.store import DeckConfigurationStore
@@ -84,6 +100,23 @@ def labware_offset_create() -> LabwareOffsetCreate:
     )
 
 
+@pytest.fixture
+def mock_nozzle_maps() -> Dict[str, NozzleMap]:
+    """Get mock NozzleMaps."""
+    return {
+        "mock-pipette-id": NozzleMap(
+            configuration=NozzleConfigurationType.FULL,
+            columns={"1": ["A1"]},
+            rows={"A": ["A1"]},
+            map_store={"A1": Point(0, 0, 0)},
+            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 +836,64 @@ async def test_get_run_commands_errors_defualt_cursor(
         cursor=expected_cursor_result, totalLength=3
     )
     assert result.status_code == 200
+
+
+async def test_get_current_state_success(
+    decoy: Decoy,
+    mock_run_data_manager: RunDataManager,
+    mock_nozzle_maps: Dict[str, NozzleMap],
+) -> None:
+    """It should return the active nozzle layout for a specific pipette."""
+    run_id = "test-run-id"
+
+    decoy.when(mock_run_data_manager.get_nozzle_maps(run_id=run_id)).then_return(
+        mock_nozzle_maps
+    )
+    decoy.when(mock_run_data_manager.get_current_command(run_id=run_id)).then_return(
+        CommandPointer(
+            command_id="current-command-id",
+            command_key="current-command-key",
+            created_at=datetime(year=2024, month=4, day=4),
+            index=101,
+        )
+    )
+
+    result = await get_current_state(
+        runId=run_id,
+        run_data_manager=mock_run_data_manager,
+    )
+
+    assert result.status_code == 200
+    assert result.content.data == RunCurrentState.construct(
+        activeNozzleLayouts={
+            "mock-pipette-id": ActiveNozzleLayout(
+                startingNozzle="A1", activeNozzles=["A1"], config="FULL"
+            )
+        }
+    )
+    assert result.content.links == CurrentStateLinks(
+        active_command=ResourceLink(
+            href="/runs/test-run-id/commands/current-command-id", meta=None
+        )
+    )
+
+
+async def test_get_current_state_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"
+
+    decoy.when(mock_run_data_manager.get_nozzle_maps(run_id=run_id)).then_raise(
+        RunNotCurrentError("Run is not current")
+    )
+
+    with pytest.raises(ApiError) as exc_info:
+        await get_current_state(
+            runId=run_id,
+            run_data_manager=mock_run_data_manager,
+        )
+
+    assert exc_info.value.status_code == 409
+    assert exc_info.value.content["errors"][0]["id"] == "RunStopped"