rough dynamic liquid tracking

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -24,6 +24,7 @@
     KeysView,
     Union,
     Mapping,
+    Literal,
 )
 from opentrons.config.types import OT3Config, GantryLoad
 from opentrons.config import gripper_config
@@ -167,6 +168,7 @@
     liquid_probe,
     check_overpressure,
     grab_pressure,
+    move_plunger_while_tracking_z,
 )
 from opentrons_hardware.hardware_control.rear_panel_settings import (
     get_door_state,
@@ -720,6 +722,32 @@ def _build_move_gear_axis_runner(
             True,
         )
 
+    @requires_update
+    @requires_estop
+    async def aspirate_while_tracking(
+        self,
+        mount: OT3Mount,
+        distance: float,
+        speed: float,
+        direction: Union[Literal[1], Literal[-1]],
+        duration: float,
+    ) -> None:
+        head_node = axis_to_node(Axis.by_mount(mount))
+        tool = sensor_node_for_pipette(OT3Mount(mount.value))
+        async with self._monitor_overpressure([tool]):
+            positions = await move_plunger_while_tracking_z(
+                messenger=self._messenger,
+                tool=tool,
+                head_node=head_node,
+                distance=distance,
+                speed=speed,
+                direction=direction,
+                duration=duration,
+            )
+        for node, point in positions.items():
+            self._position.update({node: point.motor_position})
+            self._encoder_position.update({node: point.encoder_position})
+
     @requires_update
     @requires_estop
     async def move(

api/src/opentrons/hardware_control/ot3api.py

@@ -2748,7 +2748,7 @@ async def liquid_probe(  # noqa: C901
         if not probe_settings:
             probe_settings = deepcopy(self.config.liquid_sense)
 
-        # We need to significatly slow down the 96 channel liquid probe
+        # We need to significantly slow down the 96 channel liquid probe
         if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
             max_plunger_speed = self.config.motion_settings.max_speed_discontinuity[
                 GantryLoad.HIGH_THROUGHPUT
@@ -2961,6 +2961,45 @@ async def capacitive_sweep(
 
     AMKey = TypeVar("AMKey")
 
+    async def aspirate_while_tracking(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        distance: float,
+        rate: float,
+        volume: Optional[float] = None,
+    ) -> None:
+        """
+        Aspirate a volume of liquid (in microliters/uL) using this pipette."""
+        realmount = OT3Mount.from_mount(mount)
+        aspirate_spec = self._pipette_handler.plan_check_aspirate(
+            realmount, volume, rate
+        )
+        if not aspirate_spec:
+            return
+
+        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._backend.aspirate_while_tracking(
+                    mount=mount,
+                    distance=distance,
+                    speed=rate,
+                    direction=-1,
+                    # have to actually determine duration here
+                    duration=0.0,
+                )
+        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)
+
     @property
     def attached_subsystems(self) -> Dict[SubSystem, SubSystemState]:
         """Get a view of the state of the currently-attached subsystems."""

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -135,6 +135,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = False,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -192,6 +193,7 @@ def aspirate(
                     wellLocation=well_location,
                     volume=volume,
                     flowRate=flow_rate,
+                    is_tracking=is_tracking if is_tracking else False,
                 )
             )
 

api/src/opentrons/protocol_api/core/instrument.py

@@ -42,6 +42,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = False,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -85,6 +85,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = None,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -96,6 +96,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = None,
     ) -> None:
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any

api/src/opentrons/protocol_api/instrument_context.py

@@ -292,6 +292,137 @@ def aspirate(
 
         return self
 
+    ########
+
+    @requires_version(2, 0)
+    def aspirate_while_tracking(
+        self,
+        volume: Optional[float] = None,
+        location: Optional[Union[types.Location, labware.Well]] = None,
+        rate: float = 1.0,
+    ) -> InstrumentContext:
+        """
+        Draw liquid into a pipette tip.
+
+        See :ref:`new-aspirate` for more details and examples.
+
+        :param volume: The volume to aspirate, measured in µL. If unspecified,
+                    defaults to the maximum volume for the pipette and its currently
+                    attached tip.
+
+                    If ``aspirate`` is called with a volume of precisely 0, its behavior
+                    depends on the API level of the protocol. On API levels below 2.16,
+                    it will behave the same as a volume of ``None``/unspecified: aspirate
+                    until the pipette is full. On API levels at or above 2.16, no liquid
+                    will be aspirated.
+        :type volume: int or float
+        :param location: Tells the robot where to aspirate from. The location can be
+                         a :py:class:`.Well` or a :py:class:`.Location`.
+
+                            - If the location is a ``Well``, the robot will aspirate at
+                              or above the bottom center of the well. The distance (in mm)
+                              from the well bottom is specified by
+                              :py:obj:`well_bottom_clearance.aspirate
+                              <well_bottom_clearance>`.
+
+                            - If the location is a ``Location`` (e.g., the result of
+                              :py:meth:`.Well.top` or :py:meth:`.Well.bottom`), the robot
+                              will aspirate from that specified position.
+
+                            - If the ``location`` is unspecified, the robot will
+                              aspirate from its current position.
+        :param rate: A multiplier for the default flow rate of the pipette. Calculated
+                     as ``rate`` multiplied by :py:attr:`flow_rate.aspirate
+                     <flow_rate>`. If not specified, defaults to 1.0. See
+                     :ref:`new-plunger-flow-rates`.
+        :type rate: float
+        :returns: This instance.
+
+        .. note::
+
+            If ``aspirate`` is called with a single, unnamed argument, it will treat
+            that argument as ``volume``. If you want to call ``aspirate`` with only
+            ``location``, specify it as a keyword argument:
+            ``pipette.aspirate(location=plate['A1'])``
+
+        """
+        _log.debug(
+            "aspirate {} from {} at {}".format(
+                volume, location if location else "current position", rate
+            )
+        )
+
+        move_to_location: types.Location
+        well: Optional[labware.Well] = None
+        is_meniscus: Optional[bool] = None
+        last_location = self._get_last_location_by_api_version()
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
+            )
+        except validation.NoLocationError as e:
+            raise RuntimeError(
+                "If aspirate is called without an explicit location, another"
+                " method that moves to a location (such as move_to or "
+                "dispense) must previously have been called so the robot "
+                "knows where it is."
+            ) from e
+
+        if isinstance(target, (TrashBin, WasteChute)):
+            raise ValueError(
+                "Trash Bin and Waste Chute are not acceptable location parameters for Aspirate commands."
+            )
+        move_to_location, well, is_meniscus = self._handle_aspirate_target(
+            target=target
+        )
+        if self.api_version >= APIVersion(2, 11):
+            instrument.validate_takes_liquid(
+                location=move_to_location,
+                reject_module=self.api_version >= APIVersion(2, 13),
+                reject_adapter=self.api_version >= APIVersion(2, 15),
+            )
+
+        if self.api_version >= APIVersion(2, 16):
+            c_vol = self._core.get_available_volume() if volume is None else volume
+        else:
+            c_vol = self._core.get_available_volume() if not volume else volume
+        flow_rate = self._core.get_aspirate_flow_rate(rate)
+
+        if (
+            self.api_version >= APIVersion(2, 20)
+            and well is not None
+            and self.liquid_presence_detection
+            and self._core.nozzle_configuration_valid_for_lld()
+            and self._core.get_current_volume() == 0
+        ):
+            self._raise_if_pressure_not_supported_by_pipette()
+            self.require_liquid_presence(well=well)
+
+        with publisher.publish_context(
+            broker=self.broker,
+            command=cmds.aspirate(
+                instrument=self,
+                volume=c_vol,
+                location=move_to_location,
+                flow_rate=flow_rate,
+                rate=rate,
+            ),
+        ):
+            self._core.aspirate(
+                location=move_to_location,
+                well_core=well._core if well is not None else None,
+                volume=c_vol,
+                rate=rate,
+                flow_rate=flow_rate,
+                in_place=target.in_place,
+                is_meniscus=is_meniscus,
+                is_tracking=True,
+            )
+
+        return self
+
+    ########
+
     @requires_version(2, 0)
     def dispense(  # noqa: C901
         self,

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

@@ -11,7 +11,9 @@
     FlowRateMixin,
     BaseLiquidHandlingResult,
     aspirate_in_place,
+    aspirate_while_tracking,
     prepare_for_aspirate,
+    IsTrackingMixin,
 )
 from .movement_common import (
     LiquidHandlingWellLocationMixin,
@@ -47,7 +49,11 @@
 
 
 class AspirateParams(
-    PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
+    PipetteIdMixin,
+    AspirateVolumeMixin,
+    FlowRateMixin,
+    LiquidHandlingWellLocationMixin,
+    IsTrackingMixin,
 ):
     """Parameters required to aspirate from a specific well."""
 
@@ -158,28 +164,46 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
             well_location=well_location,
             current_well=current_well,
             operation_volume=-params.volume,
+            is_tracking=params.is_tracking,
         )
         state_update.append(move_result.state_update)
         if isinstance(move_result, DefinedErrorData):
             return DefinedErrorData(
                 public=move_result.public, state_update=state_update
             )
 
-        aspirate_result = await aspirate_in_place(
-            pipette_id=pipette_id,
-            volume=params.volume,
-            flow_rate=params.flowRate,
-            location_if_error={
-                "retryLocation": (
-                    move_result.public.position.x,
-                    move_result.public.position.y,
-                    move_result.public.position.z,
-                )
-            },
-            command_note_adder=self._command_note_adder,
-            pipetting=self._pipetting,
-            model_utils=self._model_utils,
-        )
+        if params.is_tracking:
+            aspirate_result = await aspirate_while_tracking(
+                pipette_id=pipette_id,
+                volume=params.volume,
+                flow_rate=params.flowRate,
+                location_if_error={
+                    "retryLocation": (
+                        move_result.public.position.x,
+                        move_result.public.position.y,
+                        move_result.public.position.z,
+                    )
+                },
+                command_note_adder=self._command_note_adder,
+                pipetting=self._pipetting,
+                model_utils=self._model_utils,
+            )
+        else:
+            aspirate_result = await aspirate_in_place(
+                pipette_id=pipette_id,
+                volume=params.volume,
+                flow_rate=params.flowRate,
+                location_if_error={
+                    "retryLocation": (
+                        move_result.public.position.x,
+                        move_result.public.position.y,
+                        move_result.public.position.z,
+                    )
+                },
+                command_note_adder=self._command_note_adder,
+                pipetting=self._pipetting,
+                model_utils=self._model_utils,
+            )
         state_update.append(aspirate_result.state_update)
         if isinstance(aspirate_result, DefinedErrorData):
             state_update.set_liquid_operated(

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

@@ -145,6 +145,7 @@ async def move_to_well(
     minimum_z_height: Optional[float] = None,
     speed: Optional[float] = None,
     operation_volume: Optional[float] = None,
+    is_tracking: Optional[bool] = False,
 ) -> MoveToWellOperationReturn:
     """Execute a move to well microoperation."""
     try:
@@ -158,6 +159,7 @@ async def move_to_well(
             minimum_z_height=minimum_z_height,
             speed=speed,
             operation_volume=operation_volume,
+            is_tracking=is_tracking,
         )
     except StallOrCollisionDetectedError as e:
         return DefinedErrorData(