recover 96ch

Author: Andiiiiiiyy

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/__main__.py

@@ -14,11 +14,11 @@ async def _main(cfg: TestConfig) -> None:
     # BUILD REPORT
     test_name = Path(__file__).parent.name
     ui.print_title(test_name.replace("_", " ").upper())
-
+    pipette_string = "p1000_96_v3.4" if cfg.pipette == 1000 else "p200_96_v3.0"
     # BUILD API
     api = await helpers_ot3.build_async_ot3_hardware_api(
         is_simulating=cfg.simulate,
-        pipette_left="p1000_96_v3.4",
+        pipette_left=pipette_string,
     )
 
     # CSV REPORT
@@ -49,7 +49,7 @@ async def _main(cfg: TestConfig) -> None:
     # RUN TESTS
     for section, test_run in cfg.tests.items():
         ui.print_title(section.value)
-        await test_run(api, report, section.value)
+        await test_run(api, report, section.value, cfg.pipette)
 
     # RELOAD PIPETTE
     ui.print_title("DONE")
@@ -71,6 +71,7 @@ async def _main(cfg: TestConfig) -> None:
         parser.add_argument(
             f"--only-{s.value.lower()}".replace("_", "-"), action="store_true"
         )
+    parser.add_argument("--pipette", type=int, choices=[200, 1000], default=1000)
     args = parser.parse_args()
     _t_sections = {
         s: f
@@ -87,5 +88,7 @@ async def _main(cfg: TestConfig) -> None:
             for s, f in TESTS
             if not getattr(args, f"skip_{s.value.lower().replace('-', '_')}")
         }
-    _config = TestConfig(simulate=args.simulate, tests=_t_sections)
+    _config = TestConfig(
+        simulate=args.simulate, tests=_t_sections, pipette=args.pipette
+    )
     asyncio.run(_main(_config))

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/config.py

@@ -1,7 +1,7 @@
 """Config."""
 from dataclasses import dataclass
 import enum
-from typing import Dict, Callable
+from typing import Dict, Callable, Literal
 
 from hardware_testing.data.csv_report import CSVReport, CSVSection
 
@@ -34,6 +34,7 @@ class TestConfig:
 
     simulate: bool
     tests: Dict[TestSection, Callable]
+    pipette: Literal[200, 1000]
 
 
 TESTS = [

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/test_capacitance.py

@@ -1,6 +1,6 @@
 """Test Capacitance."""
 from asyncio import sleep
-from typing import List, Union, Tuple, Optional, cast
+from typing import List, Union, Tuple, Optional, cast, Literal
 
 from opentrons_hardware.hardware_control.tool_sensors import capacitive_probe
 from opentrons_hardware.firmware_bindings.constants import NodeId, SensorId
@@ -45,6 +45,8 @@
     ),
 }
 
+PROBE_POSITIONS = [InstrumentProbeType.PRIMARY, InstrumentProbeType.SECONDARY]
+
 
 def _get_test_tag(probe: InstrumentProbeType, reading: str) -> str:
     return f"{probe.name.lower()}-{reading}"
@@ -53,7 +55,7 @@ def _get_test_tag(probe: InstrumentProbeType, reading: str) -> str:
 def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     """Build CSV Lines."""
     lines: List[Union[CSVLine, CSVLineRepeating]] = list()
-    for p in InstrumentProbeType:
+    for p in PROBE_POSITIONS:
         for r in PROBE_READINGS:
             lines.append(CSVLine(_get_test_tag(p, r), [float, CSVResult]))
             if "mm" in r:
@@ -104,7 +106,9 @@ def _get_hover_and_probe_pos(
     return hover_pos + probe_offset, probe_pos + probe_offset
 
 
-async def run(api: OT3API, report: CSVReport, section: str) -> None:
+async def run(
+    api: OT3API, report: CSVReport, section: str, pipette: Literal[200, 1000]
+) -> None:
     """Run."""
     z_ax = Axis.Z_L
     p_ax = Axis.P_L
@@ -116,7 +120,7 @@ async def run(api: OT3API, report: CSVReport, section: str) -> None:
     if not api.is_simulator:
         ui.get_user_ready("REMOVE everything from the deck")
 
-    for probe in InstrumentProbeType:
+    for probe in PROBE_POSITIONS:
         # store the thresolds (for reference)
         for k in THRESHOLDS.keys():
             report(section, _get_test_tag(probe, f"{k}-min"), [THRESHOLDS[k][0]])
@@ -145,7 +149,7 @@ async def run(api: OT3API, report: CSVReport, section: str) -> None:
         # ATTACHED-pF
         if not api.is_simulator:
             ui.get_user_ready(f"ATTACH probe to {probe.name} channel")
-        await api.add_tip(OT3Mount.LEFT, api.config.calibration.probe_length)
+        api.add_tip(OT3Mount.LEFT, api.config.calibration.probe_length)
         attached_pf = await _read_from_sensor(api, sensor_id, 10)
         if not attached_pf:
             ui.print_error(f"{probe} cap sensor not working, skipping")
@@ -173,7 +177,7 @@ async def _probe(distance: float, speed: float) -> float:
                 NodeId.pipette_left,
                 NodeId.head_l,
                 distance=distance,
-                speed=speed,
+                mount_speed=speed,
                 sensor_id=sensor_id,
                 relative_threshold_pf=default_probe_cfg.sensor_threshold_pf,
             )
