/currentState returning more relevant info for all run utilized pipettes

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

@@ -487,7 +487,7 @@ def get_pipette_nozzle_map(self, pipette_id: str) -> NozzleMap:
         return self._state.nozzle_map_by_pipette_id[pipette_id]
 
     def get_pipette_nozzle_maps(self) -> Dict[str, NozzleMap]:
-        """Get the current nozzle map keyed by attached pipette id."""
+        """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:

api/src/opentrons/protocol_runner/run_orchestrator.py

@@ -399,7 +399,7 @@ def get_deck_type(self) -> DeckType:
         return self._protocol_engine.state_view.config.deck_type
 
     def get_nozzle_maps(self) -> Dict[str, NozzleMap]:
-        """Get the current nozzle map keyed by pipette id."""
+        """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:

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,7 +47,7 @@
 )
 from robot_server.protocols.router import ProtocolNotFound
 
-from ..run_models import RunNotFoundError, ActiveNozzleLayout
+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
@@ -109,14 +110,6 @@ class RunStopped(ErrorDetails):
     errorCode: str = ErrorCodes.GENERAL_ERROR.value.code
 
 
-class NozzleMapNotFound(ErrorDetails):
-    """An error if a nozzle map is not found."""
-
-    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."""
 
@@ -126,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)],
@@ -546,26 +548,48 @@ async def get_run_commands_error(
         """
     ),
     responses={
-        status.HTTP_200_OK: {"model": SimpleBody[ActiveNozzleLayout]},
+        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[SimpleBody[ActiveNozzleLayout]]:
-    """Get current state associated with a run.
+) -> 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=SimpleBody.construct(data=active_nozzle_maps),
+        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

@@ -477,7 +477,7 @@ def get_command_errors(self, run_id: str) -> list[ErrorOccurrence]:
         raise RunNotCurrentError()
 
     def get_nozzle_maps(self, run_id: str) -> Dict[str, NozzleMap]:
-        """Get the current nozzle map keyed pipette id."""
+        """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()
 

robot-server/robot_server/runs/run_models.py

@@ -24,7 +24,6 @@
     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
@@ -282,19 +281,22 @@ class LabwareDefinitionSummary(BaseModel):
 
 
 class ActiveNozzleLayout(BaseModel):
-    """Details about the active nozzle layout for a pipette."""
+    """Details about the active nozzle layout for a pipette used in the current run."""
 
-    configuration: "NozzleConfigurationType" = Field(
-        ..., description="The active nozzle configuration."
+    startingNozzle: str = Field(
+        ..., description="The nozzle used when issuing pipette commands."
     )
-    columns: Dict[str, List[str]] = Field(
+    activeNozzles: 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.",
+        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):

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

@@ -1,15 +1,18 @@
 """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
 
@@ -36,12 +39,17 @@
 
 from robot_server.runs.run_auto_deleter import RunAutoDeleter
 
-from robot_server.runs.run_models import Run, RunCreate, RunUpdate, ActiveNozzleLayout
+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,
-    NozzleMapNotFoundError,
 )
 from robot_server.runs.run_models import RunNotFoundError
 from robot_server.runs.router.base_router import (
@@ -54,7 +62,8 @@
     update_run,
     put_error_recovery_policy,
     get_run_commands_error,
-    get_active_nozzle_layout,
+    get_current_state,
+    CurrentStateLinks,
 )
 
 from robot_server.deck_configuration.store import DeckConfigurationStore
@@ -92,18 +101,20 @@ 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={},
-    )
+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(
@@ -827,72 +838,60 @@ async def test_get_run_commands_errors_defualt_cursor(
     assert result.status_code == 200
 
 
-async def test_get_active_nozzle_layout_success(
+async def test_get_current_state_success(
     decoy: Decoy,
     mock_run_data_manager: RunDataManager,
-    mock_nozzle_map: NozzleMap,
+    mock_nozzle_maps: Dict[str, 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_maps(run_id=run_id, pipette_id=pipette_id)
-    ).then_return(mock_nozzle_map)
+    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_active_nozzle_layout(
+    result = await get_current_state(
         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"]},
+    assert result.content.data == RunCurrentState.construct(
+        activeNozzleLayouts={
+            "mock-pipette-id": ActiveNozzleLayout(
+                startingNozzle="A1", activeNozzles=["A1"], config="FULL"
+            )
+        }
     )
-
-
-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_maps(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 result.content.links == CurrentStateLinks(
+        active_command=ResourceLink(
+            href="/runs/test-run-id/commands/current-command-id", meta=None
         )
-
-    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(
+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"
-    pipette_id = "test-pipette-id"
 
-    decoy.when(
-        mock_run_data_manager.get_nozzle_maps(run_id=run_id, pipette_id=pipette_id)
-    ).then_raise(RunNotCurrentError("Run is not current"))
+    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_active_nozzle_layout(
+        await get_current_state(
             runId=run_id,
-            pipetteId=pipette_id,
             run_data_manager=mock_run_data_manager,
         )