lint fix

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

@@ -181,19 +181,15 @@ def aspirate(
             well_name = well_core.get_name()
             labware_id = well_core.labware_id
 
-            well_location = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
+            (
+                well_location,
+                dynamic_liquid_tracking,
+            ) = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
                 labware_id=labware_id,
                 well_name=well_name,
                 absolute_point=location.point,
                 meniscus_tracking=meniscus_tracking,
             )
-            dynamic_liquid_tracking = False
-            # caila bookmark
-            if meniscus_tracking:
-                if meniscus_tracking.target == "end":
-                    well_location.volumeOffset = "operationVolume"
-                elif meniscus_tracking.target == "dynamic_meniscus":
-                    dynamic_liquid_tracking = True
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
                 pipette_id=self._pipette_id,
@@ -293,18 +289,16 @@ def dispense(
             well_name = well_core.get_name()
             labware_id = well_core.labware_id
 
-            well_location = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
+            (
+                well_location,
+                dynamic_liquid_tracking,
+            ) = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
                 labware_id=labware_id,
                 well_name=well_name,
                 absolute_point=location.point,
                 meniscus_tracking=meniscus_tracking,
             )
-            dynamic_liquid_tracking = False
-            if meniscus_tracking:
-                if meniscus_tracking.target == "end":
-                    well_location.volumeOffset = "operationVolume"
-                elif meniscus_tracking.target == "dynamic_meniscus":
-                    dynamic_liquid_tracking = True
+
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
                 pipette_id=self._pipette_id,

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

@@ -168,7 +168,6 @@ def aspirate_and_wait(self, volume: float) -> None:
             rate=1,
             flow_rate=aspirate_props.flow_rate_by_volume.get_for_volume(volume),
             in_place=True,
-            is_meniscus=None,  # TODO: update this once meniscus is implemented
         )
         self._tip_state.append_liquid(volume)
         delay_props = aspirate_props.delay
@@ -191,7 +190,6 @@ def dispense_and_wait(
             flow_rate=dispense_props.flow_rate_by_volume.get_for_volume(volume),
             in_place=True,
             push_out=push_out_override,
-            is_meniscus=None,
         )
         if push_out_override:
             # If a push out was performed, we need to reset the plunger before we can aspirate again
@@ -536,7 +534,6 @@ def _remove_air_gap(self, location: Location) -> None:
             rate=1,
             flow_rate=flow_rate,
             in_place=True,
-            is_meniscus=None,
             push_out=0,
         )
         self._tip_state.delete_air_gap(last_air_gap)

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

