fix(api): Lld movement adjustments (#15732)

<!--
Thanks for taking the time to open a pull request! Please make sure
you've read the "Opening Pull Requests" section of our Contributing
Guide:


https://github.com/Opentrons/opentrons/blob/edge/CONTRIBUTING.md#opening-pull-requests

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

In order to remove some magic numbers we do a more clear calculation
about where certain numbers in ot3api.py come from

additionally we added a more robust move_to_plunger_top that
incorporates backlash values, mostly due to an expectation that the 96
channel will need this due to it's current physical requirements for
backlash
<!--
Use this section to describe your pull-request at a high level. If the
PR addresses any open issues, please tag the issues here.
-->

# Test Plan

<!--
Use this section to describe the steps that you took to test your Pull
Request.
If you did not perform any testing provide justification why.

OT-3 Developers: You should default to testing on actual physical
hardware.
Once again, if you did not perform testing against hardware, justify
why.

Note: It can be helpful to write a test plan before doing development

Example Test Plan (HTTP API Change)

- Verified that new optional argument `dance-party` causes the robot to
flash its lights, move the pipettes,
then home.
- Verified that when you omit the `dance-party` option the robot homes
normally
- Added protocol that uses `dance-party` argument to G-Code Testing
Suite
- Ran protocol that did not use `dance-party` argument and everything
was successful
- Added unit tests to validate that changes to pydantic model are
correct

-->

# Changelog

<!--
List out the changes to the code in this PR. Please try your best to
categorize your changes and describe what has changed and why.

Example changelog:
- Fixed app crash when trying to calibrate an illegal pipette
- Added state to API to track pipette usage
- Updated API docs to mention only two pipettes are supported

IMPORTANT: MAKE SURE ANY BREAKING CHANGES ARE PROPERLY COMMUNICATED
-->

# Review requests

<!--
Describe any requests for your reviewers here.
-->

# Risk assessment

<!--
Carefully go over your pull request and look at the other parts of the
codebase it may affect. Look for the possibility, even if you think it's
small, that your change may affect some other part of the system - for
instance, changing return tip behavior in protocol may also change the
behavior of labware calibration.

Identify the other parts of the system your codebase may affect, so that
in addition to your own review and testing, other people who may not
have the system internalized as much as you can focus their attention
and testing there.
-->

---------

Co-authored-by: Andy Sigler <[email protected]>

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,