feat(api): dynamic liquid tracking hw control changes (#17712)

Author: caila-marashaj

api/src/opentrons/hardware_control/motion_utilities.py

@@ -192,6 +192,26 @@ def target_position_from_plunger(
     return all_axes_pos
 
 
+def target_positions_from_plunger_tracking(
+    mount: Union[Mount, OT3Mount],
+    plunger_delta: float,
+    z_delta: float,
+    current_position: Dict[Axis, float],
+) -> "OrderedDict[Axis, float]":
+    """Create a target position for machine axes including plungers for dynamic liquid tracking.
+
+    The x/y axes remain constant but the plunger and Z move to create a tracking action.
+
+    plunger_delta: the distance the plunger should move- should be determined based on desired
+    volume to aspirate/dispense.
+    z_delta: the distance to move the z axis- should be determined based on volume and well geometry.
+    """
+    all_axes_pos = target_position_from_plunger(mount, plunger_delta, current_position)
+    z_ax = Axis.by_mount(mount)
+    all_axes_pos[z_ax] = current_position[z_ax] + z_delta
+    return all_axes_pos
+
+
 def deck_point_from_machine_point(
     machine_point: Point, attitude: AttitudeMatrix, offset: Point
 ) -> Point:

api/src/opentrons/hardware_control/ot3api.py

@@ -127,6 +127,7 @@
     target_position_from_absolute,
     target_position_from_relative,
     target_position_from_plunger,
+    target_positions_from_plunger_tracking,
     offset_for_mount,
     deck_from_machine,
     machine_from_deck,
@@ -2994,6 +2995,102 @@ async def capacitive_sweep(
 
     AMKey = TypeVar("AMKey")
 
+    async def aspirate_while_tracking(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        z_distance: float,
+        volume: float,
+        flow_rate: float = 1.0,
+    ) -> None:
+        """
+        Aspirate a volume of liquid (in microliters/uL) while moving the z axis synchronously.
+
+        :param mount: A robot mount that the instrument is on.
+        :param z_distance: The distance the z axis will move during apsiration.
+        :param volume: The volume of liquid to be aspirated.
+        :param flow_rate: The flow rate to aspirate with.
+        """
+        realmount = OT3Mount.from_mount(mount)
+        aspirate_spec = self._pipette_handler.plan_check_aspirate(
+            realmount, volume, flow_rate
+        )
+        if not aspirate_spec:
+            return
+        target_pos = target_positions_from_plunger_tracking(
+            realmount,
+            aspirate_spec.plunger_distance,
+            z_distance,
+            self._current_position,
+        )
+        try:
+            await self._backend.set_active_current(
+                {aspirate_spec.axis: aspirate_spec.current}
+            )
+            async with self.restore_system_constrants():
+                await self.set_system_constraints_for_plunger_acceleration(
+                    realmount, aspirate_spec.acceleration
+                )
+                await self._move(
+                    target_pos,
+                    speed=aspirate_spec.speed,
+                    home_flagged_axes=False,
+                )
+        except Exception:
+            self._log.exception("Aspirate failed")
+            aspirate_spec.instr.set_current_volume(0)
+            raise
+        else:
+            aspirate_spec.instr.add_current_volume(aspirate_spec.volume)
+
+    async def dispense_while_tracking(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        z_distance: float,
+        volume: float,
+        push_out: Optional[float],
+        flow_rate: float = 1.0,
+    ) -> None:
+        """
+        Dispense a volume of liquid (in microliters/uL) while moving the z axis synchronously.
+
+        :param mount: A robot mount that the instrument is on.
+        :param z_distance: The distance the z axis will move during dispensing.
+        :param volume: The volume of liquid to be dispensed.
+        :param flow_rate: The flow rate to dispense with.
+        """
+        realmount = OT3Mount.from_mount(mount)
+        dispense_spec = self._pipette_handler.plan_check_dispense(
+            realmount, volume, flow_rate, push_out
+        )
+        if not dispense_spec:
+            return
+        target_pos = target_positions_from_plunger_tracking(
+            realmount,
+            dispense_spec.plunger_distance,
+            z_distance,
+            self._current_position,
+        )
+
+        try:
+            await self._backend.set_active_current(
+                {dispense_spec.axis: dispense_spec.current}
+            )
+            async with self.restore_system_constrants():
+                await self.set_system_constraints_for_plunger_acceleration(
+                    realmount, dispense_spec.acceleration
+                )
+                await self._move(
+                    target_pos,
+                    speed=dispense_spec.speed,
+                    home_flagged_axes=False,
+                )
+        except Exception:
+            self._log.exception("dispense failed")
+            dispense_spec.instr.set_current_volume(0)
+            raise
+        else:
+            dispense_spec.instr.remove_current_volume(dispense_spec.volume)
+
     @property
     def attached_subsystems(self) -> Dict[SubSystem, SubSystemState]:
         """Get a view of the state of the currently-attached subsystems."""

api/src/opentrons/hardware_control/protocols/liquid_handler.py

@@ -122,6 +122,23 @@ async def aspirate(
         """
         ...
 
+    async def aspirate_while_tracking(
+        self,
+        mount: MountArgType,
+        z_distance: float,
+        volume: float,
+        flow_rate: float = 1.0,
+    ) -> None:
+        """
+        Aspirate a volume of liquid (in microliters/uL) while moving the z axis synchronously.
+
+        :param mount: A robot mount that the instrument is on.
+        :param z_distance: The distance the z axis will move during apsiration.
+        :param volume: The volume of liquid to be aspirated.
+        :param flow_rate: The flow rate to aspirate with.
+        """
+        ...
+
     async def dispense(
         self,
         mount: MountArgType,
@@ -143,6 +160,24 @@ async def dispense(
         """
         ...
 
+    async def dispense_while_tracking(
+        self,
+        mount: MountArgType,
+        z_distance: float,
+        volume: float,
+        push_out: Optional[float],
+        flow_rate: float = 1.0,
+    ) -> None:
+        """
+        Dispense a volume of liquid (in microliters/uL) while moving the z axis synchronously.
+
+        :param mount: A robot mount that the instrument is on.
+        :param z_distance: The distance the z axis will move during dispensing.
+        :param volume: The volume of liquid to be dispensed.
+        :param flow_rate: The flow rate to dispense with.
+        """
+        ...
+
     async def blow_out(
         self, mount: MountArgType, volume: Optional[float] = None
     ) -> None:

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

@@ -100,7 +100,7 @@ async def liquid_probe_in_place(
 
 
 class HardwarePipettingHandler(PipettingHandler):
-    """Liquid handling, using the Hardware API.""" ""
+    """Liquid handling, using the Hardware API."""
 
     def __init__(self, state_view: StateView, hardware_api: HardwareControlAPI) -> None:
         """Initialize a PipettingHandler instance."""

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -75,6 +75,7 @@
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
     PipetteLiquidNotFoundError,
+    UnexpectedTipRemovalError,
 )
 from opentrons_shared_data.gripper.gripper_definition import GripperModel
 from opentrons_shared_data.pipette.types import (
@@ -1788,6 +1789,106 @@ async def test_plunger_ready_to_aspirate_after_dispense(
     )
 
 
+@pytest.mark.parametrize("is_ready", [True, False])
+@pytest.mark.parametrize("tip_present", [True, False])
+async def test_aspirate_while_tracking(
+    ot3_hardware: ThreadManager[OT3API],
+    mock_move: AsyncMock,
+    is_ready: bool,
+    tip_present: bool,
+) -> None:
+    mount = OT3Mount.LEFT
+
+    instr_data = AttachedPipette(
+        config=load_pipette_data.load_definition(
+            PipetteModelType("p1000"),
+            PipetteChannelType(1),
+            PipetteVersionType(3, 4),
+            PipetteOEMType.OT,
+        ),
+        id="fakepip",
+    )
+    await ot3_hardware.cache_pipette(OT3Mount.LEFT, instr_data, None)
+    pipette = ot3_hardware.hardware_pipettes[OT3Mount.LEFT.to_mount()]
+    assert pipette
+
+    if tip_present:
+        ot3_hardware.add_tip(mount, 100)
+        if is_ready:
+            await ot3_hardware.prepare_for_aspirate(OT3Mount.LEFT)
+
+    if not tip_present:
+        with pytest.raises(UnexpectedTipRemovalError):
+            await ot3_hardware.aspirate_while_tracking(mount, 8.0, 80.0)
+    elif not is_ready:
+        with pytest.raises(RuntimeError):
+            await ot3_hardware.aspirate_while_tracking(mount, 8.0, 80.0)
+    else:
+        await ot3_hardware.aspirate_while_tracking(mount, 8.0, 80.0)
+        # make sure the move planning math stays the same
+        expected_target_pos = {
+            Axis.X: 477.2,
+            Axis.Y: 493.8,
+            Axis.Z_L: 261.475,
+            Axis.P_L: 65.983786,
+        }
+        expected_speed = 46.504982
+        called_target_pos = mock_move.call_args_list[-1][0][0]
+        called_speed = mock_move.call_args_list[-1][1]["speed"]
+        assert expected_target_pos == called_target_pos
+        assert expected_speed == called_speed
+
+
+@pytest.mark.parametrize("tip_present", [True, False])
+@pytest.mark.parametrize("is_ready", [True, False])
+async def test_dispense_while_tracking(
+    ot3_hardware: ThreadManager[OT3API],
+    mock_move: AsyncMock,
+    tip_present: bool,
+    is_ready: bool,
+) -> None:
+    mount = OT3Mount.LEFT
+
+    instr_data = AttachedPipette(
+        config=load_pipette_data.load_definition(
+            PipetteModelType("p1000"),
+            PipetteChannelType(1),
+            PipetteVersionType(3, 4),
+            PipetteOEMType.OT,
+        ),
+        id="fakepip",
+    )
+    await ot3_hardware.cache_pipette(OT3Mount.LEFT, instr_data, None)
+    pipette = ot3_hardware.hardware_pipettes[OT3Mount.LEFT.to_mount()]
+    assert pipette
+
+    if tip_present:
+        ot3_hardware.add_tip(mount, 100)
+    if is_ready:
+        pipette.set_current_volume(80.0)
+
+    if not tip_present:
+        with pytest.raises(UnexpectedTipRemovalError):
+            await ot3_hardware.dispense_while_tracking(mount, 8.0, 80.0, push_out=None)
+    else:
+        await ot3_hardware.dispense_while_tracking(mount, 8.0, 80.0, push_out=None)
+        if is_ready:
+            # make sure the move planning math stays the same
+            expected_target_pos = {
+                Axis.X: 477.2,
+                Axis.Y: 493.8,
+                Axis.Z_L: 261.475,
+                Axis.P_L: 72.75754527162978,
+            }
+            expected_speed = 46.504982
+            called_target_pos = mock_move.call_args_list[-1][0][0]
+            called_speed = mock_move.call_args_list[-1][1]["speed"]
+            assert expected_target_pos == called_target_pos
+            assert expected_speed == called_speed
+        else:
+            assert len(mock_move.call_args_list) == 0
+
+
 async def test_move_to_plunger_bottom(
     ot3_hardware: ThreadManager[OT3API],
     mock_move: AsyncMock,