fix(api): Lld movement adjustments

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

@@ -146,6 +146,7 @@ async def liquid_probe(
         mount_speed: float,
         plunger_speed: float,
         threshold_pascals: float,
+        plunger_impulse_time: float,
         output_format: OutputOptions = OutputOptions.can_bus_only,
         data_files: Optional[Dict[InstrumentProbeType, str]] = None,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,

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

@@ -1357,6 +1357,7 @@ async def liquid_probe(
         mount_speed: float,
         plunger_speed: float,
         threshold_pascals: float,
+        plunger_impulse_time: float,
         output_option: OutputOptions = OutputOptions.can_bus_only,
         data_files: Optional[Dict[InstrumentProbeType, str]] = None,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
@@ -1387,6 +1388,7 @@ async def liquid_probe(
             plunger_speed=plunger_speed,
             mount_speed=mount_speed,
             threshold_pascals=threshold_pascals,
+            plunger_impulse_time=plunger_impulse_time,
             csv_output=csv_output,
             sync_buffer_output=sync_buffer_output,
             can_bus_only_output=can_bus_only_output,

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

@@ -345,6 +345,7 @@ async def liquid_probe(
         mount_speed: float,
         plunger_speed: float,
         threshold_pascals: float,
+        plunger_impulse_time: float,
         output_format: OutputOptions = OutputOptions.can_bus_only,
         data_files: Optional[Dict[InstrumentProbeType, str]] = None,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,

api/src/opentrons/hardware_control/ot3api.py

@@ -1932,6 +1932,37 @@ async def _move_to_plunger_bottom(
                 acquire_lock=acquire_lock,
             )
 
+    async def _move_to_plunger_top_for_liquid_probe(
+        self,
+        mount: OT3Mount,
+        rate: float,
+        acquire_lock: bool = True,
+    ) -> None:
+        """
+        Move an instrument's plunger to the top, to prepare for a following
+        liquid probe action.
+
+        The plunger backlash distance (mm) is used to ensure the plunger is pre-loaded
+        in the downward direction. This means that the final position will not be
+        the plunger's configured "top" position, but "top" plus the "backlashDistance".
+        """
+        max_speeds = self.config.motion_settings.default_max_speed
+        speed = max_speeds[self.gantry_load][OT3AxisKind.P]
+        instrument = self._pipette_handler.get_pipette(mount)
+        top_plunger_pos = target_position_from_plunger(
+            OT3Mount.from_mount(mount),
+            instrument.plunger_positions.top,
+            self._current_position,
+        )
+        target_pos = top_plunger_pos.copy()
+        target_pos[Axis.of_main_tool_actuator(mount)] += instrument.backlash_distance
+        await self._move(top_plunger_pos, speed=speed * rate, acquire_lock=acquire_lock)
+        # NOTE: This should ALWAYS be moving DOWN.
+        #       There should never be a time that this function is called and
+        #       the plunger doesn't physically move DOWN.
+        #       This is to make sure we are always engaged at the beginning of liquid-probe.
+        await self._move(target_pos, speed=speed * rate, acquire_lock=acquire_lock)
+
     async def configure_for_volume(
         self, mount: Union[top_types.Mount, OT3Mount], volume: float
     ) -> None:
@@ -2568,6 +2599,21 @@ def add_gripper_probe(self, probe: GripperProbe) -> None:
     def remove_gripper_probe(self) -> None:
         self._gripper_handler.remove_probe()
 
+    @staticmethod
+    def liquid_probe_non_responsive_z_distance(z_speed: float) -> float:
+        """Calculate the Z distance travelled where the LLD pass will be unresponsive."""
+        # NOTE: (sigler) Here lye some magic numbers.
+        #       The Z axis probing motion uses the first 20 samples to calculate
+        #       a baseline for all following samples, making the very beginning of
+        #       that Z motion unable to detect liquid. The sensor is configured for
+        #       4ms sample readings, and so we then assume it takes ~80ms to complete.
+        #       If the Z is moving at 5mm/sec, then ~80ms equates to ~0.4
+        baseline_during_z_sample_num = 20  # FIXME: (sigler) shouldn't be defined here?
+        sample_time_sec = 0.004  # FIXME: (sigler) shouldn't be defined here?
+        baseline_duration_sec = baseline_during_z_sample_num * sample_time_sec
+        non_responsive_z_mm = baseline_duration_sec * z_speed
+        return non_responsive_z_mm
+
     async def _liquid_probe_pass(
         self,
         mount: OT3Mount,
@@ -2583,6 +2629,7 @@ async def _liquid_probe_pass(
             probe_settings.mount_speed,
             (probe_settings.plunger_speed * plunger_direction),
             probe_settings.sensor_threshold_pascals,
+            probe_settings.plunger_impulse_time,
             probe_settings.output_option,
             probe_settings.data_files,
             probe=probe,
@@ -2626,11 +2673,15 @@ async def liquid_probe(
 
         probe_start_pos = await self.gantry_position(checked_mount, refresh=True)
 
-        p_travel = (
+        # plunger travel distance is from TOP->BOTTOM (minus the backlash distance + impulse)
+        # FIXME: logic for how plunger moves is divided between here and tool_sensors.py
+        p_impulse_mm = (
+            probe_settings.plunger_impulse_time * probe_settings.plunger_speed
+        )
+        p_total_mm = (
             instrument.plunger_positions.bottom - instrument.plunger_positions.top
         )
-        max_speeds = self.config.motion_settings.default_max_speed
-        p_prep_speed = max_speeds[self.gantry_load][OT3AxisKind.P]
+
         # We need to significatly slow down the 96 channel liquid probe
         if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
             max_plunger_speed = self.config.motion_settings.max_speed_discontinuity[
@@ -2640,24 +2691,44 @@ async def liquid_probe(
                 max_plunger_speed, probe_settings.plunger_speed
             )
 
+        p_working_mm = p_total_mm - (instrument.backlash_distance + p_impulse_mm)
+
+        # height where probe action will begin
+        # TODO: (sigler) add this to pipette's liquid def (per tip)
+        probe_pass_overlap_mm = 0.1
+        non_responsive_z_mm = OT3API.liquid_probe_non_responsive_z_distance(
+            probe_settings.mount_speed
+        )
+        probe_pass_z_offset_mm = non_responsive_z_mm + probe_pass_overlap_mm
+
+        # height that is considered safe to reset the plunger without disturbing liquid
+        # this usually needs to at least 1-2mm from liquid, to avoid splashes from air
+        # TODO: (sigler) add this to pipette's liquid def (per tip)
+        probe_safe_reset_mm = max(2.0, probe_pass_z_offset_mm)
+
         error: Optional[PipetteLiquidNotFoundError] = None
         pos = await self.gantry_position(checked_mount, refresh=True)
         while (probe_start_pos.z - pos.z) < max_z_dist:
             # safe distance so we don't accidentally aspirate liquid if we're already close to liquid
-            safe_plunger_pos = top_types.Point(pos.x, pos.y, pos.z + 2)
+            safe_plunger_pos = top_types.Point(
+                pos.x, pos.y, pos.z + probe_safe_reset_mm
+            )
             # overlap amount we want to use between passes
-            pass_start_pos = top_types.Point(pos.x, pos.y, pos.z + 0.5)
+            pass_start_pos = top_types.Point(
+                pos.x, pos.y, pos.z + probe_pass_z_offset_mm
+            )
             max_z_time = (
                 max_z_dist - (probe_start_pos.z - safe_plunger_pos.z)
             ) / probe_settings.mount_speed
-            pass_travel = min(max_z_time * probe_settings.plunger_speed, p_travel)
+            p_travel_required_for_z = max_z_time * probe_settings.plunger_speed
+            p_pass_travel = min(p_travel_required_for_z, p_working_mm)
             # Prep the plunger
             await self.move_to(checked_mount, safe_plunger_pos)
             if probe_settings.aspirate_while_sensing:
                 # TODO(cm, 7/8/24): remove p_prep_speed from the rate at some point
-                await self._move_to_plunger_bottom(checked_mount, rate=p_prep_speed)
+                await self._move_to_plunger_bottom(checked_mount, rate=1)
             else:
-                await self._move_to_plunger_top(checked_mount, rate=p_prep_speed)
+                await self._move_to_plunger_top_for_liquid_probe(checked_mount, rate=1)
 
             try:
                 # move to where we want to start a pass and run a pass
@@ -2666,7 +2737,7 @@ async def liquid_probe(
                     checked_mount,
                     probe_settings,
                     probe if probe else InstrumentProbeType.PRIMARY,
-                    pass_travel,
+                    p_pass_travel + p_impulse_mm,
                 )
                 # if we made it here without an error we found the liquid
                 error = None
@@ -2682,20 +2753,6 @@ async def liquid_probe(
             raise error
         return height
 
-    async def _move_to_plunger_top(
-        self,
-        mount: OT3Mount,
-        rate: float,
-        acquire_lock: bool = True,
-    ) -> None:
-        instrument = self._pipette_handler.get_pipette(mount)
-        target_pos = target_position_from_plunger(
-            OT3Mount.from_mount(mount),
-            instrument.plunger_positions.top,
-            self._current_position,
-        )
-        await self._move(target_pos, speed=rate, acquire_lock=acquire_lock)
-
     async def capacitive_probe(
         self,
         mount: OT3Mount,

api/tests/opentrons/hardware_control/backends/test_ot3_controller.py

@@ -719,6 +719,7 @@ async def test_liquid_probe(
             mount_speed=fake_liquid_settings.mount_speed,
             plunger_speed=fake_liquid_settings.plunger_speed,
             threshold_pascals=fake_liquid_settings.sensor_threshold_pascals,
+            plunger_impulse_time=fake_liquid_settings.plunger_impulse_time,
             output_option=fake_liquid_settings.output_option,
         )
     except PipetteLiquidNotFoundError:

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -116,8 +116,8 @@ def fake_settings() -> CapacitivePassSettings:
 @pytest.fixture
 def fake_liquid_settings() -> LiquidProbeSettings:
     return LiquidProbeSettings(
-        mount_speed=40,
-        plunger_speed=10,
+        mount_speed=5,
+        plunger_speed=20,
         plunger_impulse_time=0.2,
         sensor_threshold_pascals=15,
         output_option=OutputOptions.can_bus_only,
@@ -825,8 +825,8 @@ async def test_liquid_probe(
         # make sure aspirate while sensing reverses direction
         mock_liquid_probe.return_value = return_dict
         fake_settings_aspirate = LiquidProbeSettings(
-            mount_speed=40,
-            plunger_speed=10,
+            mount_speed=5,
+            plunger_speed=20,
             plunger_impulse_time=0.2,
             sensor_threshold_pascals=15,
             output_option=OutputOptions.can_bus_only,
@@ -838,10 +838,11 @@ async def test_liquid_probe(
         mock_move_to_plunger_bottom.call_count == 2
         mock_liquid_probe.assert_called_once_with(
             mount,
-            3.0,
+            52,
             fake_settings_aspirate.mount_speed,
             (fake_settings_aspirate.plunger_speed * -1),
             fake_settings_aspirate.sensor_threshold_pascals,
+            fake_settings_aspirate.plunger_impulse_time,
             fake_settings_aspirate.output_option,
             fake_settings_aspirate.data_files,
             probe=InstrumentProbeType.PRIMARY,
@@ -913,10 +914,11 @@ async def test_multi_liquid_probe(
         assert mock_move_to_plunger_bottom.call_count == 4
         mock_liquid_probe.assert_called_with(
             OT3Mount.LEFT,
-            plunger_positions.bottom - plunger_positions.top,
+            plunger_positions.bottom - plunger_positions.top - 0.1,
             fake_settings_aspirate.mount_speed,
             (fake_settings_aspirate.plunger_speed * -1),
             fake_settings_aspirate.sensor_threshold_pascals,
+            fake_settings_aspirate.plunger_impulse_time,
             fake_settings_aspirate.output_option,
             fake_settings_aspirate.data_files,
             probe=InstrumentProbeType.PRIMARY,
@@ -954,6 +956,7 @@ async def _fake_pos_update_and_raise(
         mount_speed: float,
         plunger_speed: float,
         threshold_pascals: float,
+        plunger_impulse_time: float,
         output_format: OutputOptions = OutputOptions.can_bus_only,
         data_files: Optional[Dict[InstrumentProbeType, str]] = None,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
@@ -986,8 +989,8 @@ async def _fake_pos_update_and_raise(
         await ot3_hardware.liquid_probe(
             OT3Mount.LEFT, fake_max_z_dist, fake_settings_aspirate
         )
-    # assert that it went through 3 passes and then prepared to aspirate
-    assert mock_move_to_plunger_bottom.call_count == 4
+    # assert that it went through 4 passes and then prepared to aspirate
+    assert mock_move_to_plunger_bottom.call_count == 5
 
 
 @pytest.mark.parametrize(

hardware/opentrons_hardware/hardware_control/tool_sensors.py

@@ -82,8 +82,6 @@
 # FIXME we should organize all of these functions to use the sensor drivers.
 # FIXME we should restrict some of these functions by instrument type.
 
-PLUNGER_SOLO_MOVE_TIME = 0.2
-
 
 def _fix_pass_step_for_buffer(
     move_group: MoveGroupStep,
@@ -393,6 +391,7 @@ async def liquid_probe(
     plunger_speed: float,
     mount_speed: float,
     threshold_pascals: float,
+    plunger_impulse_time: float,
     csv_output: bool = False,
     sync_buffer_output: bool = False,
     can_bus_only_output: bool = False,
@@ -425,15 +424,15 @@ async def liquid_probe(
         sensor_driver,
         True,
     )
-    p_prep_distance = float(PLUNGER_SOLO_MOVE_TIME * plunger_speed)
+    p_prep_distance = float(plunger_impulse_time * plunger_speed)
     p_pass_distance = float(max_p_distance - p_prep_distance)
     max_z_distance = (p_pass_distance / plunger_speed) * mount_speed
 
     lower_plunger = create_step(
         distance={tool: float64(p_prep_distance)},
         velocity={tool: float64(plunger_speed)},
         acceleration={},
-        duration=float64(PLUNGER_SOLO_MOVE_TIME),
+        duration=float64(plunger_impulse_time),
         present_nodes=[tool],
     )
     sensor_group = _build_pass_step(

hardware/tests/opentrons_hardware/hardware_control/test_tool_sensors.py

@@ -217,6 +217,7 @@ def move_responder(
         mount_speed=10,
         plunger_speed=8,
         threshold_pascals=threshold_pascals,
+        plunger_impulse_time=0.2,
         csv_output=False,
         sync_buffer_output=False,
         can_bus_only_output=False,
@@ -348,6 +349,7 @@ def move_responder(
             mount_speed=10,
             plunger_speed=8,
             threshold_pascals=14,
+            plunger_impulse_time=0.2,
             csv_output=csv_output,
             sync_buffer_output=sync_buffer_output,
             can_bus_only_output=can_bus_only_output,