@@ -229,4 +233,4 @@ async def _probe(distance: float, speed: float) -> float:
         await api.home_z(OT3Mount.LEFT)
         if not api.is_simulator:
             ui.get_user_ready("REMOVE probe")
-        await api.remove_tip(OT3Mount.LEFT)
+        api.remove_tip(OT3Mount.LEFT)

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/test_droplets.py

@@ -1,7 +1,7 @@
 """Test Droplets."""
 from asyncio import sleep
 from time import time
-from typing import List, Union, Tuple, Optional, Dict
+from typing import List, Union, Tuple, Optional, Dict, Literal
 
 from opentrons.hardware_control.ot3api import OT3API
 from opentrons.hardware_control.motion_utilities import target_position_from_relative
@@ -16,17 +16,14 @@
 from hardware_testing.opentrons_api import helpers_ot3
 from hardware_testing.opentrons_api.types import OT3Mount, Point, Axis
 
-TIP_VOLUME = 1000
-ASPIRATE_VOLUME = [1000, 5]
 NUM_SECONDS_TO_WAIT = 30
 HOVER_HEIGHT_MM = 50
 DEPTH_INTO_RESERVOIR_FOR_ASPIRATE = -24
 DEPTH_INTO_RESERVOIR_FOR_DISPENSE = DEPTH_INTO_RESERVOIR_FOR_ASPIRATE
 
-TIP_RACK_LABWARE = f"opentrons_flex_96_tiprack_{TIP_VOLUME}ul"
 RESERVOIR_LABWARE = "nest_1_reservoir_195ml"
 
-TIP_RACK_96_SLOT = 10
+TIP_RACK_96_SLOT = 4
 TIP_RACK_PARTIAL_SLOT = 5
 RESERVOIR_SLOT = 2
 TRASH_SLOT = 12
@@ -86,31 +83,31 @@ def get_reservoir_nominal() -> Point:
     return reservoir_a1_nominal
 
 
-def get_tiprack_96_nominal() -> Point:
+def get_tiprack_96_nominal(pipette: Literal[200, 1000]) -> Point:
     """Get nominal tiprack position for 96-tip pick-up."""
     tip_rack_a1_nominal = helpers_ot3.get_theoretical_a1_position(
-        TIP_RACK_96_SLOT, TIP_RACK_LABWARE
+        TIP_RACK_96_SLOT, f"opentrons_flex_96_tiprack_{pipette}ul"
     )
     return tip_rack_a1_nominal + Point(z=TIP_RACK_96_ADAPTER_HEIGHT)
 
 
-def get_tiprack_partial_nominal() -> Point:
+def get_tiprack_partial_nominal(pipette: Literal[200, 1000]) -> Point:
     """Get nominal tiprack position for partial-tip pick-up."""
     tip_rack_a1_nominal = helpers_ot3.get_theoretical_a1_position(
-        TIP_RACK_PARTIAL_SLOT, TIP_RACK_LABWARE
+        TIP_RACK_PARTIAL_SLOT, f"opentrons_flex_96_tiprack_{pipette}ul"
     )
     return tip_rack_a1_nominal
 
 
 async def aspirate_and_wait(
-    api: OT3API, reservoir: Point, aspirate_value, seconds: int = 30
+    api: OT3API, reservoir: Point, volume: int, seconds: int = 30
 ) -> Tuple[bool, float]:
     """Aspirate and wait."""
     await helpers_ot3.move_to_arched_ot3(api, OT3Mount.LEFT, reservoir)
     await api.move_to(
         OT3Mount.LEFT, reservoir + Point(z=DEPTH_INTO_RESERVOIR_FOR_ASPIRATE)
     )
-    await api.aspirate(OT3Mount.LEFT, aspirate_value)
+    await api.aspirate(OT3Mount.LEFT, volume)
     await api.move_to(OT3Mount.LEFT, reservoir + Point(z=HOVER_HEIGHT_MM))
 
     start_time = time()
