Also apply validation to the hardware pipette handler.

Author: SyntaxColoring

api/src/opentrons/protocol_engine/execution/pipetting.py

@@ -91,14 +91,19 @@ async def aspirate_in_place(
     ) -> float:
         """Set flow-rate and aspirate."""
         # get mount and config data from state and hardware controller
+        adjusted_volume = _validate_aspirate_volume(
+            state_view=self._state_view, pipette_id=pipette_id, aspirate_volume=volume
+        )
         hw_pipette = self._state_view.pipettes.get_hardware_pipette(
             pipette_id=pipette_id,
             attached_pipettes=self._hardware_api.attached_instruments,
         )
         with self._set_flow_rate(pipette=hw_pipette, aspirate_flow_rate=flow_rate):
-            await self._hardware_api.aspirate(mount=hw_pipette.mount, volume=volume)
+            await self._hardware_api.aspirate(
+                mount=hw_pipette.mount, volume=adjusted_volume
+            )
 
-        return volume
+        return adjusted_volume
 
     async def dispense_in_place(
         self,
@@ -108,6 +113,9 @@ async def dispense_in_place(
         push_out: Optional[float],
     ) -> float:
         """Dispense liquid without moving the pipette."""
+        adjusted_volume = _validate_dispense_volume(
+            state_view=self._state_view, pipette_id=pipette_id, dispense_volume=volume
+        )
         hw_pipette = self._state_view.pipettes.get_hardware_pipette(
             pipette_id=pipette_id,
             attached_pipettes=self._hardware_api.attached_instruments,
@@ -119,10 +127,10 @@ async def dispense_in_place(
             )
         with self._set_flow_rate(pipette=hw_pipette, dispense_flow_rate=flow_rate):
             await self._hardware_api.dispense(
-                mount=hw_pipette.mount, volume=volume, push_out=push_out
+                mount=hw_pipette.mount, volume=adjusted_volume, push_out=push_out
             )
 
-        return volume
+        return adjusted_volume
 
     async def blow_out_in_place(
         self,
@@ -194,8 +202,8 @@ async def aspirate_in_place(
     ) -> float:
         """Virtually aspirate (no-op)."""
         self._validate_tip_attached(pipette_id=pipette_id, command_name="aspirate")
-        return self._validate_aspirate_volume(
-            pipette_id=pipette_id, aspirate_volume=volume
+        return _validate_aspirate_volume(
+            state_view=self._state_view, pipette_id=pipette_id, aspirate_volume=volume
         )
 
     async def dispense_in_place(
@@ -212,8 +220,8 @@ async def dispense_in_place(
                 "push out value cannot have a negative value."
             )
         self._validate_tip_attached(pipette_id=pipette_id, command_name="dispense")
-        return self._validate_dispense_volume(
-            pipette_id=pipette_id, dispense_volume=volume
+        return _validate_dispense_volume(
+            state_view=self._state_view, pipette_id=pipette_id, dispense_volume=volume
         )
 
     async def blow_out_in_place(
@@ -231,57 +239,6 @@ def _validate_tip_attached(self, pipette_id: str, command_name: str) -> None:
                 f"Cannot perform {command_name} without a tip attached"
             )
 
-    def _validate_aspirate_volume(
-        self, pipette_id: str, aspirate_volume: float
-    ) -> float:
-        """Get whether the aspirated volume is valid to aspirate."""
-        working_volume = self._state_view.pipettes.get_working_volume(
-            pipette_id=pipette_id
-        )
-
-        current_volume = (
-            self._state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id) or 0
-        )
-
-        available_volume = working_volume - current_volume
-        available_volume_with_tolerance = (
-            available_volume + _VOLUME_ROUNDING_ERROR_TOLERANCE
-        )
-
-        if aspirate_volume > available_volume_with_tolerance:
-            raise InvalidAspirateVolumeError(
-                attempted_aspirate_volume=aspirate_volume,
-                available_volume=available_volume,
-                max_pipette_volume=self._state_view.pipettes.get_maximum_volume(
-                    pipette_id=pipette_id
-                ),
-                max_tip_volume=self._get_max_tip_volume(pipette_id=pipette_id),
-            )
-        else:
-            return min(aspirate_volume, available_volume)
-
-    def _validate_dispense_volume(
-        self, pipette_id: str, dispense_volume: float
-    ) -> float:
-        """Validate dispense volume."""
-        aspirated_volume = self._state_view.pipettes.get_aspirated_volume(pipette_id)
-        if aspirated_volume is None:
-            raise InvalidDispenseVolumeError(
-                "Cannot perform a dispense if there is no volume in attached tip."
-            )
-        else:
-            remaining = aspirated_volume - dispense_volume
-            if remaining < -_VOLUME_ROUNDING_ERROR_TOLERANCE:
-                raise InvalidDispenseVolumeError(
-                    f"Cannot dispense {dispense_volume} µL when only {aspirated_volume} µL has been aspirated."
-                )
-            else:
-                return min(dispense_volume, aspirated_volume)
-
-    def _get_max_tip_volume(self, pipette_id: str) -> Optional[float]:
-        attached_tip = self._state_view.pipettes.get_attached_tip(pipette_id=pipette_id)
-        return None if attached_tip is None else attached_tip.volume
-
 
 def create_pipetting_handler(
     state_view: StateView, hardware_api: HardwareControlAPI
@@ -292,3 +249,68 @@ def create_pipetting_handler(
         if state_view.config.use_virtual_pipettes is False
         else VirtualPipettingHandler(state_view=state_view)
     )
+
+
+def _validate_aspirate_volume(
+    state_view: StateView, pipette_id: str, aspirate_volume: float
+) -> float:
+    """Get whether the given volume is valid to aspirate right now.
+
+    Return the volume to aspirate, possibly clamped, or raise an
+    InvalidAspirateVolumeError.
+    """
+    working_volume = state_view.pipettes.get_working_volume(pipette_id=pipette_id)
+
+    current_volume = (
+        state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id) or 0
+    )
+
+    # TODO(mm, 2024-01-11): We should probably just use
+    # state_view.pipettes.get_available_volume()? Its whole `None` return vs. exception
+    # raising thing is confusing me.
+    available_volume = working_volume - current_volume
+    available_volume_with_tolerance = (
+        available_volume + _VOLUME_ROUNDING_ERROR_TOLERANCE
+    )
+
+    if aspirate_volume > available_volume_with_tolerance:
+        raise InvalidAspirateVolumeError(
+            attempted_aspirate_volume=aspirate_volume,
+            available_volume=available_volume,
+            max_pipette_volume=state_view.pipettes.get_maximum_volume(
+                pipette_id=pipette_id
+            ),
+            max_tip_volume=_get_max_tip_volume(
+                state_view=state_view, pipette_id=pipette_id
+            ),
+        )
+    else:
+        return min(aspirate_volume, available_volume)
+
+
+def _validate_dispense_volume(
+    state_view: StateView, pipette_id: str, dispense_volume: float
+) -> float:
+    """Get whether the given volume is valid to dispense right now.
+
+    Return the volume to dispense, possibly clamped, or raise an
+    InvalidDispenseVolumeError.
+    """
+    aspirated_volume = state_view.pipettes.get_aspirated_volume(pipette_id)
+    if aspirated_volume is None:
+        raise InvalidDispenseVolumeError(
+            "Cannot perform a dispense if there is no volume in attached tip."
+        )
+    else:
+        remaining = aspirated_volume - dispense_volume
+        if remaining < -_VOLUME_ROUNDING_ERROR_TOLERANCE:
+            raise InvalidDispenseVolumeError(
+                f"Cannot dispense {dispense_volume} µL when only {aspirated_volume} µL has been aspirated."
+            )
+        else:
+            return min(dispense_volume, aspirated_volume)
+
+
+def _get_max_tip_volume(state_view: StateView, pipette_id: str) -> Optional[float]:
+    attached_tip = state_view.pipettes.get_attached_tip(pipette_id=pipette_id)
+    return None if attached_tip is None else attached_tip.volume