fix(app,robot-server): Account for failed commands not having a pipetteId

api/src/opentrons/protocol_runner/run_orchestrator.py

@@ -418,6 +418,21 @@ 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 get_tip_attached(self) -> Dict[str, bool]:
+        """Get current tip state keyed by pipette id."""
+
+        def has_tip_attached(pipette_id: str) -> bool:
+            return (
+                self._protocol_engine.state_view.pipettes.get_attached_tip(pipette_id)
+                is not None
+            )
+
+        pipette_ids = (
+            pipette.id
+            for pipette in self._protocol_engine.state_view.pipettes.get_all()
+        )
+        return {pipette_id: has_tip_attached(pipette_id) for pipette_id in pipette_ids}
+
     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

@@ -61,6 +61,7 @@
     RunCurrentState,
     CommandLinkNoMeta,
     NozzleLayoutConfig,
+    TipState,
 )
 from ..run_auto_deleter import RunAutoDeleter
 from ..run_models import Run, BadRun, RunCreate, RunUpdate
@@ -591,33 +592,27 @@ async def get_current_state(  # noqa: C901
     """
     try:
         run = run_data_manager.get(run_id=runId)
-        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=NozzleLayoutConfig(nozzle_map.configuration.value.lower()),
-            )
-            for pipetteId, nozzle_map in active_nozzle_maps.items()
-        }
-
-        run = run_data_manager.get(run_id=runId)
-        current_command = run_data_manager.get_current_command(run_id=runId)
-        last_completed_command = run_data_manager.get_last_completed_command(
-            run_id=runId
-        )
     except RunNotCurrentError as e:
         raise RunStopped(detail=str(e)).as_error(status.HTTP_409_CONFLICT)
 
-    links = CurrentStateLinks.construct(
-        lastCompleted=CommandLinkNoMeta.construct(
-            id=last_completed_command.command_id,
-            href=f"/runs/{runId}/commands/{last_completed_command.command_id}",
+    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=NozzleLayoutConfig(nozzle_map.configuration.value.lower()),
         )
-        if last_completed_command is not None
-        else None
-    )
+        for pipetteId, nozzle_map in active_nozzle_maps.items()
+    }
+
+    tip_states = {
+        pipette_id: TipState.construct(hasTip=has_tip)
+        for pipette_id, has_tip in run_data_manager.get_tip_attached(
+            run_id=runId
+        ).items()
+    }
+
+    current_command = run_data_manager.get_current_command(run_id=runId)
 
     estop_engaged = False
     place_labware = None
@@ -672,11 +667,22 @@ async def get_current_state(  # noqa: C901
                 if place_labware:
                     break
 
+    last_completed_command = run_data_manager.get_last_completed_command(run_id=runId)
+    links = CurrentStateLinks.construct(
+        lastCompleted=CommandLinkNoMeta.construct(
+            id=last_completed_command.command_id,
+            href=f"/runs/{runId}/commands/{last_completed_command.command_id}",
+        )
+        if last_completed_command is not None
+        else None
+    )
+
     return await PydanticResponse.create(
         content=Body.construct(
             data=RunCurrentState.construct(
                 estopEngaged=estop_engaged,
                 activeNozzleLayouts=nozzle_layouts,
+                tipStates=tip_states,
                 placeLabwareState=place_labware,
             ),
             links=links,

robot-server/robot_server/runs/run_data_manager.py

@@ -516,6 +516,13 @@ def get_nozzle_maps(self, run_id: str) -> Dict[str, NozzleMap]:
 
         raise RunNotCurrentError()
 
+    def get_tip_attached(self, run_id: str) -> Dict[str, bool]:
+        """Get current tip attached states, keyed by pipette id."""
+        if run_id == self._run_orchestrator_store.current_run_id:
+            return self._run_orchestrator_store.get_tip_attached()
+
+        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

@@ -316,6 +316,16 @@ class ActiveNozzleLayout(BaseModel):
     )
 
 
+class TipState(BaseModel):
+    """Information about the tip, if any, currently attached to a pipette."""
+
+    hasTip: bool
+
+    # todo(mm, 2024-11-15): I think the frontend is currently scraping the commands
+    # list to figure out where the current tip came from. Extend this class with that
+    # information so the frontend doesn't have to do that.
+
+
 class PlaceLabwareState(BaseModel):
     """Details the labware being placed by the gripper."""
 
@@ -331,9 +341,21 @@ class PlaceLabwareState(BaseModel):
 class RunCurrentState(BaseModel):
     """Current details about a run."""
 
-    estopEngaged: bool = Field(..., description="")
-    activeNozzleLayouts: Dict[str, ActiveNozzleLayout] = Field(...)
-    placeLabwareState: Optional[PlaceLabwareState] = Field(None)
+    # todo(mm, 2024-11-15): Having estopEngaged here is a bit of an odd man out because
+    # it's sensor state that can change on its own at any time, whereas the rest of
+    # these fields are logical state that changes only when commands are run.
+    #
+    # Our current mechanism for anchoring these fields to a specific point in time
+    # (important for avoiding torn-read problems when a client combines this info with
+    # info from other endpoints) is `links.currentCommand`, which is based on the idea
+    # that these fields only change when the current command changes.
+    #
+    # We should see if clients can replace this with `GET /robot/control/estopStatus`.
+    estopEngaged: bool
+
+    activeNozzleLayouts: Dict[str, ActiveNozzleLayout]
+    tipStates: Dict[str, TipState]
+    placeLabwareState: Optional[PlaceLabwareState]
 
 
 class CommandLinkNoMeta(BaseModel):

robot-server/robot_server/runs/run_orchestrator_store.py

@@ -295,6 +295,9 @@ async def clear(self) -> RunResult:
             state_summary=run_data, commands=commands, parameters=run_time_parameters
         )
 
+    # todo(mm, 2024-11-15): Are all of these pass-through methods helpful?
+    # Can we delete them and make callers just call .run_orchestrator.play(), etc.?
+
     def play(self, deck_configuration: Optional[DeckConfigurationType] = None) -> None:
         """Start or resume the run."""
         self.run_orchestrator.play(deck_configuration=deck_configuration)
@@ -331,6 +334,10 @@ 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_tip_attached(self) -> Dict[str, bool]:
+        """Get current tip state keyed by pipette id."""
+        return self.run_orchestrator.get_tip_attached()
+
     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()