@@ -136,16 +133,16 @@ async def aspirate_and_wait(
     return result, duration_seconds
 
 
-async def _drop_tip(api: OT3API, trash: Point) -> None:
+async def _drop_tip(api: OT3API, trash: Point, pipette: Literal[200, 1000]) -> None:
     print("drop in trash")
     await helpers_ot3.move_to_arched_ot3(api, OT3Mount.LEFT, trash + Point(z=20))
     await api.move_to(OT3Mount.LEFT, trash)
     await api.drop_tip(OT3Mount.LEFT)
     # NOTE: a FW bug (as of v14) will sometimes not fully drop tips.
     #       so here we ask if the operator needs to try again
-    # while not api.is_simulator and ui.get_user_answer("try dropping again"):
-    #     await api.add_tip(OT3Mount.LEFT, helpers_ot3.get_default_tip_length(TIP_VOLUME))
-    #     await api.drop_tip(OT3Mount.LEFT)
+    while not api.is_simulator and ui.get_user_answer("try dropping again"):
+        api.add_tip(OT3Mount.LEFT, helpers_ot3.get_default_tip_length(pipette))
+        await api.drop_tip(OT3Mount.LEFT)
     await api.home_z(OT3Mount.LEFT)
 
 
@@ -164,32 +161,36 @@ async def _partial_pick_up_z_motion(
     await api._update_position_estimation([Axis.Z_L])
 
 
-async def _partial_pick_up(api: OT3API, position: Point, current: float) -> None:
+async def _partial_pick_up(
+    api: OT3API, position: Point, current: float, pipette: Literal[200, 1000]
+) -> None:
     await helpers_ot3.move_to_arched_ot3(
         api,
         OT3Mount.LEFT,
         position,
         safe_height=position.z + 10,
     )
     await _partial_pick_up_z_motion(api, current=current, distance=13, speed=5)
-    await api.add_tip(OT3Mount.LEFT, helpers_ot3.get_default_tip_length(TIP_VOLUME))
+    api.add_tip(OT3Mount.LEFT, helpers_ot3.get_default_tip_length(pipette))
     await api.prepare_for_aspirate(OT3Mount.LEFT)
     await api.home_z(OT3Mount.LEFT)
 
 
-async def run(api: OT3API, report: CSVReport, section: str) -> None:
+async def run(
+    api: OT3API, report: CSVReport, section: str, pipette: Literal[200, 1000]
+) -> None:
     """Run."""
     # GATHER NOMINAL POSITIONS
     trash_nominal = get_trash_nominal()
-    tip_rack_96_a1_nominal = get_tiprack_96_nominal()
+    tip_rack_96_a1_nominal = get_tiprack_96_nominal(pipette)
     # tip_rack_partial_a1_nominal = get_tiprack_partial_nominal()
     reservoir_a1_nominal = get_reservoir_nominal()
     reservoir_a1_actual: Optional[Point] = None
 
     async def _find_reservoir_pos() -> None:
         nonlocal reservoir_a1_actual
-        # if reservoir_a1_actual:  # re-find reservoir position for 5ul
-        #     return
+        if reservoir_a1_actual:
+            return
         # SAVE RESERVOIR POSITION
         ui.print_header("JOG to TOP of RESERVOIR")
         print("jog tips to the TOP of the RESERVOIR")
@@ -199,37 +200,47 @@ async def _find_reservoir_pos() -> None:
         await helpers_ot3.jog_mount_ot3(api, OT3Mount.LEFT)
         reservoir_a1_actual = await api.gantry_position(OT3Mount.LEFT)
 
-    result = True
-    for test_volume in ASPIRATE_VOLUME:
-        answer = ui.get_user_answer(f"Test {test_volume}uL")
-        if not answer:
-            continue
-        tip_volume = 50 if test_volume<=50 else 1000
-        # PICK-UP 96 TIPS
+    # PICK-UP 96 TIPS
+    droplets_result = True
+    for trial in range(2):
         ui.print_header("JOG to 96-Tip RACK")
+        if trial == 0:
+            tip_rack = str(pipette) + "ul"
+            test_volume: int = pipette
+        else:
+            tip_rack = "50ul"
+            test_volume = 1 if pipette == 200 else 5
         if not api.is_simulator:
-            ui.get_user_ready(f"picking up tips, place tip-rack {tip_volume} on slot {TIP_RACK_96_SLOT}")
+            ui.get_user_ready(f"ADD 96 tip-rack-{tip_rack} to slot #{TIP_RACK_96_SLOT}")
         await helpers_ot3.move_to_arched_ot3(
             api, OT3Mount.LEFT, tip_rack_96_a1_nominal + Point(z=30)
         )
         await helpers_ot3.jog_mount_ot3(api, OT3Mount.LEFT)
-
-        await api.pick_up_tip(OT3Mount.LEFT, helpers_ot3.get_default_tip_length(tip_volume))
+        print("picking up tips")
+        await api.pick_up_tip(
+            OT3Mount.LEFT, helpers_ot3.get_default_tip_length(pipette)
+        )
         await api.home_z(OT3Mount.LEFT)
-        if not api.is_simulator:
-            ui.get_user_ready("about to move to RESERVOIR")
-        # TEST DROPLETS for 96 TIPS
-        ui.print_header("96 Tips: ASPIRATE and WAIT")
-        await _find_reservoir_pos()
+        if reservoir_a1_actual is None:
+            if not api.is_simulator:
+                ui.get_user_ready("about to move to RESERVOIR")
+
+            # TEST DROPLETS for 96 TIPS
+            ui.print_header("96 Tips: ASPIRATE and WAIT")
+            await _find_reservoir_pos()
         assert reservoir_a1_actual
-        ret, duration = await aspirate_and_wait(
-            api, reservoir_a1_actual, test_volume, seconds=NUM_SECONDS_TO_WAIT
+        result, duration = await aspirate_and_wait(
+            api,
+            reservoir_a1_actual,
+            test_volume,
+            seconds=NUM_SECONDS_TO_WAIT,
         )
-        result = result&ret
+        droplets_result = droplets_result & result
+        await _drop_tip(api, trash_nominal, pipette)
         await api.home_z(OT3Mount.LEFT)
-        await _drop_tip(api, trash_nominal)
-    report(section, "droplets-96-tips", [duration, CSVResult.from_bool(result)])
-    
+    report(
+        section, "droplets-96-tips", [duration, CSVResult.from_bool(droplets_result)]
+    )
 
     # if not api.is_simulator:
     #     ui.get_user_ready(f"REMOVE 96 tip-rack from slot #{TIP_RACK_96_SLOT}")

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/test_environmental_sensor.py

@@ -1,6 +1,6 @@
 """Test Environmental Sensor."""
 from asyncio import sleep
-from typing import List, Union
+from typing import List, Union, Literal
 
 from opentrons.hardware_control.ot3api import OT3API
 
@@ -11,22 +11,19 @@
 from hardware_testing.opentrons_api.types import OT3Mount
 from hardware_testing.data.csv_report import (
     CSVReport,
-    CSVResult,
     CSVLine,
     CSVLineRepeating,
 )
 
 
 NUM_SAMPLES = 10
 INTER_SAMPLE_DELAY_SECONDS = 0.25
-TEMPERATURE_THRESHOLD = [10, 40]
-HUMIDITY_THRESHOLD = [10, 90]
 
 
 def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     """Build CSV Lines."""
     return [
-        CSVLine(f"environment-{sensor_id.name}-celsius-humidity", [float, float, CSVResult])
+        CSVLine(f"environment-{sensor_id.name}-celsius-humidity", [float, float])
         for sensor_id in [SensorId.S0, SensorId.S1]
     ]
 
@@ -36,7 +33,9 @@ def _remove_outliers_and_average(values: List[float]) -> float:
     return sum(no_outliers) / len(no_outliers)
 
 
-async def run(api: OT3API, report: CSVReport, section: str) -> None:
+async def run(
+    api: OT3API, report: CSVReport, section: str, pipette: Literal[200, 1000]
+) -> None:
     """Run."""
     await api.home_z(OT3Mount.LEFT)
     slot_5 = helpers_ot3.get_slot_calibration_square_position_ot3(5)
@@ -46,7 +45,6 @@ async def run(api: OT3API, report: CSVReport, section: str) -> None:
         ui.print_header(sensor_id.name.upper())
         celsius_samples = []
         humidity_samples = []
-        air_params = True
         print(f"averaging {NUM_SAMPLES} samples:")
         print("\tc\th")
         for _ in range(NUM_SAMPLES):
@@ -64,10 +62,8 @@ async def run(api: OT3API, report: CSVReport, section: str) -> None:
         humidity = _remove_outliers_and_average(humidity_samples)
         print(f"[{sensor_id.name}] Celsius = {round(celsius, 2)} degrees")
         print(f"[{sensor_id.name}] Humidity = {round(humidity, 2)} percent")
-        air_params = air_params & True if TEMPERATURE_THRESHOLD[0] <= celsius <= TEMPERATURE_THRESHOLD[1] and HUMIDITY_THRESHOLD[0] <= humidity <= HUMIDITY_THRESHOLD[1] else False
-    
         report(
             section,
             f"environment-{sensor_id.name}-celsius-humidity",
-            [celsius, humidity, CSVResult.from_bool(air_params)]
+            [celsius, humidity],
         )