updated WellView get methods, fixed LiquidProbe to prevent errors, ad…

…ded tests for IncompleteLabwareDefinitionError

api/src/opentrons/protocol_engine/commands/liquid_probe.py

@@ -11,6 +11,7 @@
     MustHomeError,
     PipetteNotReadyToAspirateError,
     TipNotEmptyError,
+    IncompleteLabwareDefinitionError,
 )
 from opentrons.types import MountType
 from opentrons_shared_data.errors.exceptions import (
@@ -227,11 +228,14 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
                 state_update=state_update,
             )
         else:
-            well_volume = self._state_view.geometry.get_well_volume_at_height(
-                labware_id=params.labwareId,
-                well_name=params.wellName,
-                height=z_pos_or_error,
-            )
+            try:
+                well_volume = self._state_view.geometry.get_well_volume_at_height(
+                    labware_id=params.labwareId,
+                    well_name=params.wellName,
+                    height=z_pos_or_error,
+                )
+            except IncompleteLabwareDefinitionError:
+                well_volume = None
             state_update.set_liquid_probed(
                 labware_id=params.labwareId,
                 well_name=params.wellName,
@@ -282,9 +286,12 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
             well_volume = None
         else:
             z_pos = z_pos_or_error
-            well_volume = self._state_view.geometry.get_well_volume_at_height(
-                labware_id=params.labwareId, well_name=params.wellName, height=z_pos
-            )
+            try:
+                well_volume = self._state_view.geometry.get_well_volume_at_height(
+                    labware_id=params.labwareId, well_name=params.wellName, height=z_pos
+                )
+            except IncompleteLabwareDefinitionError:
+                well_volume = None
 
         state_update.set_liquid_probed(
             labware_id=params.labwareId,

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

@@ -1371,7 +1371,7 @@ def get_well_offset_adjustment(
 
         Distance is with reference to the well bottom.
         """
-        # refactor? Consider how many conversions are being done
+        # TODO(pbm, 10-23-24): refactor to smartly reduce height/volume conversions
         initial_handling_height = self.get_well_handling_height(
             labware_id=labware_id,
             well_name=well_name,
@@ -1402,25 +1402,23 @@ def get_meniscus_height(
     ) -> float:
         """Returns stored meniscus height in specified well."""
         (
-            loaded_volume,
-            probed_height,
-            probed_volume,
-        ) = self._wells.get_well_liquid_values(
-            labware_id=labware_id, well_name=well_name
-        )
-        if probed_height:
-            return probed_height
-        elif loaded_volume:
+            loaded_volume_info,
+            probed_height_info,
+            probed_volume_info,
+        ) = self._wells.get_well_liquid_info(labware_id=labware_id, well_name=well_name)
+        if probed_height_info is not None and probed_height_info.height is not None:
+            return probed_height_info.height
+        elif loaded_volume_info is not None and loaded_volume_info.volume is not None:
             return self.get_well_height_at_volume(
                 labware_id=labware_id,
                 well_name=well_name,
-                volume=loaded_volume,
+                volume=loaded_volume_info.volume,
             )
-        elif probed_volume:
+        elif probed_volume_info is not None and probed_volume_info.volume is not None:
             return self.get_well_height_at_volume(
                 labware_id=labware_id,
                 well_name=well_name,
-                volume=probed_volume,
+                volume=probed_volume_info.volume,
             )
         else:
             raise errors.LiquidHeightUnknownError(

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

@@ -1,17 +1,19 @@
 """Public protocol run data models."""
 from pydantic import BaseModel, Field
-from typing import List, Optional
+from typing import List, Optional, Union
 from datetime import datetime
 
 from ..errors import ErrorOccurrence
 from ..types import (
     EngineStatus,
-    LiquidHeightSummary,
     LoadedLabware,
     LabwareOffset,
     LoadedModule,
     LoadedPipette,
     Liquid,
+    LoadedVolumeSummary,
+    ProbedHeightSummary,
+    ProbedVolumeSummary,
 )
 
 
@@ -30,5 +32,7 @@ class StateSummary(BaseModel):
     startedAt: Optional[datetime]
     completedAt: Optional[datetime]
     liquids: List[Liquid] = Field(default_factory=list)
-    wells: List[LiquidHeightSummary] = Field(default_factory=list)
+    wells: List[
+        Union[LoadedVolumeSummary, ProbedHeightSummary, ProbedVolumeSummary]
+    ] = Field(default_factory=list)
     files: List[str] = Field(default_factory=list)

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

@@ -1,12 +1,14 @@
 """Basic well data state and store."""
 from dataclasses import dataclass
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
 
 from opentrons.protocol_engine.types import (
     ProbedHeightInfo,
     ProbedVolumeInfo,
     LoadedVolumeInfo,
-    LiquidHeightSummary,
+    ProbedHeightSummary,
+    ProbedVolumeSummary,
+    LoadedVolumeSummary,
 )
 
 from . import update_types
@@ -36,11 +38,11 @@ def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
         state_update = get_state_update(action)
         if state_update is not None:
-            self._handle_loaded_liquid_update(state_update)
-            self._handle_probed_liquid_update(state_update)
-            self._handle_operated_liquid_update(state_update)
+            self._handle_liquid_loaded_update(state_update)
+            self._handle_liquid_probed_update(state_update)
+            self._handle_liquid_operated_update(state_update)
 
-    def _handle_loaded_liquid_update(
+    def _handle_liquid_loaded_update(
         self, state_update: update_types.StateUpdate
     ) -> None:
         if state_update.liquid_loaded != update_types.NO_CHANGE:
@@ -54,7 +56,7 @@ def _handle_loaded_liquid_update(
                     operations_since_load=0,
                 )
 
-    def _handle_probed_liquid_update(
+    def _handle_liquid_probed_update(
         self, state_update: update_types.StateUpdate
     ) -> None:
         if state_update.liquid_probed != update_types.NO_CHANGE:
@@ -74,7 +76,7 @@ def _handle_probed_liquid_update(
                 operations_since_probe=0,
             )
 
-    def _handle_operated_liquid_update(
+    def _handle_liquid_operated_update(
         self, state_update: update_types.StateUpdate
     ) -> None:
         if state_update.liquid_operated != update_types.NO_CHANGE:
@@ -130,60 +132,71 @@ def __init__(self, state: WellState) -> None:
     # if height requested, probed_heights or loaded_vols_to_height or probed_vols_to_height
     # to get height, call GeometryView.get_well_height, which does conversion if needed
 
-    # update this
-    def get_all(self) -> List[LiquidHeightSummary]:
-        """Get all well liquid heights."""
-        all_heights: List[LiquidHeightSummary] = []
-        for labware, wells in self._state.probed_heights.items():
-            for well, lhi in wells.items():
-                lhs = LiquidHeightSummary(
-                    labware_id=labware,
-                    well_name=well,
-                    height=lhi.height,
-                    last_measured=lhi.last_probed,
+    def get_all(
+        self,
+    ) -> List[Union[ProbedHeightSummary, ProbedVolumeSummary, LoadedVolumeSummary]]:
+        """Get all well liquid info summaries."""
+        all_summaries: List[
+            Union[ProbedHeightSummary, ProbedVolumeSummary, LoadedVolumeSummary]
+        ] = []
+        for lv_labware, lv_wells in self._state.loaded_volumes.items():
+            for lv_well, lvi in lv_wells.items():
+                lvs = LoadedVolumeSummary(
+                    labware_id=lv_labware,
+                    well_name=lv_well,
+                    volume=lvi.volume,
+                    last_loaded=lvi.last_loaded,
+                    operations_since_load=lvi.operations_since_load,
                 )
-            all_heights.append(lhs)
-        return all_heights
-
-    # update this
-    def get_all_in_labware(self, labware_id: str) -> List[LiquidHeightSummary]:
-        """Get all well liquid heights for a particular labware."""
-        all_heights: List[LiquidHeightSummary] = []
-        for well, lhi in self._state.probed_heights[labware_id].items():
-            lhs = LiquidHeightSummary(
-                labware_id=labware_id,
-                well_name=well,
-                height=lhi.height,
-                last_measured=lhi.last_probed,
-            )
-            all_heights.append(lhs)
-        return all_heights
-
-    def get_last_measured_liquid_height(
-        self, labware_id: str, well_name: str
-    ) -> Optional[float]:
-        """Returns the height of the liquid according to the most recent liquid level probe to this well.
+                all_summaries.append(lvs)
+        for ph_labware, ph_wells in self._state.probed_heights.items():
+            for ph_well, phi in ph_wells.items():
+                phs = ProbedHeightSummary(
+                    labware_id=ph_labware,
+                    well_name=ph_well,
+                    height=phi.height,
+                    last_probed=phi.last_probed,
+                )
+                all_summaries.append(phs)
+        for pv_labware, pv_wells in self._state.probed_volumes.items():
+            for pv_well, pvi in pv_wells.items():
+                pvs = ProbedVolumeSummary(
+                    labware_id=pv_labware,
+                    well_name=pv_well,
+                    volume=pvi.volume,
+                    last_probed=pvi.last_probed,
+                    operations_since_probe=pvi.operations_since_probe,
+                )
+                all_summaries.append(pvs)
+        return all_summaries
 
-        Returns None if no liquid probe has been done.
-        """
-        try:
-            height = self._state.probed_heights[labware_id][well_name].height
-            return height
-        except KeyError:
-            return None
-
-    def has_measured_liquid_height(self, labware_id: str, well_name: str) -> bool:
-        """Returns True if the well has been liquid level probed previously."""
-        try:
-            return bool(self._state.probed_heights[labware_id][well_name].height)
-        except KeyError:
-            return False
-
-    def get_well_liquid_values(
+    def get_well_liquid_info(
         self, labware_id: str, well_name: str
-    ) -> Tuple[Optional[float], Optional[float], Optional[float]]:
-        """Return all the liquid values for a well."""
-        loaded_volume = self._state.loaded_volumes[labware_id][well_name].volume
-        probed_height = self._state.probed_heights[labware_id][well_name].height
-        probed_volume = self._state.probed_volumes[labware_id][well_name].volume
-        return loaded_volume, probed_height, probed_volume
+    ) -> Tuple[
+        Optional[LoadedVolumeInfo],
+        Optional[ProbedHeightInfo],
+        Optional[ProbedVolumeInfo],
+    ]:
+        """Return all the liquid info for a well."""
+        if (
+            labware_id not in self._state.loaded_volumes
+            or well_name not in self._state.loaded_volumes[labware_id]
+        ):
+            loaded_volume_info = None
+        else:
+            loaded_volume_info = self._state.loaded_volumes[labware_id][well_name]
+        if (
+            labware_id not in self._state.probed_heights
+            or well_name not in self._state.probed_heights[labware_id]
+        ):
+            probed_height_info = None
+        else:
+            probed_height_info = self._state.probed_heights[labware_id][well_name]
+        if (
+            labware_id not in self._state.probed_volumes
+            or well_name not in self._state.probed_volumes[labware_id]
+        ):
+            probed_volume_info = None
+        else:
+            probed_volume_info = self._state.probed_volumes[labware_id][well_name]
+        return loaded_volume_info, probed_height_info, probed_volume_info

api/src/opentrons/protocol_engine/types.py

@@ -355,6 +355,14 @@ class CurrentWell:
     well_name: str
 
 
+class LoadedVolumeInfo(BaseModel):
+    """A well's liquid volume, initialized by a LoadLiquid, updated by Aspirate and Dispense."""
+
+    volume: Optional[float] = None
+    last_loaded: datetime
+    operations_since_load: int
+
+
 class ProbedHeightInfo(BaseModel):
     """A well's liquid height, initialized by a LiquidProbe, cleared by Aspirate and Dispense."""
 
@@ -370,21 +378,25 @@ class ProbedVolumeInfo(BaseModel):
     operations_since_probe: int
 
 
-class LoadedVolumeInfo(BaseModel):
-    """A well's liquid volume, initialized by a LoadLiquid, updated by Aspirate and Dispense."""
+class LoadedVolumeSummary(LoadedVolumeInfo):
+    """Payload for a well's loaded volume in StateSummary."""
 
-    volume: Optional[float] = None
-    last_loaded: datetime
-    operations_since_load: int
+    labware_id: str
+    well_name: str
 
 
-class LiquidHeightSummary(BaseModel):
-    """Payload for liquid state height in StateSummary."""
+class ProbedHeightSummary(ProbedHeightInfo):
+    """Payload for a well's probed height in StateSummary."""
+
+    labware_id: str
+    well_name: str
+
+
+class ProbedVolumeSummary(ProbedVolumeInfo):
+    """Payload for a well's probed volume in StateSummary."""
 
     labware_id: str
     well_name: str
-    height: Optional[float] = None
-    last_measured: datetime
 
 
 @dataclass(frozen=True)