@@ -98,6 +98,7 @@ def aspirate(
         rate: float,
         flow_rate: float,
         in_place: bool,
+        meniscus_tracking: Optional[types.MeniscusTracking] = None,
     ) -> None:
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any
@@ -139,6 +140,7 @@ def dispense(
         flow_rate: float,
         in_place: bool,
         push_out: Optional[float],
+        meniscus_tracking: Optional[types.MeniscusTracking] = None,
     ) -> None:
         if isinstance(location, (TrashBin, WasteChute)):
             raise APIVersionError(

api/src/opentrons/protocol_api/instrument_context.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 import logging
 from contextlib import ExitStack
-from typing import Any, List, Optional, Sequence, Union, cast, Dict, Tuple, Literal
+from typing import Any, List, Optional, Sequence, Union, cast, Dict
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
@@ -550,7 +550,86 @@ def dispense_while_tracking(
         return self
 
     @requires_version(2, 0)
-    def dispense(  # noqa: C901
+    def mix(
+        self,
+        repetitions: int = 1,
+        volume: Optional[float] = None,
+        location: Optional[Union[types.Location, labware.Well]] = None,
+        rate: float = 1.0,
+    ) -> InstrumentContext:
+        """
+        Mix a volume of liquid by repeatedly aspirating and dispensing it in a single location.
+
+        See :ref:`mix` for examples.
+
+        :param repetitions: Number of times to mix (default is 1).
+        :param volume: The volume to mix, measured in µL. If unspecified, defaults
+                       to the maximum volume for the pipette and its attached tip.
+
+                       If ``mix`` 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: mix
+                       the full working volume of the pipette. On API levels at or above 2.16,
+                       no liquid will be mixed.
+        :param location: The :py:class:`.Well` or :py:class:`~.types.Location` where the
+                        pipette will mix. If unspecified, the pipette will mix at its
+                        current position.
+        :param rate: How quickly the pipette aspirates and dispenses liquid while
+                     mixing. The aspiration flow rate is calculated as ``rate``
+                     multiplied by :py:attr:`flow_rate.aspirate <flow_rate>`. The
+                     dispensing flow rate is calculated as ``rate`` multiplied by
+                     :py:attr:`flow_rate.dispense <flow_rate>`. See
+                     :ref:`new-plunger-flow-rates`.
+        :raises: ``UnexpectedTipRemovalError`` -- If no tip is attached to the pipette.
+        :returns: This instance.
+
+        .. note::
+
+            All the arguments of ``mix`` are optional. However, if you omit one of them,
+            all subsequent arguments must be passed as keyword arguments. For instance,
+            ``pipette.mix(1, location=wellplate['A1'])`` is a valid call, but
+            ``pipette.mix(1, wellplate['A1'])`` is not.
+
+        .. versionchanged:: 2.21
+            Does not repeatedly check for liquid presence.
+        """
+        _log.debug(
+            "mixing {}uL with {} repetitions in {} at rate={}".format(
+                volume, repetitions, location if location else "current position", rate
+            )
+        )
+        if not self._core.has_tip():
+            raise UnexpectedTipRemovalError("mix", self.name, self.mount)
+
+        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
+
+        dispense_kwargs: Dict[str, Any] = {}
+        if self.api_version >= APIVersion(2, 16):
+            dispense_kwargs["push_out"] = 0.0
+
+        with publisher.publish_context(
+            broker=self.broker,
+            command=cmds.mix(
+                instrument=self,
+                repetitions=repetitions,
+                volume=c_vol,
+                location=location,
+            ),
+        ):
+            self.aspirate(volume, location, rate)
+            with AutoProbeDisable(self):
+                while repetitions - 1 > 0:
+                    self.dispense(volume, rate=rate, **dispense_kwargs)
+                    self.aspirate(volume, rate=rate)
+                    repetitions -= 1
+                self.dispense(volume, rate=rate)
+        return self
+
+    @requires_version(2, 0)
+    def dispense(
         self,
         volume: Optional[float] = None,
         location: Optional[
@@ -727,85 +806,6 @@ def dispense(  # noqa: C901
 
         return self
 
-    @requires_version(2, 0)
-    def mix(
-        self,
-        repetitions: int = 1,
-        volume: Optional[float] = None,
-        location: Optional[Union[types.Location, labware.Well]] = None,
-        rate: float = 1.0,
-    ) -> InstrumentContext:
-        """
-        Mix a volume of liquid by repeatedly aspirating and dispensing it in a single location.
-
-        See :ref:`mix` for examples.
-
-        :param repetitions: Number of times to mix (default is 1).
-        :param volume: The volume to mix, measured in µL. If unspecified, defaults
-                       to the maximum volume for the pipette and its attached tip.
-
-                       If ``mix`` 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: mix
-                       the full working volume of the pipette. On API levels at or above 2.16,
-                       no liquid will be mixed.
-        :param location: The :py:class:`.Well` or :py:class:`~.types.Location` where the
-                        pipette will mix. If unspecified, the pipette will mix at its
-                        current position.
-        :param rate: How quickly the pipette aspirates and dispenses liquid while
-                     mixing. The aspiration flow rate is calculated as ``rate``
-                     multiplied by :py:attr:`flow_rate.aspirate <flow_rate>`. The
-                     dispensing flow rate is calculated as ``rate`` multiplied by
-                     :py:attr:`flow_rate.dispense <flow_rate>`. See
-                     :ref:`new-plunger-flow-rates`.
-        :raises: ``UnexpectedTipRemovalError`` -- If no tip is attached to the pipette.
-        :returns: This instance.
-
-        .. note::
-
-            All the arguments of ``mix`` are optional. However, if you omit one of them,
-            all subsequent arguments must be passed as keyword arguments. For instance,
-            ``pipette.mix(1, location=wellplate['A1'])`` is a valid call, but
-            ``pipette.mix(1, wellplate['A1'])`` is not.
-
-        .. versionchanged:: 2.21
-            Does not repeatedly check for liquid presence.
-        """
-        _log.debug(
-            "mixing {}uL with {} repetitions in {} at rate={}".format(
-                volume, repetitions, location if location else "current position", rate
-            )
-        )
-        if not self._core.has_tip():
-            raise UnexpectedTipRemovalError("mix", self.name, self.mount)
-
-        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
-
-        dispense_kwargs: Dict[str, Any] = {}
-        if self.api_version >= APIVersion(2, 16):
-            dispense_kwargs["push_out"] = 0.0
-
-        with publisher.publish_context(
-            broker=self.broker,
-            command=cmds.mix(
-                instrument=self,
-                repetitions=repetitions,
-                volume=c_vol,
-                location=location,
-            ),
-        ):
-            self.aspirate(volume, location, rate)
-            with AutoProbeDisable(self):
-                while repetitions - 1 > 0:
-                    self.dispense(volume, rate=rate, **dispense_kwargs)
-                    self.aspirate(volume, rate=rate)
-                    repetitions -= 1
-                self.dispense(volume, rate=rate)
-        return self
-
     @requires_version(2, 0)
     def blow_out(
         self,

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

@@ -47,7 +47,10 @@
 
 
 class AspirateParams(
-    PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
+    PipetteIdMixin,
+    AspirateVolumeMixin,
+    FlowRateMixin,
+    LiquidHandlingWellLocationMixin,
 ):
     """Parameters required to aspirate from a specific well."""
 

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

@@ -217,6 +217,104 @@ async def aspirate_in_place(
         )
 
 
+async def aspirate_while_tracking(
+    pipette_id: str,
+    labware_id: str,
+    well_name: str,
+    volume: float,
+    flow_rate: float,
+    location_if_error: ErrorLocationInfo,
+    command_note_adder: CommandNoteAdder,
+    pipetting: PipettingHandler,
+    model_utils: ModelUtils,
+) -> SuccessData[BaseLiquidHandlingResult] | DefinedErrorData[OverpressureError]:
+    """Execute an aspirate while tracking microoperation."""
+    try:
+        volume_aspirated = await pipetting.aspirate_while_tracking(
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            volume=volume,
+            flow_rate=flow_rate,
+            command_note_adder=command_note_adder,
+        )
+    except PipetteOverpressureError as e:
+        return DefinedErrorData(
+            public=OverpressureError(
+                id=model_utils.generate_id(),
+                createdAt=model_utils.get_timestamp(),
+                wrappedErrors=[
+                    ErrorOccurrence.from_failed(
+                        id=model_utils.generate_id(),
+                        createdAt=model_utils.get_timestamp(),
+                        error=e,
+                    )
+                ],
+                errorInfo=location_if_error,
+            ),
+            state_update=StateUpdate().set_fluid_unknown(pipette_id=pipette_id),
+        )
+    else:
+        return SuccessData(
+            public=BaseLiquidHandlingResult(
+                volume=volume_aspirated,
+            ),
+            state_update=StateUpdate().set_fluid_aspirated(
+                pipette_id=pipette_id,
+                fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume_aspirated),
+            ),
+        )
+
+
+async def dispense_while_tracking(
+    pipette_id: str,
+    labware_id: str,
+    well_name: str,
+    volume: float,
+    flow_rate: float,
+    push_out: float | None,
+    location_if_error: ErrorLocationInfo,
+    pipetting: PipettingHandler,
+    model_utils: ModelUtils,
+) -> SuccessData[BaseLiquidHandlingResult] | DefinedErrorData[OverpressureError]:
+    """Execute an dispense while tracking microoperation."""
+    try:
+        volume_dispensed = await pipetting.dispense_while_tracking(
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            volume=volume,
+            flow_rate=flow_rate,
+            push_out=push_out,
+        )
+    except PipetteOverpressureError as e:
+        return DefinedErrorData(
+            public=OverpressureError(
+                id=model_utils.generate_id(),
+                createdAt=model_utils.get_timestamp(),
+                wrappedErrors=[
+                    ErrorOccurrence.from_failed(
+                        id=model_utils.generate_id(),
+                        createdAt=model_utils.get_timestamp(),
+                        error=e,
+                    )
+                ],
+                errorInfo=location_if_error,
+            ),
+            state_update=StateUpdate().set_fluid_unknown(pipette_id=pipette_id),
+        )
+    else:
+        return SuccessData(
+            public=BaseLiquidHandlingResult(
+                volume=volume_dispensed,
+            ),
+            state_update=StateUpdate().set_fluid_ejected(
+                pipette_id=pipette_id,
+                volume=volume_dispensed,
+            ),
+        )
+
+
 async def dispense_in_place(
     pipette_id: str,
     volume: float,