Rh 23 remove hw api from hw tests (#14209)

* call instrument context configure

* don't force an ALL tip config, its that by default and breaks the single channel

* remove easy instances of using hw_api

* at least unify all the calls of get_hardware

* call retract less now that we don't need it as much with the tip rack adapter feature

* use some updated protocol api commands

* add a retract command to intrument context

* we have to load the trash manually now

* format

* fix tests

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

@@ -778,3 +778,7 @@ def configure_nozzle_layout(
         self._engine_client.configure_nozzle_layout(
             pipette_id=self._pipette_id, configuration_params=configuration_model
         )
+
+    def retract(self) -> None:
+        """Retract this instrument to the top of the gantry."""
+        self._sync_hardware_api.retract(self.get_mount())

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

@@ -284,6 +284,9 @@ def configure_nozzle_layout(
     @abstractmethod
     def is_tip_tracking_available(self) -> bool:
         """Return whether auto tip tracking is available for the pipette's current nozzle configuration."""
+
+    def _retract(self) -> None:
+        """Retract this instrument to the top of the gantry."""
         ...
 
 

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

@@ -553,3 +553,7 @@ def get_active_channels(self) -> int:
     def is_tip_tracking_available(self) -> bool:
         # Tip tracking is always available in legacy context
         return True
+
+    def _retract(self) -> None:
+        """Retract this instrument to the top of the gantry."""
+        self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]

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

@@ -471,3 +471,7 @@ def get_active_channels(self) -> int:
     def is_tip_tracking_available(self) -> bool:
         # Tip tracking is always available in legacy context
         return True
+
+    def _retract(self) -> None:
+        """Retract this instrument to the top of the gantry."""
+        self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]

api/src/opentrons/protocol_api/instrument_context.py

@@ -1478,7 +1478,13 @@ def move_to(
 
         return self
 
-    @property
+    @requires_version(2, 0)
+    def _retract(
+        self,
+    ) -> None:
+        self._core._retract()
+
+    @property  # type: ignore
     @requires_version(2, 0)
     def mount(self) -> str:
         """

hardware-testing/hardware_testing/gravimetric/__main__.py

@@ -49,8 +49,7 @@
 from opentrons.protocol_api import InstrumentContext
 from opentrons.protocol_engine.types import LabwareOffset
 
-# FIXME: bump to v2.15 to utilize protocol engine
-API_LEVEL = "2.15"
+API_LEVEL = "2.16"
 
 LABWARE_OFFSETS: List[LabwareOffset] = []
 
@@ -314,7 +313,7 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":  # noqa: C901
                 trials=trials,
                 name=name,
                 robot_serial=robot_serial,
-                fw_version=_ctx._core.get_hardware().fw_version,
+                fw_version=workarounds.get_sync_hw_api(_ctx).fw_version,
             )
         else:
             if args.increment:
@@ -347,7 +346,7 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":  # noqa: C901
                 name=name,
                 environment_sensor=environment_sensor,
                 trials=trials,
-                fw_version=_ctx._core.get_hardware().fw_version,
+                fw_version=workarounds.get_sync_hw_api(_ctx).fw_version,
             )
 
         return RunArgs(
@@ -591,7 +590,7 @@ def _main(
             shell=True,
         )
         sleep(1)
-    hw = run_args.ctx._core.get_hardware()
+    hw = workarounds.get_sync_hw_api(run_args.ctx)
     try:
         if not run_args.ctx.is_simulating() and not args.photometric:
             ui.get_user_ready("CLOSE the door, and MOVE AWAY from machine")

hardware-testing/hardware_testing/gravimetric/daily_setup.py

@@ -13,8 +13,9 @@
 )
 from hardware_testing.gravimetric.config import GANTRY_MAX_SPEED
 from hardware_testing.gravimetric.measurement.scale import Scale  # type: ignore[import]
-from hardware_testing.gravimetric import helpers, workarounds
+from hardware_testing.gravimetric import helpers
 from hardware_testing.gravimetric.__main__ import API_LEVEL
+from hardware_testing.gravimetric.workarounds import get_sync_hw_api
 
 TEST_NAME = "gravimetric-daily-setup"
 
@@ -253,7 +254,7 @@ def _calibrate() -> None:
         API_LEVEL,  # type: ignore[attr-defined]
         is_simulating=args.simulate,
     )
-    _hw = workarounds.get_sync_hw_api(_ctx)
+    _hw = get_sync_hw_api(_ctx)
     _hw.set_status_bar_state(COLOR_STATES["idle"])
     _rec = GravimetricRecorder(
         GravimetricRecorderConfig(

hardware-testing/hardware_testing/gravimetric/execute.py

@@ -52,6 +52,7 @@
 import glob
 
 from opentrons.hardware_control.types import StatusBarState
+from hardware_testing.gravimetric.workarounds import get_sync_hw_api
 
 _MEASUREMENTS: List[Tuple[str, MeasurementData]] = list()
 
@@ -88,7 +89,7 @@ def _generate_callbacks_for_trial(
     if blank_measurement:
         volume = None
 
-    hw_api = ctx._core.get_hardware()
+    hw_api = get_sync_hw_api(ctx)
     hw_mount = OT3Mount.LEFT if pipette.mount == "left" else OT3Mount.RIGHT
     pip_ax = Axis.of_main_tool_actuator(hw_mount)
     estimate_bottom: float = -1
@@ -329,8 +330,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> MeasurementData:
     else:
         # center channel over well
         trial.pipette.move_to(trial.well.top(50).move(trial.channel_offset))
-    mnt = OT3Mount.RIGHT if trial.pipette.mount == "right" else OT3Mount.LEFT
-    trial.ctx._core.get_hardware().retract(mnt)  # retract to top of gantry
+    trial.pipette._retract()  # retract to top of gantry
     m_data_init = _record_measurement_and_store(MeasurementType.INIT)
     ui.print_info(f"\tinitial grams: {m_data_init.grams_average} g")
     # update the vials volumes, using the last-known weight
@@ -359,7 +359,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> MeasurementData:
         mode=trial.mode,
         clear_accuracy_function=trial.cfg.increment,
     )
-    trial.ctx._core.get_hardware().retract(mnt)  # retract to top of gantry
+    trial.pipette._retract()  # retract to top of gantry
 
     _take_photos(trial, "aspirate")
     m_data_aspirate = _record_measurement_and_store(MeasurementType.ASPIRATE)
@@ -381,7 +381,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> MeasurementData:
         mode=trial.mode,
         clear_accuracy_function=trial.cfg.increment,
     )
-    trial.ctx._core.get_hardware().retract(mnt)  # retract to top of gantry
+    trial.pipette._retract()  # retract to top of gantry
     _take_photos(trial, "dispense")
     m_data_dispense = _record_measurement_and_store(MeasurementType.DISPENSE)
     ui.print_info(f"\tgrams after dispense: {m_data_dispense.grams_average} g")
@@ -502,8 +502,7 @@ def _calculate_evaporation(
         resources.env_sensor,
     )
     ui.print_info(f"running {config.NUM_BLANK_TRIALS}x blank measurements")
-    mnt = OT3Mount.RIGHT if resources.pipette.mount == "right" else OT3Mount.LEFT
-    resources.ctx._core.get_hardware().retract(mnt)
+    resources.pipette._retract()
     for i in range(config.SCALE_SECONDS_TO_TRUE_STABILIZE):
         ui.print_info(
             f"wait for scale to stabilize "
@@ -547,7 +546,7 @@ def _get_liquid_height(
         if not resources.ctx.is_simulating() and not cfg.same_tip:
             ui.alert_user_ready(
                 f"Please replace the {cfg.tip_volume}ul tips in slot 2",
-                resources.ctx._core.get_hardware(),
+                get_sync_hw_api(resources.ctx),
             )
         _tip_counter[0] = 0
     if cfg.jog:
@@ -597,7 +596,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None:  # noq
     recorder._recording = GravimetricRecording()
     report.store_config_gm(resources.test_report, cfg)
     calibration_tip_in_use = True
-    hw_api = resources.ctx._core.get_hardware()
+    hw_api = get_sync_hw_api(resources.ctx)
     if resources.ctx.is_simulating():
         _PREV_TRIAL_GRAMS = None
         _MEASUREMENTS = list()
@@ -607,8 +606,6 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None:  # noq
         setup_channel_offset = _get_channel_offset(cfg, channel=0)
         first_tip_location = first_tip.top().move(setup_channel_offset)
         _pick_up_tip(resources.ctx, resources.pipette, cfg, location=first_tip_location)
-        mnt = OT3Mount.LEFT if cfg.pipette_mount == "left" else OT3Mount.RIGHT
-        resources.ctx._core.get_hardware().retract(mnt)
         ui.print_info("moving to scale")
         well = labware_on_scale["A1"]
         _liquid_height = _get_liquid_height(resources, cfg, well)
@@ -696,12 +693,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None:  # noq
                             cfg,
                             location=next_tip_location,
                         )
-                        mnt = (
-                            OT3Mount.LEFT
-                            if cfg.pipette_mount == "left"
-                            else OT3Mount.RIGHT
-                        )
-                        resources.ctx._core.get_hardware().retract(mnt)
+                        resources.pipette._retract()  # retract to top of gantry
                     (
                         actual_aspirate,
                         aspirate_data,
@@ -744,12 +736,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None:  # noq
                     )
                     ui.print_info("dropping tip")
                     if not cfg.same_tip:
-                        mnt = (
-                            OT3Mount.LEFT
-                            if cfg.pipette_mount == "left"
-                            else OT3Mount.RIGHT
-                        )
-                        resources.ctx._core.get_hardware().retract(mnt)
+                        resources.pipette._retract()  # retract to top of gantry
                         _drop_tip(
                             resources.pipette, cfg.return_tip, _minimum_z_height(cfg)
                         )

hardware-testing/hardware_testing/gravimetric/execute_photometric.py

@@ -5,7 +5,7 @@
 from opentrons.protocol_api import ProtocolContext, Well, Labware
 
 from hardware_testing.data import ui
-from hardware_testing.opentrons_api.types import Point, OT3Mount
+from hardware_testing.opentrons_api.types import Point
 from .measurement import (
     MeasurementType,
     create_measurement_tag,
@@ -215,7 +215,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> EnvironmentData:
             touch_tip=trial.cfg.touch_tip,
         )
         _record_measurement_and_store(MeasurementType.DISPENSE)
-        trial.ctx._core.get_hardware().retract(OT3Mount.LEFT)
+        trial.pipette._retract()  # retract to top of gantry
         if (i + 1) == num_dispenses:
             if not trial.cfg.same_tip:
                 _drop_tip(trial.pipette, trial.cfg.return_tip)

hardware-testing/hardware_testing/gravimetric/helpers.py

@@ -82,19 +82,23 @@ async def _thread_manager_build_hw_api(
             stall_detection_enable=stall_detection_enable,
         )
 
+    papi: protocol_api.ProtocolContext
     if is_simulating:
-        return simulate.get_protocol_api(
+        papi = simulate.get_protocol_api(
             version=APIVersion.from_string(api_level),
             extra_labware=extra_labware,
             hardware_simulator=ThreadManager(_thread_manager_build_hw_api),
             robot_type="Flex",
             use_virtual_hardware=False,
         )
     else:
-        return execute.get_protocol_api(
+        papi = execute.get_protocol_api(
             version=APIVersion.from_string(api_level), extra_labware=extra_labware
         )
 
+    papi.load_labware("opentrons_1_trash_3200ml_fixed", "A3")
+    return papi
+
 
 def well_is_reservoir(well: protocol_api.labware.Well) -> bool:
     """Well is reservoir."""
@@ -295,8 +299,6 @@ def _pick_up_tip(
         f"from slot #{location.labware.parent.parent}"
     )
     pipette.pick_up_tip(location)
-    if pipette.channels == 96:
-        get_sync_hw_api(ctx).retract(OT3Mount.LEFT)
     # NOTE: the accuracy-adjust function gets set on the Pipette
     #       each time we pick-up a new tip.
     if cfg.increment:

hardware-testing/hardware_testing/gravimetric/liquid_class/pipetting.py

@@ -8,6 +8,7 @@
 
 from hardware_testing.opentrons_api.types import OT3AxisKind
 from hardware_testing.gravimetric import config
+from hardware_testing.gravimetric.workarounds import get_sync_hw_api
 from hardware_testing.gravimetric.liquid_height.height import LiquidTracker
 from hardware_testing.opentrons_api.types import OT3Mount, Point
 from hardware_testing.opentrons_api.helpers_ot3 import clear_pipette_ul_per_mm
@@ -131,7 +132,7 @@ def _retract(
     z_discontinuity: float,
 ) -> None:
     # change discontinuity per the liquid-class settings
-    hw_api = ctx._core.get_hardware()
+    hw_api = get_sync_hw_api(ctx)
     if pipette.channels == 96:
         hw_api.config.motion_settings.max_speed_discontinuity.high_throughput[
             OT3AxisKind.Z
@@ -177,7 +178,7 @@ def _pipette_with_liquid_settings(  # noqa: C901
 ) -> None:
     """Run a pipette given some Pipetting Liquid Settings."""
     # FIXME: stop using hwapi, and get those functions into core software
-    hw_api = ctx._core.get_hardware()
+    hw_api = get_sync_hw_api(ctx)
     hw_mount = OT3Mount.LEFT if pipette.mount == "left" else OT3Mount.RIGHT
     hw_pipette = hw_api.hardware_pipettes[hw_mount.to_mount()]
     _check_aspirate_dispense_args(mix, aspirate, dispense)
@@ -223,16 +224,17 @@ def _aspirate_on_approach() -> None:
                 "WARNING: removing trailing air-gap from pipette, "
                 "this should only happen during blank trials"
             )
-            hw_api.dispense(hw_mount)
+            pipette.dispense(volume=pipette.current_volume)
         if mode:
             # NOTE: increment test requires the plunger's "bottom" position
             #       does not change during the entire test run
             hw_api.set_liquid_class(hw_mount, mode)
         else:
-            hw_api.configure_for_volume(hw_mount, aspirate if aspirate else dispense)
+            cfg_volume: float = aspirate if aspirate else dispense  # type: ignore[assignment]
+            pipette.configure_for_volume(cfg_volume)
         if clear_accuracy_function:
             clear_pipette_ul_per_mm(hw_api, hw_mount)  # type: ignore[arg-type]
-        hw_api.prepare_for_aspirate(hw_mount)
+        pipette.prepare_to_aspirate()
         if liquid_class.aspirate.leading_air_gap > 0:
             pipette.aspirate(liquid_class.aspirate.leading_air_gap)
 
@@ -257,7 +259,7 @@ def _aspirate_on_mix() -> None:
             ctx, pipette, well, channel_offset, approach_mm, retract_speed, _z_disc
         )
         pipette.blow_out()
-        hw_api.prepare_for_aspirate(hw_mount)
+        pipette.prepare_to_aspirate()
         assert pipette.current_volume == 0
 
     def _aspirate_on_submerge() -> None:

hardware-testing/tests/hardware_testing/liquid/test_heights.py

@@ -17,7 +17,7 @@
 
 
 def _create_context() -> ProtocolContext:
-    return get_api_context(api_level="2.15", is_simulating=True)
+    return get_api_context(api_level="2.16", is_simulating=True)
 
 
 def _load_labware(ctx: ProtocolContext) -> Tuple[Labware, Labware, Labware, Labware]: