From 5144feca255f2b2c5eddb2de0d3f0f090a882991 Mon Sep 17 00:00:00 2001 From: Ryan Howard Date: Mon, 21 Aug 2023 13:18:08 -0400 Subject: [PATCH] feat(hardware-testing): add upgrades for improved 8 channel QC tests (#13261) * Collect all user input at the begining of a multi-tip run * combine the p1k multip QC tests into a single protocol * fix the get-tip-per-channel for the 8 channel * add code to read serial from asair sensor * make the driver return the correct type values * temp * move more things into the shared section of main * pass around the recorder * get the report into shared resources * fixups * make user messages better * use labware's tiptracker to track unused tips * more fixups for 8 channel * pack the %d cv standards into the trials * raise an error if the channel doesn't pass * get liquid probe working * we no longer need the alert user ready call * run the 8 channel with 10 trials instead of 8 * add list of calls to the read me * lint * fix CI taking too long and format .md file * add ability to turn off liquid probe. and add paramaterization for liquid probe * use the tested liquid probe values * set the spec dictionary to the advertised spec, then add saftey factor as a seperate variable * style fixes * fix liquid probe in simulator and remove unneeded print statements * fix the asair driver tests to match how the device actually responds * force a 2cm max distance on liquid probe --- hardware-testing/Makefile | 16 +- .../hardware_testing/drivers/asair_sensor.py | 40 +- .../hardware_testing/gravimetric/README.md | 27 ++ .../hardware_testing/gravimetric/__main__.py | 438 ++++++++++++------ .../hardware_testing/gravimetric/config.py | 321 ++++++++++--- .../hardware_testing/gravimetric/execute.py | 139 ++++-- .../gravimetric/execute_photometric.py | 35 +- .../hardware_testing/gravimetric/helpers.py | 140 ++++-- .../gravimetric/measurement/record.py | 6 +- .../hardware_testing/gravimetric/report.py | 57 ++- .../hardware_testing/gravimetric/tips.py | 96 ++-- .../hardware_testing/gravimetric/trial.py | 45 +- .../gravimetric/workarounds.py | 2 - .../protocols/gravimetric_ot3_p1000_96.py | 28 ++ ..._tip.py => gravimetric_ot3_p1000_multi.py} | 6 +- .../gravimetric_ot3_p1000_multi_200ul_tip.py | 27 -- ...ul_tip.py => gravimetric_ot3_p50_multi.py} | 10 +- .../scripts/gravimetric_rnd.py | 2 + .../drivers/test_asair_sensor.py | 2 +- 19 files changed, 1019 insertions(+), 418 deletions(-) create mode 100644 hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_96.py rename hardware-testing/hardware_testing/protocols/{gravimetric_ot3_p1000_multi_1000ul_tip.py => gravimetric_ot3_p1000_multi.py} (86%) delete mode 100644 hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_200ul_tip.py rename hardware-testing/hardware_testing/protocols/{gravimetric_ot3_p1000_multi_50ul_tip.py => gravimetric_ot3_p50_multi.py} (68%) diff --git a/hardware-testing/Makefile b/hardware-testing/Makefile index 29811bb1a31..2a3c86657b8 100755 --- a/hardware-testing/Makefile +++ b/hardware-testing/Makefile @@ -95,14 +95,18 @@ test-photometric: .PHONY: test-gravimetric-single test-gravimetric-single: - $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --trials 1 - $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --extra --no-blank - $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 1 --no-blank - $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --trials 1 --increment --no-blank + $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 + $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --extra --no-blank --trials 1 + $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 1 --no-blank --trials 1 + $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 1 --no-blank --trials 1 --increment --tip 50 + $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --trials 1 --increment --no-blank --tip 50 + $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --trials 1 --increment --no-blank --tip 200 + $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 1 --trials 1 --increment --no-blank --tip 1000 .PHONY: test-gravimetric-multi test-gravimetric-multi: - $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 8 --tip 50 --trials 1 --no-blank + $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 8 --tip 50 + $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 8 --tip 50 --trials 1 --no-blank --extra $(python) -m hardware_testing.gravimetric --simulate --pipette 50 --channels 8 --tip 50 --trials 1 --increment --no-blank $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 8 --tip 1000 --trials 1 --no-blank $(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 8 --tip 200 --trials 1 --extra --no-blank @@ -117,6 +121,8 @@ test-gravimetric: -$(MAKE) apply-patches-gravimetric $(MAKE) test-gravimetric-single $(MAKE) test-gravimetric-multi + $(MAKE) test-gravimetric-96 + $(MAKE) test-photometric -$(MAKE) remove-patches-gravimetric .PHONY: test-production-qc diff --git a/hardware-testing/hardware_testing/drivers/asair_sensor.py b/hardware-testing/hardware_testing/drivers/asair_sensor.py index be1d9937667..eb9a360eace 100644 --- a/hardware-testing/hardware_testing/drivers/asair_sensor.py +++ b/hardware-testing/hardware_testing/drivers/asair_sensor.py @@ -33,6 +33,7 @@ "08": "C492", "09": "C543", "10": "C74A", + "0A": "48d9", } @@ -65,6 +66,11 @@ def get_reading(self) -> Reading: """Get a temp and humidity reading.""" ... + @abc.abstractmethod + def get_serial(self) -> str: + """Read the device ID register.""" + ... + def BuildAsairSensor(simulate: bool) -> AsairSensorBase: """Try to find and return an Asair sensor, if not found return a simulator.""" @@ -144,8 +150,8 @@ def get_reading(self) -> Reading: log.debug(f"received {res}") res = codecs.encode(res, "hex") - temp = res[6:10] - relative_hum = res[10:14] + relative_hum = res[6:10] + temp = res[10:14] log.info(f"Temp: {temp}, RelativeHum: {relative_hum}") temp = float(int(temp, 16)) / 10 @@ -160,10 +166,40 @@ def get_reading(self) -> Reading: error_msg = "Asair Sensor not connected. Check if port number is correct." raise AsairSensorError(error_msg) + def get_serial(self) -> str: + """Read the device ID register.""" + serial_addr = "0A" + data_packet = "{}0300000002{}".format(serial_addr, addrs[serial_addr]) + log.debug(f"sending {data_packet}") + command_bytes = codecs.decode(data_packet.encode(), "hex") + try: + self._th_sensor.flushInput() + self._th_sensor.flushOutput() + self._th_sensor.write(command_bytes) + time.sleep(0.1) + + length = self._th_sensor.inWaiting() + res = self._th_sensor.read(length) + log.debug(f"received {res}") + dev_id = res[6:14] + return dev_id.decode() + + except (IndexError, ValueError) as e: + log.exception("Bad value read") + raise AsairSensorError(str(e)) + except SerialException: + log.exception("Communication error") + error_msg = "Asair Sensor not connected. Check if port number is correct." + raise AsairSensorError(error_msg) + class SimAsairSensor(AsairSensorBase): """Simulating Asair sensor driver.""" + def get_serial(self) -> str: + """Read the device ID register.""" + return "0102030405060708" + def get_reading(self) -> Reading: """Get a reading.""" temp = 25.0 diff --git a/hardware-testing/hardware_testing/gravimetric/README.md b/hardware-testing/hardware_testing/gravimetric/README.md index f1c1d08d610..2d3b571674c 100644 --- a/hardware-testing/hardware_testing/gravimetric/README.md +++ b/hardware-testing/hardware_testing/gravimetric/README.md @@ -18,4 +18,31 @@ and substitute `internal-release` for whatever branch you're merging in to. ## Photometric tests +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 96 --photometric --tip 50 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 96 --photometric --tip 200 + ## Gravimetric tests + +###P1000 single channel QC +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 1 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 1 --extra +###P1000 multi channel QC +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 8 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 8 --extra +###P1000 96 channel QC +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 96 +###P50 single channel QC +python3 -m hardware_testing.gravimetric --pipette 50 --channels 1 +python3 -m hardware_testing.gravimetric --pipette 50 --channels 1 --extra +###P50 multi channel QC +python3 -m hardware_testing.gravimetric --pipette 50 --channels 8 +python3 -m hardware_testing.gravimetric --pipette 50 --channels 8 --extra +###Increment tests +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 1 --increment --tip 50 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 1 --increment --tip 200 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 1 --increment --tip 1000 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 8 --increment --tip 50 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 8 --increment --tip 200 +python3 -m hardware_testing.gravimetric --pipette 1000 --channels 8 --increment --tip 1000 +python3 -m hardware_testing.gravimetric --pipette 50 --channels 1 --increment --tip 50 +python3 -m hardware_testing.gravimetric --pipette 50 --channels 8 --increment --tip 50 diff --git a/hardware-testing/hardware_testing/gravimetric/__main__.py b/hardware-testing/hardware_testing/gravimetric/__main__.py index 07609e72da4..f077339a606 100644 --- a/hardware-testing/hardware_testing/gravimetric/__main__.py +++ b/hardware-testing/hardware_testing/gravimetric/__main__.py @@ -2,18 +2,17 @@ from json import load as json_load from pathlib import Path import argparse -from typing import List, Union - +from typing import List, Union, Dict, Optional, Any, Tuple +from dataclasses import dataclass from opentrons.protocol_api import ProtocolContext +from . import report from hardware_testing.data import create_run_id_and_start_time, ui, get_git_description from hardware_testing.protocols import ( gravimetric_ot3_p50_single, - gravimetric_ot3_p50_multi_50ul_tip, + gravimetric_ot3_p50_multi, gravimetric_ot3_p1000_single, - gravimetric_ot3_p1000_multi_50ul_tip, - gravimetric_ot3_p1000_multi_200ul_tip, - gravimetric_ot3_p1000_multi_1000ul_tip, + gravimetric_ot3_p1000_multi, gravimetric_ot3_p1000_96_50ul_tip, gravimetric_ot3_p1000_96_200ul_tip, gravimetric_ot3_p1000_96_1000ul_tip, @@ -33,10 +32,13 @@ ConfigType, get_tip_volumes_for_qc, ) +from .measurement.record import GravimetricRecorder from .measurement import DELAY_FOR_MEASUREMENT -from .trial import TestResources +from .measurement.scale import Scale +from .trial import TestResources, _change_pipettes from .tips import get_tips from hardware_testing.drivers import asair_sensor +from opentrons.protocol_api import InstrumentContext # FIXME: bump to v2.15 to utilize protocol engine API_LEVEL = "2.13" @@ -46,25 +48,13 @@ # Keyed by pipette volume, channel count, and tip volume in that order GRAVIMETRIC_CFG = { 50: { - 1: {50: gravimetric_ot3_p50_single}, - 8: {50: gravimetric_ot3_p50_multi_50ul_tip}, + 1: gravimetric_ot3_p50_single, + 8: gravimetric_ot3_p50_multi, }, 1000: { - 1: { - 50: gravimetric_ot3_p1000_single, - 200: gravimetric_ot3_p1000_single, - 1000: gravimetric_ot3_p1000_single, - }, - 8: { - 50: gravimetric_ot3_p1000_multi_50ul_tip, - 200: gravimetric_ot3_p1000_multi_200ul_tip, - 1000: gravimetric_ot3_p1000_multi_1000ul_tip, - }, - 96: { - 50: gravimetric_ot3_p1000_96_50ul_tip, - 200: gravimetric_ot3_p1000_96_200ul_tip, - 1000: gravimetric_ot3_p1000_96_1000ul_tip, - }, + 1: gravimetric_ot3_p1000_single, + 8: gravimetric_ot3_p1000_multi, + 96: gravimetric_ot3_p1000_96_1000ul_tip, }, } @@ -98,12 +88,227 @@ } +@dataclass +class RunArgs: + """Common resources across multiple runs.""" + + tip_volumes: List[int] + volumes: List[Tuple[int, List[float]]] + run_id: str + pipette: InstrumentContext + pipette_tag: str + operator_name: str + git_description: str + robot_serial: str + tip_batchs: Dict[str, str] + recorder: Optional[GravimetricRecorder] + pipette_volume: int + pipette_channels: int + increment: bool + name: str + environment_sensor: asair_sensor.AsairSensorBase + trials: int + ctx: ProtocolContext + protocol_cfg: Any + test_report: report.CSVReport + + @classmethod + def _get_protocol_context(cls, args: argparse.Namespace) -> ProtocolContext: + if not args.simulate and not args.skip_labware_offsets: + # getting labware offsets must be done before creating the protocol context + # because it requires the robot-server to be running + ui.print_title("SETUP") + ui.print_info( + "Starting opentrons-robot-server, so we can http GET labware offsets" + ) + offsets = workarounds.http_get_all_labware_offsets() + ui.print_info(f"found {len(offsets)} offsets:") + for offset in offsets: + ui.print_info(f"\t{offset['createdAt']}:") + ui.print_info(f"\t\t{offset['definitionUri']}") + ui.print_info(f"\t\t{offset['vector']}") + LABWARE_OFFSETS.append(offset) + # gather the custom labware (for simulation) + custom_defs = {} + if args.simulate: + labware_dir = Path(__file__).parent.parent / "labware" + custom_def_uris = [ + "radwag_pipette_calibration_vial", + "opentrons_flex_96_tiprack_50ul_adp", + "opentrons_flex_96_tiprack_200ul_adp", + "opentrons_flex_96_tiprack_1000ul_adp", + ] + for def_uri in custom_def_uris: + with open(labware_dir / def_uri / "1.json", "r") as f: + custom_def = json_load(f) + custom_defs[def_uri] = custom_def + _ctx = helpers.get_api_context( + API_LEVEL, # type: ignore[attr-defined] + is_simulating=args.simulate, + deck_version="2", + extra_labware=custom_defs, + ) + return _ctx + + @classmethod + def build_run_args(cls, args: argparse.Namespace) -> "RunArgs": + """Build.""" + _ctx = RunArgs._get_protocol_context(args) + operator_name = helpers._get_operator_name(_ctx.is_simulating()) + robot_serial = helpers._get_robot_serial(_ctx.is_simulating()) + run_id, start_time = create_run_id_and_start_time() + environment_sensor = asair_sensor.BuildAsairSensor(_ctx.is_simulating()) + git_description = get_git_description() + if not args.photometric: + scale = Scale.build(simulate=_ctx.is_simulating()) + ui.print_header("LOAD PIPETTE") + pipette = helpers._load_pipette( + _ctx, + args.channels, + args.pipette, + "left", + args.increment, + args.gantry_speed if not args.photometric else None, + ) + pipette_tag = helpers._get_tag_from_pipette( + pipette, args.increment, args.user_volumes + ) + recorder: Optional[GravimetricRecorder] = None + kind = ConfigType.photometric if args.photometric else ConfigType.gravimetric + tip_batches: Dict[str, str] = {} + if args.tip == 0: + tip_volumes: List[int] = get_tip_volumes_for_qc( + args.pipette, args.channels, args.extra, args.photometric + ) + for tip in tip_volumes: + tip_batches[f"tips_{tip}ul"] = helpers._get_tip_batch( + _ctx.is_simulating(), tip + ) + else: + tip_volumes = [args.tip] + tip_batches[f"tips_{args.tip}ul"] = helpers._get_tip_batch( + _ctx.is_simulating(), args.tip + ) + + volumes: List[Tuple[int, List[float]]] = [] + for tip in tip_volumes: + vls = helpers._get_volumes( + _ctx, + args.increment, + args.pipette, + tip, + args.user_volumes, + kind, + False, # set extra to false so we always do the normal tests first + args.channels, + ) + if len(vls) > 0: + volumes.append( + ( + tip, + vls, + ) + ) + if args.extra: + # if we use extra, add those tests after + for tip in tip_volumes: + vls = helpers._get_volumes( + _ctx, + args.increment, + args.pipette, + tip, + args.user_volumes, + kind, + True, + args.channels, + ) + if len(vls) > 0: + volumes.append( + ( + tip, + vls, + ) + ) + if not volumes: + raise ValueError("no volumes to test, check the configuration") + volumes_list: List[float] = [] + for _, vls in volumes: + volumes_list.extend(vls) + + if args.trials == 0: + trials = helpers.get_default_trials(args.increment, kind, args.channels) + else: + trials = args.trials + + if args.photometric: + protocol_cfg = PHOTOMETRIC_CFG[args.tip] + name = protocol_cfg.metadata["protocolName"] # type: ignore[attr-defined] + report = execute_photometric.build_pm_report( + test_volumes=volumes_list, + run_id=run_id, + pipette_tag=pipette_tag, + operator_name=operator_name, + git_description=git_description, + tip_batches=tip_batches, + environment_sensor=environment_sensor, + trials=trials, + name=name, + robot_serial=robot_serial, + ) + else: + if args.increment: + protocol_cfg = GRAVIMETRIC_CFG_INCREMENT[args.pipette][args.channels][ + args.tip + ] + else: + protocol_cfg = GRAVIMETRIC_CFG[args.pipette][args.channels] + name = protocol_cfg.metadata["protocolName"] # type: ignore[attr-defined] + recorder = execute._load_scale( + name, scale, run_id, pipette_tag, start_time, _ctx.is_simulating() + ) + + report = execute.build_gm_report( + test_volumes=volumes_list, + run_id=run_id, + pipette_tag=pipette_tag, + operator_name=operator_name, + git_description=git_description, + robot_serial=robot_serial, + tip_batchs=tip_batches, + recorder=recorder, + pipette_channels=args.channels, + increment=args.increment, + name=name, + environment_sensor=environment_sensor, + trials=trials, + ) + + return RunArgs( + tip_volumes=tip_volumes, + volumes=volumes, + run_id=run_id, + pipette=pipette, + pipette_tag=pipette_tag, + operator_name=operator_name, + git_description=git_description, + robot_serial=robot_serial, + tip_batchs=tip_batches, + recorder=recorder, + pipette_volume=args.pipette, + pipette_channels=args.channels, + increment=args.increment, + name=name, + environment_sensor=environment_sensor, + trials=trials, + ctx=_ctx, + protocol_cfg=protocol_cfg, + test_report=report, + ) + + def build_gravimetric_cfg( protocol: ProtocolContext, - pipette_volume: int, - pipette_channels: int, tip_volume: int, - trials: int, increment: bool, return_tip: bool, blank: bool, @@ -114,25 +319,21 @@ def build_gravimetric_cfg( scale_delay: int, isolate_channels: List[int], extra: bool, + jog: bool, + run_args: RunArgs, ) -> GravimetricConfig: - """Run.""" - if increment: - protocol_cfg = GRAVIMETRIC_CFG_INCREMENT[pipette_volume][pipette_channels][ - tip_volume - ] - else: - protocol_cfg = GRAVIMETRIC_CFG[pipette_volume][pipette_channels][tip_volume] + """Build.""" return GravimetricConfig( - name=protocol_cfg.metadata["protocolName"], # type: ignore[attr-defined] + name=run_args.name, pipette_mount="left", - pipette_volume=pipette_volume, - pipette_channels=pipette_channels, + pipette_volume=run_args.pipette_volume, + pipette_channels=run_args.pipette_channels, tip_volume=tip_volume, - trials=trials, + trials=run_args.trials, labware_offsets=LABWARE_OFFSETS, - labware_on_scale=protocol_cfg.LABWARE_ON_SCALE, # type: ignore[attr-defined] - slot_scale=protocol_cfg.SLOT_SCALE, # type: ignore[attr-defined] - slots_tiprack=protocol_cfg.SLOTS_TIPRACK[tip_volume], # type: ignore[attr-defined] + labware_on_scale=run_args.protocol_cfg.LABWARE_ON_SCALE, # type: ignore[attr-defined] + slot_scale=run_args.protocol_cfg.SLOT_SCALE, # type: ignore[attr-defined] + slots_tiprack=run_args.protocol_cfg.SLOTS_TIPRACK[tip_volume], # type: ignore[attr-defined] increment=increment, return_tip=return_tip, blank=blank, @@ -143,15 +344,14 @@ def build_gravimetric_cfg( scale_delay=scale_delay, isolate_channels=isolate_channels, kind=ConfigType.gravimetric, - extra=args.extra, + extra=extra, + jog=jog, ) def build_photometric_cfg( protocol: ProtocolContext, - pipette_volume: int, tip_volume: int, - trials: int, return_tip: bool, mix: bool, inspect: bool, @@ -159,23 +359,24 @@ def build_photometric_cfg( touch_tip: bool, refill: bool, extra: bool, + jog: bool, + run_args: RunArgs, ) -> PhotometricConfig: """Run.""" - protocol_cfg = PHOTOMETRIC_CFG[tip_volume] return PhotometricConfig( - name=protocol_cfg.metadata["protocolName"], # type: ignore[attr-defined] + name=run_args.name, pipette_mount="left", - pipette_volume=pipette_volume, + pipette_volume=run_args.pipette_volume, pipette_channels=96, increment=False, tip_volume=tip_volume, - trials=trials, + trials=run_args.trials, labware_offsets=LABWARE_OFFSETS, - photoplate=protocol_cfg.PHOTOPLATE_LABWARE, # type: ignore[attr-defined] - photoplate_slot=protocol_cfg.SLOT_PLATE, # type: ignore[attr-defined] - reservoir=protocol_cfg.RESERVOIR_LABWARE, # type: ignore[attr-defined] - reservoir_slot=protocol_cfg.SLOT_RESERVOIR, # type: ignore[attr-defined] - slots_tiprack=protocol_cfg.SLOTS_TIPRACK[tip_volume], # type: ignore[attr-defined] + photoplate=run_args.protocol_cfg.PHOTOPLATE_LABWARE, # type: ignore[attr-defined] + photoplate_slot=run_args.protocol_cfg.SLOT_PLATE, # type: ignore[attr-defined] + reservoir=run_args.protocol_cfg.RESERVOIR_LABWARE, # type: ignore[attr-defined] + reservoir_slot=run_args.protocol_cfg.SLOT_RESERVOIR, # type: ignore[attr-defined] + slots_tiprack=run_args.protocol_cfg.SLOTS_TIPRACK[tip_volume], # type: ignore[attr-defined] return_tip=return_tip, mix=mix, inspect=inspect, @@ -183,18 +384,22 @@ def build_photometric_cfg( touch_tip=touch_tip, refill=refill, kind=ConfigType.photometric, - extra=args.extra, + extra=extra, + jog=jog, ) -def _main(args: argparse.Namespace, _ctx: ProtocolContext) -> None: +def _main( + args: argparse.Namespace, + run_args: RunArgs, + tip: int, + volumes: List[float], +) -> None: union_cfg: Union[PhotometricConfig, GravimetricConfig] if args.photometric: cfg_pm: PhotometricConfig = build_photometric_cfg( - _ctx, - args.pipette, - args.tip, - args.trials, + run_args.ctx, + tip, args.return_tip, args.mix, args.inspect, @@ -202,17 +407,14 @@ def _main(args: argparse.Namespace, _ctx: ProtocolContext) -> None: args.touch_tip, args.refill, args.extra, + args.jog, + run_args, ) - if args.trials == 0: - cfg_pm.trials = helpers.get_default_trials(cfg_pm) union_cfg = cfg_pm else: cfg_gm: GravimetricConfig = build_gravimetric_cfg( - _ctx, - args.pipette, - args.channels, - args.tip, - args.trials, + run_args.ctx, + tip, args.increment, args.return_tip, False if args.no_blank else True, @@ -223,42 +425,40 @@ def _main(args: argparse.Namespace, _ctx: ProtocolContext) -> None: args.scale_delay, args.isolate_channels if args.isolate_channels else [], args.extra, + args.jog, + run_args, ) - if args.trials == 0: - cfg_gm.trials = helpers.get_default_trials(cfg_gm) + union_cfg = cfg_gm - run_id, start_time = create_run_id_and_start_time() - ui.print_header("LOAD PIPETTE") - pipette = helpers._load_pipette(_ctx, union_cfg) ui.print_header("GET PARAMETERS") - test_volumes = helpers._get_volumes(_ctx, union_cfg) - for v in test_volumes: + + for v in volumes: ui.print_info(f"\t{v} uL") all_channels_same_time = ( getattr(union_cfg, "increment", False) or union_cfg.pipette_channels == 96 ) - run_args = TestResources( - ctx=_ctx, - pipette=pipette, - pipette_tag=helpers._get_tag_from_pipette(pipette, union_cfg), + test_resources = TestResources( + ctx=run_args.ctx, + pipette=run_args.pipette, tipracks=helpers._load_tipracks( - _ctx, union_cfg, use_adapters=args.channels == 96 + run_args.ctx, union_cfg, use_adapters=args.channels == 96 + ), + test_volumes=volumes, + tips=get_tips( + run_args.ctx, + run_args.pipette, + tip, + all_channels=all_channels_same_time, ), - test_volumes=test_volumes, - run_id=run_id, - start_time=start_time, - operator_name=helpers._get_operator_name(_ctx.is_simulating()), - robot_serial=helpers._get_robot_serial(_ctx.is_simulating()), - tip_batch=helpers._get_tip_batch(_ctx.is_simulating()), - git_description=get_git_description(), - tips=get_tips(_ctx, pipette, args.tip, all_channels=all_channels_same_time), - env_sensor=asair_sensor.BuildAsairSensor(_ctx.is_simulating()), + env_sensor=run_args.environment_sensor, + recorder=run_args.recorder, + test_report=run_args.test_report, ) if args.photometric: - execute_photometric.run(cfg_pm, run_args) + execute_photometric.run(cfg_pm, test_resources) else: - execute.run(cfg_gm, run_args) + execute.run(cfg_gm, test_resources) if __name__ == "__main__": @@ -282,49 +482,19 @@ def _main(args: argparse.Namespace, _ctx: ProtocolContext) -> None: parser.add_argument("--refill", action="store_true") parser.add_argument("--isolate-channels", nargs="+", type=int, default=None) parser.add_argument("--extra", action="store_true") + parser.add_argument("--jog", action="store_true") args = parser.parse_args() - if not args.simulate and not args.skip_labware_offsets: - # getting labware offsets must be done before creating the protocol context - # because it requires the robot-server to be running - ui.print_title("SETUP") - ui.print_info( - "Starting opentrons-robot-server, so we can http GET labware offsets" - ) - offsets = workarounds.http_get_all_labware_offsets() - ui.print_info(f"found {len(offsets)} offsets:") - for offset in offsets: - ui.print_info(f"\t{offset['createdAt']}:") - ui.print_info(f"\t\t{offset['definitionUri']}") - ui.print_info(f"\t\t{offset['vector']}") - LABWARE_OFFSETS.append(offset) - # gather the custom labware (for simulation) - custom_defs = {} - if args.simulate: - labware_dir = Path(__file__).parent.parent / "labware" - custom_def_uris = [ - "radwag_pipette_calibration_vial", - "opentrons_flex_96_tiprack_50ul_adp", - "opentrons_flex_96_tiprack_200ul_adp", - "opentrons_flex_96_tiprack_1000ul_adp", - ] - for def_uri in custom_def_uris: - with open(labware_dir / def_uri / "1.json", "r") as f: - custom_def = json_load(f) - custom_defs[def_uri] = custom_def - _ctx = helpers.get_api_context( - API_LEVEL, # type: ignore[attr-defined] - is_simulating=args.simulate, - deck_version="2", - extra_labware=custom_defs, - ) - if args.tip == 0: - for tip in get_tip_volumes_for_qc( - args.pipette, args.channels, args.extra, args.photometric - ): - hw = _ctx._core.get_hardware() - if not _ctx.is_simulating(): - ui.alert_user_ready(f"Ready to run with {tip}ul tip?", hw) - args.tip = tip - _main(args, _ctx) - else: - _main(args, _ctx) + run_args = RunArgs.build_run_args(args) + try: + if not run_args.ctx.is_simulating(): + ui.get_user_ready("CLOSE the door, and MOVE AWAY from machine") + for tip, volumes in run_args.volumes: + hw = run_args.ctx._core.get_hardware() + _main(args, run_args, tip, volumes) + finally: + if run_args.recorder is not None: + ui.print_info("ending recording") + run_args.recorder.stop() + run_args.recorder.deactivate() + _change_pipettes(run_args.ctx, run_args.pipette) + print("done\n\n") diff --git a/hardware-testing/hardware_testing/gravimetric/config.py b/hardware-testing/hardware_testing/gravimetric/config.py index f5cbc3ba8e8..17cb041f37d 100644 --- a/hardware-testing/hardware_testing/gravimetric/config.py +++ b/hardware-testing/hardware_testing/gravimetric/config.py @@ -1,8 +1,10 @@ """Config.""" from dataclasses import dataclass -from typing import List, Dict +from typing import List, Dict, Tuple from typing_extensions import Final from enum import Enum +from opentrons.config.types import LiquidProbeSettings +from opentrons.protocol_api.labware import Well class ConfigType(Enum): @@ -31,6 +33,7 @@ class VolumetricConfig: user_volumes: bool kind: ConfigType extra: bool + jog: bool @dataclass @@ -77,82 +80,198 @@ class PhotometricConfig(VolumetricConfig): VIAL_SAFE_Z_OFFSET: Final = 25 LABWARE_BOTTOM_CLEARANCE = 1.5 - -QC_VOLUMES_G: Dict[int, Dict[int, Dict[int, List[float]]]] = { - 1: { - 50: { # P50 - 50: [1.0, 50.0], # T50 +LIQUID_PROBE_SETTINGS: Dict[int, Dict[int, Dict[int, Dict[str, int]]]] = { + 50: { + 1: { + 50: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 11, + "plunger_speed": 21, + "sensor_threshold_pascals": 150, + }, }, - 1000: { # P1000 - 50: [5.0], # T50 - 200: [], # T200 - 1000: [1000.0], # T1000 + 8: { + 50: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 11, + "plunger_speed": 21, + "sensor_threshold_pascals": 150, + }, }, }, - 8: { - 50: { # P50 - 50: [1.0, 50.0], # T50 + 1000: { + 1: { + 50: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 200: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 1000: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 11, + "sensor_threshold_pascals": 150, + }, }, - 1000: { # P1000 - 50: [5.0], # T50 - 200: [], # T200 - 1000: [1000.0], # T1000 + 8: { + 50: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 200: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 1000: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 11, + "sensor_threshold_pascals": 150, + }, + }, + 96: { + 50: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 200: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 10, + "sensor_threshold_pascals": 200, + }, + 1000: { + "max_z_distance": 20, + "min_z_distance": 5, + "mount_speed": 5, + "plunger_speed": 11, + "sensor_threshold_pascals": 150, + }, }, }, +} + + +def _get_liquid_probe_settings( + cfg: VolumetricConfig, well: Well +) -> LiquidProbeSettings: + lqid_cfg: Dict[str, int] = LIQUID_PROBE_SETTINGS[cfg.pipette_volume][ + cfg.pipette_channels + ][cfg.tip_volume] + return LiquidProbeSettings( + starting_mount_height=well.top().point.z, + max_z_distance=min(well.depth, lqid_cfg["max_z_distance"]), + min_z_distance=lqid_cfg["min_z_distance"], + mount_speed=lqid_cfg["mount_speed"], + plunger_speed=lqid_cfg["plunger_speed"], + sensor_threshold_pascals=lqid_cfg["sensor_threshold_pascals"], + expected_liquid_height=110, + log_pressure=True, + aspirate_while_sensing=False, + auto_zero_sensor=True, + num_baseline_reads=10, + data_file="/var/pressure_sensor_data.csv", + ) + + +QC_VOLUMES_G: Dict[int, Dict[int, List[Tuple[int, List[float]]]]] = { + 1: { + 50: [ # P50 + (50, [1.0, 50.0]), # T50 + ], + 1000: [ # P1000 + (50, [5.0]), # T50 + (200, []), # T200 + (1000, [1000.0]), # T1000 + ], + }, + 8: { + 50: [ # P50 + (50, [1.0, 50.0]), # T50 + ], + 1000: [ # P1000 + (50, [5.0]), # T50 + (200, []), # T200 + (1000, [1000.0]), # T1000 + ], + }, 96: { - 1000: { # P1000 - 50: [], # T50 - 200: [], # T200 - 1000: [1000.0], # T1000 - }, + 1000: [ # P1000 + (50, []), # T50 + (200, []), # T200 + (1000, [1000.0]), # T1000 + ], }, } -QC_VOLUMES_EXTRA_G: Dict[int, Dict[int, Dict[int, List[float]]]] = { +QC_VOLUMES_EXTRA_G: Dict[int, Dict[int, List[Tuple[int, List[float]]]]] = { 1: { - 50: { # P50 - 50: [1.0, 10.0, 50.0], # T50 - }, - 1000: { # P1000 - 50: [5.0, 50], # T50 - 200: [200.0], # T200 - 1000: [1000.0], # T1000 - }, + 50: [ # P50 + (50, [10.0]), # T50 + ], + 1000: [ # P1000 + (50, [50]), # T50 + (200, [200.0]), # T200 + (1000, []), # T1000 + ], }, 8: { - 50: { # P50 - 50: [1.0, 10.0, 50.0], # T50 - }, - 1000: { # P1000 - 50: [5.0, 50], # T50 - 200: [200.0], # T200 - 1000: [1000.0], # T1000 - }, + 50: [ # P50 + (50, [10.0]), # T50 + ], + 1000: [ # P1000 + (50, [50.0]), # T50 + (200, [200.0]), # T200 + (1000, []), # T1000 + ], }, 96: { - 1000: { # P1000 - 50: [], # T50 - 200: [], # T200 - 1000: [1000.0], # T1000 - }, + 1000: [ # P1000 + (50, []), # T50 + (200, []), # T200 + (1000, []), # T1000 + ], }, } -QC_VOLUMES_P: Dict[int, Dict[int, Dict[int, List[float]]]] = { +QC_VOLUMES_P: Dict[int, Dict[int, List[Tuple[int, List[float]]]]] = { 96: { - 1000: { # P1000 - 50: [5.0], # T50 - 200: [200.0], # T200 - 1000: [], # T1000 - }, + 1000: [ # P1000 + (50, [5.0]), # T50 + (200, [200.0]), # T200 + (1000, []), # T1000 + ], }, } QC_DEFAULT_TRIALS: Dict[ConfigType, Dict[int, int]] = { ConfigType.gravimetric: { 1: 10, - 8: 8, + 8: 10, 96: 9, }, ConfigType.photometric: { @@ -160,23 +279,103 @@ class PhotometricConfig(VolumetricConfig): }, } +QC_TEST_SAFETY_FACTOR = 0.0 + +QC_TEST_MIN_REQUIREMENTS: Dict[ + int, Dict[int, Dict[int, Dict[float, Tuple[float, float]]]] +] = { + # channels: [Pipette: [tip: [Volume: (%d, Cv)]]] + 1: { + 50: { # P50 + 50: { + 1.0: (5.0, 4.0), + 10.0: (1.0, 0.5), + 50.0: (1, 0.4), + }, + }, # T50 + 1000: { # P1000 + 50: { # T50 + 5.0: (5.0, 5.0), + 10.0: (2.0, 2.0), + 50.0: (1.0, 1.0), + }, + 200: { # T200 + 5.0: (7.0, 4.00), + 50.0: (2.0, 1.0), + 200.0: (0.5, 0.2), + }, + 1000: { # T1000 + 10.0: (7.5, 3.5), + 100.0: (2.0, 0.75), + 1000.0: (0.7, 0.15), + }, + }, + }, + 8: { + 50: { # P50 + 50: { # T50 + 1.0: (20.0, 5.0), + 10.0: (3.0, 2.0), + 50.0: (1.25, 0.4), + }, + }, + 1000: { # P1000 + 50: { # T50 + 5.0: (5.0, 5.0), + 10.0: (1.5, 1.5), + 50.0: (1.0, 1.0), + }, + 200: { # T200 + 5.0: (5.0, 5.0), + 50.0: (1.5, 1.5), + 200.0: (1.0, 0.4), + }, + 1000: { # T1000 + 10.0: (10.0, 5.0), + 100.0: (2.5, 1.0), + 1000.0: (0.7, 0.15), + }, + }, + }, + 96: { + 1000: { # P1000 + 50: { # T50 + 5.0: (2.5, 2.0), + 10.0: (3.1, 1.7), + 50.0: (1.5, 0.75), + }, + 200: { # T200 + 5.0: (2.5, 4.0), + 50.0: (1.5, 2.0), + 200.0: (1.4, 0.9), + }, + 1000: { # T1000 + 10.0: (5.0, 5.0), + 100.0: (2.5, 1.5), + 1000.0: (1.0, 0.75), + }, + }, + }, +} + def get_tip_volumes_for_qc( pipette_volume: int, pipette_channels: int, extra: bool, photometric: bool ) -> List[int]: """Build the default testing volumes for qc.""" - config: Dict[int, Dict[int, Dict[int, List[float]]]] = {} + config: Dict[int, Dict[int, List[Tuple[int, List[float]]]]] = {} + tip_volumes: List[int] = [] if photometric: config = QC_VOLUMES_P else: - if extra: - config = QC_VOLUMES_EXTRA_G - else: - config = QC_VOLUMES_G - tip_volumes = [ - t - for t in config[pipette_channels][pipette_volume].keys() - if len(config[pipette_channels][pipette_volume][t]) > 0 - ] + config = QC_VOLUMES_G + for t, vls in config[pipette_channels][pipette_volume]: + if len(vls) > 0 and t not in tip_volumes: + tip_volumes.append(t) + if extra: + for t, vls in QC_VOLUMES_EXTRA_G[pipette_channels][pipette_volume]: + if len(vls) > 0 and t not in tip_volumes: + tip_volumes.append(t) + assert len(tip_volumes) > 0 return tip_volumes diff --git a/hardware-testing/hardware_testing/gravimetric/execute.py b/hardware-testing/hardware_testing/gravimetric/execute.py index ce98ee66bb7..c9aee41c06c 100644 --- a/hardware-testing/hardware_testing/gravimetric/execute.py +++ b/hardware-testing/hardware_testing/gravimetric/execute.py @@ -7,6 +7,7 @@ from hardware_testing.data import ui from hardware_testing.data.csv_report import CSVReport from hardware_testing.opentrons_api.types import Point, OT3Mount +from hardware_testing.drivers import asair_sensor from . import report from . import config @@ -15,6 +16,7 @@ _get_channel_offset, _calculate_average, _jog_to_find_liquid_height, + _sense_liquid_height, _apply_labware_offsets, _pick_up_tip, _drop_tip, @@ -42,7 +44,9 @@ from .measurement.record import ( GravimetricRecorder, GravimetricRecorderConfig, + GravimetricRecording, ) +from .measurement.scale import Scale from .tips import MULTI_CHANNEL_TEST_ORDER @@ -191,9 +195,10 @@ def _run_trial( ) def _tag(m_type: MeasurementType) -> str: - return create_measurement_tag( + tag = create_measurement_tag( m_type, None if trial.blank else trial.volume, trial.channel, trial.trial ) + return tag def _record_measurement_and_store(m_type: MeasurementType) -> MeasurementData: m_tag = _tag(m_type) @@ -302,32 +307,47 @@ def _get_channel_divider(cfg: config.GravimetricConfig) -> float: def build_gm_report( - cfg: config.GravimetricConfig, - resources: TestResources, + test_volumes: List[float], + run_id: str, + pipette_tag: str, + operator_name: str, + git_description: str, + robot_serial: str, + tip_batchs: Dict[str, str], recorder: GravimetricRecorder, + pipette_channels: int, + increment: bool, + name: str, + environment_sensor: asair_sensor.AsairSensorBase, + trials: int, ) -> report.CSVReport: """Build a CSVReport formated for gravimetric tests.""" ui.print_header("CREATE TEST-REPORT") test_report = report.create_csv_test_report( - resources.test_volumes, cfg, run_id=resources.run_id + test_volumes, pipette_channels, increment, trials, name, run_id=run_id ) - test_report.set_tag(resources.pipette_tag) - test_report.set_operator(resources.operator_name) - test_report.set_version(resources.git_description) + test_report.set_tag(pipette_tag) + test_report.set_operator(operator_name) + test_report.set_version(git_description) report.store_serial_numbers( test_report, - robot=resources.robot_serial, - pipette=resources.pipette_tag, - tips=resources.tip_batch, + robot=robot_serial, + pipette=pipette_tag, + tips=tip_batchs, scale=recorder.serial_number, - environment="None", + environment=environment_sensor.get_serial(), liquid="None", ) return test_report def _load_scale( - cfg: config.GravimetricConfig, resources: TestResources + name: str, + scale: Scale, + run_id: str, + pipette_tag: str, + start_time: float, + simulating: bool, ) -> GravimetricRecorder: ui.print_header("LOAD SCALE") ui.print_info( @@ -338,20 +358,20 @@ def _load_scale( ) recorder = GravimetricRecorder( GravimetricRecorderConfig( - test_name=cfg.name, - run_id=resources.run_id, - tag=resources.pipette_tag, - start_time=resources.start_time, + test_name=name, + run_id=run_id, + tag=pipette_tag, + start_time=start_time, duration=0, - frequency=1000 if resources.ctx.is_simulating() else 5, + frequency=1000 if simulating else 5, stable=False, ), - simulate=resources.ctx.is_simulating(), + scale, + simulate=simulating, ) ui.print_info(f'found scale "{recorder.serial_number}"') - if resources.ctx.is_simulating(): - start_sim_mass = {50: 15, 200: 200, 1000: 200} - recorder.set_simulation_mass(start_sim_mass[cfg.tip_volume]) + if simulating: + recorder.set_simulation_mass(0) recorder.record(in_thread=True) ui.print_info(f'scale is recording to "{recorder.file_name}"') return recorder @@ -417,6 +437,22 @@ def _calculate_evaporation( return average_aspirate_evaporation_ul, average_dispense_evaporation_ul +def _get_liquid_height( + resources: TestResources, cfg: config.GravimetricConfig, well: Well +) -> float: + resources.pipette.move_to(well.top(0), minimum_z_height=_minimum_z_height(cfg)) + if cfg.jog: + _liquid_height = _jog_to_find_liquid_height( + resources.ctx, resources.pipette, well + ) + else: + _liquid_height = _sense_liquid_height( + resources.ctx, resources.pipette, well, cfg + ) + resources.pipette.move_to(well.top().move(Point(0, 0, _minimum_z_height(cfg)))) + return _liquid_height + + def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: """Run.""" global _PREV_TRIAL_GRAMS @@ -439,9 +475,13 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: raise ValueError(f"more trials ({trial_total}) than tips ({total_tips})") elif not resources.ctx.is_simulating(): ui.get_user_ready(f"prepare {trial_total - total_tips} extra tip-racks") - recorder = _load_scale(cfg, resources) - test_report = build_gm_report(cfg, resources, recorder) - + assert resources.recorder is not None + recorder = resources.recorder + if resources.ctx.is_simulating(): + start_sim_mass = {50: 15, 200: 200, 1000: 200} + resources.recorder.set_simulation_mass(start_sim_mass[cfg.tip_volume]) + recorder._recording = GravimetricRecording() + report.store_config_gm(resources.test_report, cfg) calibration_tip_in_use = True if resources.ctx.is_simulating(): @@ -452,22 +492,15 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: ui.print_info("homing...") resources.ctx.home() resources.pipette.home_plunger() - first_tip = resources.tips[0][0] + first_tip = _next_tip_for_channel(cfg, resources, 0, total_tips) 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) - if not resources.ctx.is_simulating(): - ui.get_user_ready("REPLACE first tip with NEW TIP") - ui.get_user_ready("CLOSE the door, and MOVE AWAY from machine") ui.print_info("moving to scale") well = labware_on_scale["A1"] - resources.pipette.move_to(well.top(0), minimum_z_height=_minimum_z_height(cfg)) - _liquid_height = _jog_to_find_liquid_height( - resources.ctx, resources.pipette, well - ) - resources.pipette.move_to(well.top().move(Point(0, 0, _minimum_z_height(cfg)))) + _liquid_height = _get_liquid_height(resources, cfg, well) height_below_top = well.depth - _liquid_height ui.print_info(f"liquid is {height_below_top} mm below top of vial") liquid_tracker.set_start_volume_from_liquid_height( @@ -489,7 +522,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: resources, recorder, liquid_tracker, - test_report, + resources.test_report, labware_on_scale, ) @@ -507,7 +540,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: resources.test_volumes, channels_to_test, recorder, - test_report, + resources.test_report, liquid_tracker, False, resources.env_sensor, @@ -579,7 +612,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: dispense_data_list.append(dispense_data) report.store_trial( - test_report, + resources.test_report, run_trial.trial, run_trial.volume, run_trial.channel, @@ -616,7 +649,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: _print_stats("dispense", dispense_average, dispense_cv, dispense_d) report.store_volume_per_channel( - report=test_report, + report=resources.test_report, mode="aspirate", volume=volume, channel=channel, @@ -627,7 +660,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: humidity=aspirate_humidity_avg, ) report.store_volume_per_channel( - report=test_report, + report=resources.test_report, mode="dispense", volume=volume, channel=channel, @@ -640,6 +673,23 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: actual_asp_list_all.extend(actual_asp_list_channel) actual_disp_list_all.extend(actual_disp_list_channel) + acceptable_cv = trials[volume][channel][0].acceptable_cv + acceptable_d = trials[volume][channel][0].acceptable_d + print(f"acceptable cv {acceptable_cv} acceptable_d {acceptable_d}") + print(f"dispense cv {dispense_cv} aspirate_cv {aspirate_cv}") + print(f"dispense d {dispense_cv} aspirate_d {aspirate_d}") + if acceptable_cv is not None and acceptable_d is not None: + acceptable_cv /= 100 + acceptable_d /= 100 + if ( + dispense_cv > acceptable_cv + or aspirate_cv > acceptable_cv + or aspirate_d > acceptable_d + or dispense_d > acceptable_d + ): + raise RuntimeError( + f"Trial with volume {volume} on channel {channel} did not pass spec" + ) for trial in range(cfg.trials): trial_asp_list = trial_asp_dict[trial] trial_disp_list = trial_disp_dict[trial] @@ -652,7 +702,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: ) report.store_volume_per_trial( - report=test_report, + report=resources.test_report, mode="aspirate", volume=volume, trial=trial, @@ -661,7 +711,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: d=aspirate_d, ) report.store_volume_per_trial( - report=test_report, + report=resources.test_report, mode="dispense", volume=volume, trial=trial, @@ -682,7 +732,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: _print_stats("dispense", dispense_average, dispense_cv, dispense_d) report.store_volume_all( - report=test_report, + report=resources.test_report, mode="aspirate", volume=volume, average=aspirate_average, @@ -690,7 +740,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: d=aspirate_d, ) report.store_volume_all( - report=test_report, + report=resources.test_report, mode="dispense", volume=volume, average=dispense_average, @@ -698,14 +748,11 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None: d=dispense_d, ) finally: - ui.print_info("ending recording") - recorder.stop() - recorder.deactivate() _return_tip = False if calibration_tip_in_use else cfg.return_tip _finish_test(cfg, resources, _return_tip) ui.print_title("RESULTS") _print_final_results( volumes=resources.test_volumes, channel_count=len(channels_to_test), - test_report=test_report, + test_report=resources.test_report, ) diff --git a/hardware-testing/hardware_testing/gravimetric/execute_photometric.py b/hardware-testing/hardware_testing/gravimetric/execute_photometric.py index 8e68e36fea4..cd4f517a26e 100644 --- a/hardware-testing/hardware_testing/gravimetric/execute_photometric.py +++ b/hardware-testing/hardware_testing/gravimetric/execute_photometric.py @@ -11,6 +11,7 @@ create_measurement_tag, EnvironmentData, ) +from hardware_testing.drivers import asair_sensor from .measurement.environment import read_environment_data from . import report from . import config @@ -141,6 +142,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> EnvironmentData: # what volumes need to be added between trials. ui.get_user_ready("check DYE is enough") + ui.print_info(f"aspirating from {trial.source}") _record_measurement_and_store(MeasurementType.INIT) trial.pipette.move_to(location=trial.source.top(), minimum_z_height=133) # RUN ASPIRATE @@ -166,7 +168,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> EnvironmentData: for w in trial.dest.wells(): trial.liquid_tracker.set_start_volume(w, photoplate_preped_vol) trial.pipette.move_to(trial.dest["A1"].top()) - + ui.print_info(f"dispensing to {trial.dest}") # RUN DISPENSE dispense_with_liquid_class( trial.ctx, @@ -241,22 +243,31 @@ def _ul_to_ml(x: float) -> float: def build_pm_report( - cfg: config.PhotometricConfig, resources: TestResources + test_volumes: List[float], + run_id: str, + pipette_tag: str, + operator_name: str, + git_description: str, + tip_batches: Dict[str, str], + environment_sensor: asair_sensor.AsairSensorBase, + trials: int, + name: str, + robot_serial: str, ) -> report.CSVReport: """Build a CSVReport formated for photometric tests.""" ui.print_header("CREATE TEST-REPORT") test_report = report.create_csv_test_report_photometric( - resources.test_volumes, cfg, run_id=resources.run_id + test_volumes, trials, name, run_id ) - test_report.set_tag(resources.pipette_tag) - test_report.set_operator(resources.operator_name) - test_report.set_version(resources.git_description) + test_report.set_tag(pipette_tag) + test_report.set_operator(operator_name) + test_report.set_version(git_description) report.store_serial_numbers_pm( test_report, - robot=resources.robot_serial, - pipette=resources.pipette_tag, - tips=resources.tip_batch, - environment="None", + robot=robot_serial, + pipette=pipette_tag, + tips=tip_batches, + environment=environment_sensor.get_serial(), liquid="None", ) return test_report @@ -382,14 +393,12 @@ def run(cfg: config.PhotometricConfig, resources: TestResources) -> None: trial_total <= total_tips ), f"more trials ({trial_total}) than tips ({total_tips})" - test_report = build_pm_report(cfg, resources) - _display_dye_information(cfg, resources) _find_liquid_height(cfg, resources, liquid_tracker, reservoir["A1"]) trials = build_photometric_trials( resources.ctx, - test_report, + resources.test_report, resources.pipette, reservoir["A1"], photoplate, diff --git a/hardware-testing/hardware_testing/gravimetric/helpers.py b/hardware-testing/hardware_testing/gravimetric/helpers.py index 03594dde337..d3a0fa2f740 100644 --- a/hardware-testing/hardware_testing/gravimetric/helpers.py +++ b/hardware-testing/hardware_testing/gravimetric/helpers.py @@ -141,19 +141,35 @@ def _jog_to_find_liquid_height( return _liquid_height +def _sense_liquid_height( + ctx: ProtocolContext, + pipette: InstrumentContext, + well: Well, + cfg: config.VolumetricConfig, +) -> float: + if ctx.is_simulating(): + return well.depth - 1 + hwapi = get_sync_hw_api(ctx) + pipette.move_to(well.top()) + lps = config._get_liquid_probe_settings(cfg, well) + height = well.top().point.z - hwapi.liquid_probe(OT3Mount.LEFT, lps) + depth = well.depth - height + return depth + + def _calculate_average(volume_list: List[float]) -> float: return sum(volume_list) / len(volume_list) def _reduce_volumes_to_not_exceed_software_limit( test_volumes: List[float], - cfg: config.VolumetricConfig, + pipette_volume: int, + pipette_channels: int, + tip_volume: int, ) -> List[float]: for i, v in enumerate(test_volumes): - liq_cls = get_liquid_class( - cfg.pipette_volume, cfg.pipette_channels, cfg.tip_volume, int(v) - ) - max_vol = cfg.tip_volume - liq_cls.aspirate.trailing_air_gap + liq_cls = get_liquid_class(pipette_volume, pipette_channels, tip_volume, int(v)) + max_vol = tip_volume - liq_cls.aspirate.trailing_air_gap test_volumes[i] = min(v, max_vol - 0.1) return test_volumes @@ -208,9 +224,9 @@ def _calculate_stats( return average, cv, d -def _get_tip_batch(is_simulating: bool) -> str: +def _get_tip_batch(is_simulating: bool, tip: int) -> str: if not is_simulating: - return input("TIP BATCH:").strip() + return input(f"TIP BATCH for {tip}ul tips:").strip() else: return "simulation-tip-batch" @@ -266,71 +282,89 @@ def _drop_tip( pipette.move_to(cur_location.move(Point(0, 0, minimum_z_height))) -def _get_volumes(ctx: ProtocolContext, cfg: config.VolumetricConfig) -> List[float]: - if cfg.increment: - test_volumes = get_volume_increments(cfg.pipette_volume, cfg.tip_volume) - elif cfg.user_volumes and not ctx.is_simulating(): - _inp = input('Enter desired volumes, comma separated (eg: "10,100,1000") :') +def _get_volumes( + ctx: ProtocolContext, + increment: bool, + pipette_volume: int, + tip_volume: int, + user_volumes: bool, + kind: config.ConfigType, + extra: bool, + channels: int, +) -> List[float]: + if increment: + print("if") + test_volumes = get_volume_increments(pipette_volume, tip_volume) + elif user_volumes and not ctx.is_simulating(): + print("elif") + _inp = input( + f'Enter desired volumes for tip{tip_volume}, comma separated (eg: "10,100,1000") :' + ) test_volumes = [ float(vol_str) for vol_str in _inp.strip().split(",") if vol_str ] else: - test_volumes = get_test_volumes(cfg) - if not test_volumes: - raise ValueError("no volumes to test, check the configuration") + print("else") + test_volumes = get_test_volumes( + kind, channels, pipette_volume, tip_volume, extra + ) if not _check_if_software_supports_high_volumes(): if ctx.is_simulating(): test_volumes = _reduce_volumes_to_not_exceed_software_limit( - test_volumes, cfg + test_volumes, pipette_volume, channels, tip_volume ) else: raise RuntimeError("you are not the correct branch") - return sorted(test_volumes, reverse=False) # lowest volumes first + return test_volumes def _load_pipette( - ctx: ProtocolContext, cfg: config.VolumetricConfig + ctx: ProtocolContext, + pipette_channels: int, + pipette_volume: int, + pipette_mount: str, + increment: bool, + gantry_speed: Optional[int] = None, ) -> InstrumentContext: load_str_channels = {1: "single_gen3", 8: "multi_gen3", 96: "96"} - pip_channels = cfg.pipette_channels + pip_channels = pipette_channels if pip_channels not in load_str_channels: raise ValueError(f"unexpected number of channels: {pip_channels}") chnl_str = load_str_channels[pip_channels] - pip_name = f"p{cfg.pipette_volume}_{chnl_str}" - ui.print_info(f'pipette "{pip_name}" on mount "{cfg.pipette_mount}"') + pip_name = f"p{pipette_volume}_{chnl_str}" + ui.print_info(f'pipette "{pip_name}" on mount "{pipette_mount}"') # if we're doing multiple tests in one run, the pipette may already be loaded loaded_pipettes = ctx.loaded_instruments - if cfg.pipette_mount in loaded_pipettes.keys(): - return loaded_pipettes[cfg.pipette_mount] + if pipette_mount in loaded_pipettes.keys(): + return loaded_pipettes[pipette_mount] - pipette = ctx.load_instrument(pip_name, cfg.pipette_mount) - assert pipette.max_volume == cfg.pipette_volume, ( - f"expected {cfg.pipette_volume} uL pipette, " + pipette = ctx.load_instrument(pip_name, pipette_mount) + assert pipette.max_volume == pipette_volume, ( + f"expected {pipette_volume} uL pipette, " f"but got a {pipette.max_volume} uL pipette" ) - if hasattr(cfg, "gantry_speed"): - pipette.default_speed = getattr(cfg, "gantry_speed") + if gantry_speed is not None: + pipette.default_speed = gantry_speed # NOTE: 8ch QC testing means testing 1 channel at a time, # so we need to decrease the pick-up current to work with 1 tip. - if pipette.channels == 8 and not cfg.increment: + if pipette.channels == 8 and not increment: hwapi = get_sync_hw_api(ctx) - mnt = OT3Mount.LEFT if cfg.pipette_mount == "left" else OT3Mount.RIGHT + mnt = OT3Mount.LEFT if pipette_mount == "left" else OT3Mount.RIGHT hwpipette: Pipette = hwapi.hardware_pipettes[mnt.to_mount()] hwpipette.pick_up_configurations.current = 0.2 return pipette def _get_tag_from_pipette( - pipette: InstrumentContext, - cfg: config.VolumetricConfig, + pipette: InstrumentContext, increment: bool, user_volumes: bool ) -> str: pipette_tag = get_pipette_unique_name(pipette) ui.print_info(f'found pipette "{pipette_tag}"') - if cfg.increment: + if increment: pipette_tag += "-increment" - elif cfg.user_volumes: + elif user_volumes: pipette_tag += "-user-volume" else: pipette_tag += "-qc" @@ -374,26 +408,34 @@ def _load_tipracks( return tipracks -def get_test_volumes(cfg: config.VolumetricConfig) -> List[float]: +def get_test_volumes( + kind: config.ConfigType, pipette: int, volume: int, tip: int, extra: bool +) -> List[float]: """Get test volumes.""" - if cfg.kind is config.ConfigType.photometric: - return config.QC_VOLUMES_P[cfg.pipette_channels][cfg.pipette_volume][ - cfg.tip_volume - ] + volumes: List[float] = [] + print(f"Finding volumes for p {pipette} {volume} with tip {tip}, extra: {extra}") + if kind is config.ConfigType.photometric: + for t, vls in config.QC_VOLUMES_P[pipette][volume]: + if t == tip: + volumes = vls else: - if cfg.extra: - return config.QC_VOLUMES_EXTRA_G[cfg.pipette_channels][cfg.pipette_volume][ - cfg.tip_volume - ] + if extra: + cfg = config.QC_VOLUMES_EXTRA_G else: - return config.QC_VOLUMES_G[cfg.pipette_channels][cfg.pipette_volume][ - cfg.tip_volume - ] + cfg = config.QC_VOLUMES_G + + for t, vls in cfg[pipette][volume]: + print(f"tip {t} volumes {vls}") + if t == tip: + volumes = vls + break + print(f"final volumes{volumes}") + return volumes -def get_default_trials(cfg: config.VolumetricConfig) -> int: +def get_default_trials(increment: bool, kind: config.ConfigType, channels: int) -> int: """Return the default number of trials for QC tests.""" - if cfg.increment: + if increment: return 3 else: - return config.QC_DEFAULT_TRIALS[cfg.kind][cfg.pipette_channels] + return config.QC_DEFAULT_TRIALS[kind][channels] diff --git a/hardware-testing/hardware_testing/gravimetric/measurement/record.py b/hardware-testing/hardware_testing/gravimetric/measurement/record.py index 48048490835..0d59ece1419 100644 --- a/hardware-testing/hardware_testing/gravimetric/measurement/record.py +++ b/hardware-testing/hardware_testing/gravimetric/measurement/record.py @@ -268,11 +268,13 @@ def _record_get_interval_overlap(samples: GravimetricRecording, period: float) - class GravimetricRecorder: """Gravimetric Recorder.""" - def __init__(self, cfg: GravimetricRecorderConfig, simulate: bool = False) -> None: + def __init__( + self, cfg: GravimetricRecorderConfig, scale: Scale, simulate: bool = False + ) -> None: """Gravimetric Recorder.""" self._cfg = cfg self._file_name: Optional[str] = None - self._scale: Scale = Scale.build(simulate=simulate) + self._scale: Scale = scale self._recording = GravimetricRecording() self._is_recording = Event() self._reading_samples = Event() diff --git a/hardware-testing/hardware_testing/gravimetric/report.py b/hardware-testing/hardware_testing/gravimetric/report.py index b13d25e3358..0c06c21cfd3 100644 --- a/hardware-testing/hardware_testing/gravimetric/report.py +++ b/hardware-testing/hardware_testing/gravimetric/report.py @@ -1,7 +1,7 @@ """Report.""" from dataclasses import fields from enum import Enum -from typing import List, Tuple, Any +from typing import List, Tuple, Any, Dict from hardware_testing.data.csv_report import ( CSVReport, @@ -79,7 +79,7 @@ class EnvironmentReportState(str, Enum): def create_csv_test_report_photometric( - volumes: List[float], cfg: config.PhotometricConfig, run_id: str + volumes: List[float], trials: int, name: str, run_id: str ) -> CSVReport: """Create CSV test report.""" env_info = [field.name.replace("_", "-") for field in fields(EnvironmentData)] @@ -92,11 +92,11 @@ def create_csv_test_report_photometric( 0, trial, ) - for trial in range(cfg.trials) + for trial in range(trials) ] report = CSVReport( - test_name=cfg.name, + test_name=name, run_id=run_id, sections=[ CSVSection( @@ -104,7 +104,9 @@ def create_csv_test_report_photometric( lines=[ CSVLine("robot", [str]), CSVLine("pipette", [str]), - CSVLine("tips", [str]), + CSVLine("tips_50ul", [str]), + CSVLine("tips_200ul", [str]), + CSVLine("tips_1000ul", [str]), CSVLine("environment", [str]), CSVLine("liquid", [str]), ], @@ -133,21 +135,29 @@ def create_csv_test_report_photometric( ), ], ) - # might as well set the configuration values now + return report + + +def store_config_pm(report: CSVReport, cfg: config.PhotometricConfig) -> None: + """Store the config file list.""" for field in fields(config.PhotometricConfig): if field.name in config.PHOTO_CONFIG_EXCLUDE_FROM_REPORT: continue report("CONFIG", field.name, [getattr(cfg, field.name)]) - return report def create_csv_test_report( - volumes: List[float], cfg: config.GravimetricConfig, run_id: str + volumes: List[float], + pipette_channels: int, + increment: bool, + trials: int, + name: str, + run_id: str, ) -> CSVReport: """Create CSV test report.""" env_info = [field.name.replace("_", "-") for field in fields(EnvironmentData)] meas_info = [field.name.replace("_", "-") for field in fields(MeasurementData)] - if cfg.pipette_channels == 8 and not cfg.increment: + if pipette_channels == 8 and not increment: pip_channels_tested = 8 else: pip_channels_tested = 1 @@ -167,7 +177,7 @@ def create_csv_test_report( trial, ) for channel in range(pip_channels_tested) - for trial in range(cfg.trials) + for trial in range(trials) ] # Get label for different volume stores, "channel_all", "channel_1" through channel count, @@ -175,7 +185,7 @@ def create_csv_test_report( volume_stat_type = ( ["channel_all"] + [f"channel_{c+1}" for c in range(pip_channels_tested)] - + [f"trial_{t+1}" for t in range(cfg.trials)] + + [f"trial_{t+1}" for t in range(trials)] ) def _field_type_not_using_typing(t: Any) -> Any: @@ -184,7 +194,7 @@ def _field_type_not_using_typing(t: Any) -> Any: return t report = CSVReport( - test_name=cfg.name, + test_name=name, run_id=run_id, sections=[ CSVSection( @@ -192,7 +202,9 @@ def _field_type_not_using_typing(t: Any) -> Any: lines=[ CSVLine("robot", [str]), CSVLine("pipette", [str]), - CSVLine("tips", [str]), + CSVLine("tips_50ul", [str]), + CSVLine("tips_200ul", [str]), + CSVLine("tips_1000ul", [str]), CSVLine("scale", [str]), CSVLine("environment", [str]), CSVLine("liquid", [str]), @@ -230,7 +242,7 @@ def _field_type_not_using_typing(t: Any) -> Any: ) for v in volumes for c in range(pip_channels_tested) - for t in range(cfg.trials) + for t in range(trials) for m in ["aspirate", "dispense"] ], ), @@ -265,26 +277,30 @@ def _field_type_not_using_typing(t: Any) -> Any: ), ], ) - # might as well set the configuration values now + return report + + +def store_config_gm(report: CSVReport, cfg: config.GravimetricConfig) -> None: + """Store the config file list.""" for field in fields(config.GravimetricConfig): if field.name in config.GRAV_CONFIG_EXCLUDE_FROM_REPORT: continue report("CONFIG", field.name, [getattr(cfg, field.name)]) - return report def store_serial_numbers_pm( report: CSVReport, robot: str, pipette: str, - tips: str, + tips: Dict[str, str], environment: str, liquid: str, ) -> None: """Report serial numbers.""" report("SERIAL-NUMBERS", "robot", [robot]) report("SERIAL-NUMBERS", "pipette", [pipette]) - report("SERIAL-NUMBERS", "tips", [tips]) + for tip in tips.keys(): + report("SERIAL-NUMBERS", tip, [tips[tip]]) report("SERIAL-NUMBERS", "environment", [environment]) report("SERIAL-NUMBERS", "liquid", [liquid]) @@ -304,7 +320,7 @@ def store_serial_numbers( report: CSVReport, robot: str, pipette: str, - tips: str, + tips: Dict[str, str], scale: str, environment: str, liquid: str, @@ -314,7 +330,8 @@ def store_serial_numbers( report.set_device_id(pipette, pipette) report("SERIAL-NUMBERS", "robot", [robot]) report("SERIAL-NUMBERS", "pipette", [pipette]) - report("SERIAL-NUMBERS", "tips", [tips]) + for tip in tips.keys(): + report("SERIAL-NUMBERS", tip, [tips[tip]]) report("SERIAL-NUMBERS", "scale", [scale]) report("SERIAL-NUMBERS", "environment", [environment]) report("SERIAL-NUMBERS", "liquid", [liquid]) diff --git a/hardware-testing/hardware_testing/gravimetric/tips.py b/hardware-testing/hardware_testing/gravimetric/tips.py index 34ca135774a..1e8aaa46f1b 100644 --- a/hardware-testing/hardware_testing/gravimetric/tips.py +++ b/hardware-testing/hardware_testing/gravimetric/tips.py @@ -35,24 +35,26 @@ MULTI_CHANNEL_TEST_ORDER = [0, 1, 2, 3, 7, 6, 5, 4] # zero indexed CHANNEL_TO_TIP_ROW_LOOKUP = { # zero indexed - 0: "H", - 1: "G", - 2: "E", - 3: "B", - 4: "G", - 5: "D", - 6: "B", - 7: "A", -} -CHANNEL_TO_SLOT_ROW_LOOKUP = { # zero indexed - 0: "B", - 1: "B", - 2: "B", - 3: "B", - 4: "C", - 5: "C", + 0: "G", + 1: "F", + 2: "D", + 3: "A", + 4: "H", + 5: "E", 6: "C", - 7: "C", + 7: "B", +} +REAR_CHANNELS = [0, 1, 2, 3] +FRONT_CHANNELS = [4, 5, 6, 7] +REAR_CHANNELS_TIP_SLOTS = { + 50: [2, 7], + 200: [10], + 1000: [3], +} +FRONT_CHANNELS_TIP_SLOTS = { + 50: [8, 6], + 200: [5], + 1000: [9], } @@ -64,32 +66,52 @@ def _get_racks(ctx: ProtocolContext) -> Dict[int, Labware]: } +def _unused_tips_for_racks(racks: List[Labware]) -> List[Well]: + wells: List[Well] = [] + rows = "ABCDEFGH" + for rack in racks: + for col in range(1, 13): + for row in rows: + wellname = f"{row}{col}" + next_well = rack.next_tip(1, rack[wellname]) + if next_well is not None and wellname == next_well.well_name: + wells.append(rack[wellname]) + return wells + + +def get_unused_tips(ctx: ProtocolContext, tip_volume: int) -> List[Well]: + """Use the labware's tip tracker to get a list of all unused tips for a given tip volume.""" + racks = [ + r for r in _get_racks(ctx).values() if r.wells()[0].max_volume == tip_volume + ] + return _unused_tips_for_racks(racks) + + def get_tips_for_single(ctx: ProtocolContext, tip_volume: int) -> List[Well]: """Get tips for single channel.""" - racks = _get_racks(ctx) - return [ - tip - for rack in racks.values() - for tip in rack.wells() - if tip.max_volume == tip_volume - ] + return get_unused_tips(ctx, tip_volume) def get_tips_for_individual_channel_on_multi( - ctx: ProtocolContext, channel: int + ctx: ProtocolContext, channel: int, tip_volume: int, pipette_volume: int ) -> List[Well]: """Get tips for a multi's channel.""" - racks = _get_racks(ctx) - slot_row = CHANNEL_TO_SLOT_ROW_LOOKUP[channel] + print(f"getting {tip_volume} tips for channel {channel}") + if pipette_volume == 1000: + if channel in FRONT_CHANNELS: + slots = FRONT_CHANNELS_TIP_SLOTS[tip_volume] + else: + slots = REAR_CHANNELS_TIP_SLOTS[tip_volume] + print(f"Slots for this channel/tip {slots}") + all_racks = _get_racks(ctx) + specific_racks: List[Labware] = [] + for slot in slots: + specific_racks.append(all_racks[slot]) + unused_tips = _unused_tips_for_racks(specific_racks) + else: + unused_tips = get_unused_tips(ctx, tip_volume) tip_row = CHANNEL_TO_TIP_ROW_LOOKUP[channel] - # FIXME: need custom deck to support 3x racks horizontally - slots = [5, 6] if slot_row == "B" else [8, 9] - tips = [ - tip - for slot in slots - for tip in racks[slot].wells() - if tip.well_name[0] == tip_row - ] + tips = [tip for tip in unused_tips if tip.well_name[0] == tip_row] return tips @@ -119,7 +141,9 @@ def get_tips( return {0: get_tips_for_all_channels_on_multi(ctx)} else: return { - channel: get_tips_for_individual_channel_on_multi(ctx, channel) + channel: get_tips_for_individual_channel_on_multi( + ctx, channel, tip_volume, int(pipette.max_volume) + ) for channel in range(pipette.channels) } elif pipette.channels == 96: diff --git a/hardware-testing/hardware_testing/gravimetric/trial.py b/hardware-testing/hardware_testing/gravimetric/trial.py index 936b3c25b32..8b91b03eee2 100644 --- a/hardware-testing/hardware_testing/gravimetric/trial.py +++ b/hardware-testing/hardware_testing/gravimetric/trial.py @@ -28,6 +28,7 @@ class VolumetricTrial: volume: float mix: bool acceptable_cv: Optional[float] + acceptable_d: Optional[float] env_sensor: asair_sensor.AsairSensorBase @@ -68,17 +69,12 @@ class TestResources: ctx: ProtocolContext pipette: InstrumentContext - pipette_tag: str tipracks: List[Labware] test_volumes: List[float] - run_id: str - start_time: float - operator_name: str - robot_serial: str - tip_batch: str - git_description: str tips: Dict[int, List[Well]] env_sensor: asair_sensor.AsairSensorBase + recorder: Optional[GravimetricRecorder] + test_report: CSVReport def build_gravimetric_trials( @@ -123,6 +119,7 @@ def build_gravimetric_trials( stable=True, scale_delay=cfg.scale_delay, acceptable_cv=None, + acceptable_d=None, cfg=cfg, env_sensor=env_sensor, ) @@ -137,6 +134,14 @@ def build_gravimetric_trials( trial_list[volume][channel] = [] channel_offset = helpers._get_channel_offset(cfg, channel) for trial in range(cfg.trials): + d: Optional[float] = None + cv: Optional[float] = None + if not cfg.increment: + d, cv = config.QC_TEST_MIN_REQUIREMENTS[cfg.pipette_channels][ + cfg.pipette_volume + ][cfg.tip_volume][volume] + d = d * (1 - config.QC_TEST_SAFETY_FACTOR) + cv = cv * (1 - config.QC_TEST_SAFETY_FACTOR) trial_list[volume][channel].append( GravimetricTrial( ctx=ctx, @@ -156,7 +161,8 @@ def build_gravimetric_trials( mix=cfg.mix, stable=True, scale_delay=cfg.scale_delay, - acceptable_cv=None, + acceptable_cv=cv, + acceptable_d=d, cfg=cfg, env_sensor=env_sensor, ) @@ -180,6 +186,14 @@ def build_photometric_trials( for volume in test_volumes: trial_list[volume] = [] for trial in range(cfg.trials): + d: Optional[float] = None + cv: Optional[float] = None + if not cfg.increment: + d, cv = config.QC_TEST_MIN_REQUIREMENTS[96][cfg.pipette_volume][ + cfg.tip_volume + ][volume] + d = d * (1 - config.QC_TEST_SAFETY_FACTOR) + cv = cv * (1 - config.QC_TEST_SAFETY_FACTOR) trial_list[volume].append( PhotometricTrial( ctx=ctx, @@ -194,7 +208,8 @@ def build_photometric_trials( inspect=cfg.inspect, cfg=cfg, mix=cfg.mix, - acceptable_cv=None, + acceptable_cv=cv, + acceptable_d=d, env_sensor=env_sensor, ) ) @@ -206,7 +221,6 @@ def _finish_test( resources: TestResources, return_tip: bool, ) -> None: - ui.print_title("CHANGE PIPETTES") if resources.pipette.has_tip: if resources.pipette.current_volume > 0: ui.print_info("dispensing liquid to trash") @@ -218,7 +232,12 @@ def _finish_test( resources.pipette.aspirate(10) # to pull any droplets back up ui.print_info("dropping tip") helpers._drop_tip(resources.pipette, return_tip) + + +def _change_pipettes( + ctx: ProtocolContext, + pipette: InstrumentContext, +) -> None: + ui.print_title("CHANGE PIPETTES") ui.print_info("moving to attach position") - resources.pipette.move_to( - resources.ctx.deck.position_for(5).move(Point(x=0, y=9 * 7, z=150)) - ) + pipette.move_to(ctx.deck.position_for(5).move(Point(x=0, y=9 * 7, z=150))) diff --git a/hardware-testing/hardware_testing/gravimetric/workarounds.py b/hardware-testing/hardware_testing/gravimetric/workarounds.py index 98da9e90ce8..63b574d9f43 100644 --- a/hardware-testing/hardware_testing/gravimetric/workarounds.py +++ b/hardware-testing/hardware_testing/gravimetric/workarounds.py @@ -1,5 +1,4 @@ """Opentrons API Workarounds.""" -from atexit import register as atexit_register from datetime import datetime from urllib.request import Request, urlopen from typing import List @@ -44,7 +43,6 @@ def http_get_all_labware_offsets() -> List[dict]: runs_response = urlopen(req) runs_response_data = runs_response.read() stop_server_ot3() - atexit_register(start_server_ot3) runs_json = json_loads(runs_response_data) protocols_list = runs_json["data"] diff --git a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_96.py b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_96.py new file mode 100644 index 00000000000..992fe0c86fe --- /dev/null +++ b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_96.py @@ -0,0 +1,28 @@ +"""Photometric OT3 P1000.""" +from opentrons.protocol_api import ProtocolContext + +metadata = {"protocolName": "gravimetric-ot3-p1000-96-1000ul-tip"} +requirements = {"robotType": "Flex", "apiLevel": "2.15"} + +SLOT_SCALE = 4 +SLOTS_TIPRACK = { + # TODO: add slot 12 when tipracks are disposable + 1000: [2, 3, 5, 6, 7, 8, 9, 10, 11], +} +LABWARE_ON_SCALE = "nest_1_reservoir_195ml" + + +def run(ctx: ProtocolContext) -> None: + """Run.""" + tipracks = [ + ctx.load_labware(f"opentrons_flex_96_tiprack_{size}uL_adp", slot) + for size, slots in SLOTS_TIPRACK.items() + for slot in slots + ] + scale_labware = ctx.load_labware(LABWARE_ON_SCALE, SLOT_SCALE) + pipette = ctx.load_instrument("p1000_96", "left") + for rack in tipracks: + pipette.pick_up_tip(rack["A1"]) + pipette.aspirate(10, scale_labware["A1"].top()) + pipette.dispense(10, scale_labware["A1"].top()) + pipette.drop_tip(home_after=False) diff --git a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_1000ul_tip.py b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi.py similarity index 86% rename from hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_1000ul_tip.py rename to hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi.py index b71452c871d..6b621a3c770 100644 --- a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_1000ul_tip.py +++ b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi.py @@ -1,12 +1,14 @@ """Gravimetric OT3 P1000.""" from opentrons.protocol_api import ProtocolContext -metadata = {"protocolName": "gravimetric-ot3-p1000-multi-1000ul-tip"} +metadata = {"protocolName": "gravimetric-ot3-p1000-multi"} requirements = {"robotType": "Flex", "apiLevel": "2.15"} SLOT_SCALE = 4 SLOTS_TIPRACK = { - 1000: [5, 6, 8, 9], + 50: [2, 6, 7, 8], + 200: [10, 5], + 1000: [3, 9], } LABWARE_ON_SCALE = "radwag_pipette_calibration_vial" diff --git a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_200ul_tip.py b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_200ul_tip.py deleted file mode 100644 index bd9ee86afd6..00000000000 --- a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_200ul_tip.py +++ /dev/null @@ -1,27 +0,0 @@ -"""Gravimetric OT3 P1000.""" -from opentrons.protocol_api import ProtocolContext - -metadata = {"protocolName": "gravimetric-ot3-p1000-multi-200ul-tip"} -requirements = {"robotType": "Flex", "apiLevel": "2.15"} - -SLOT_SCALE = 4 -SLOTS_TIPRACK = { - 200: [5, 6, 8, 9], -} -LABWARE_ON_SCALE = "radwag_pipette_calibration_vial" - - -def run(ctx: ProtocolContext) -> None: - """Run.""" - tipracks = [ - ctx.load_labware(f"opentrons_flex_96_tiprack_{size}uL", slot) - for size, slots in SLOTS_TIPRACK.items() - for slot in slots - ] - vial = ctx.load_labware(LABWARE_ON_SCALE, SLOT_SCALE) - pipette = ctx.load_instrument("p1000_multi_gen3", "left") - for rack in tipracks: - pipette.pick_up_tip(rack["A1"]) - pipette.aspirate(10, vial["A1"].top()) - pipette.dispense(10, vial["A1"].top()) - pipette.drop_tip(home_after=False) diff --git a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_50ul_tip.py b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p50_multi.py similarity index 68% rename from hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_50ul_tip.py rename to hardware-testing/hardware_testing/protocols/gravimetric_ot3_p50_multi.py index 556a133aa95..95e88293539 100644 --- a/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p1000_multi_50ul_tip.py +++ b/hardware-testing/hardware_testing/protocols/gravimetric_ot3_p50_multi.py @@ -1,7 +1,7 @@ -"""Gravimetric OT3 P1000.""" +"""Gravimetric OT3.""" from opentrons.protocol_api import ProtocolContext -metadata = {"protocolName": "gravimetric-ot3-p1000-multi-50ul-tip"} +metadata = {"protocolName": "gravimetric-ot3-p50-multi-50ul-tip"} requirements = {"robotType": "Flex", "apiLevel": "2.15"} SLOT_SCALE = 4 @@ -19,9 +19,9 @@ def run(ctx: ProtocolContext) -> None: for slot in slots ] vial = ctx.load_labware(LABWARE_ON_SCALE, SLOT_SCALE) - pipette = ctx.load_instrument("p1000_multi_gen3", "left") + pipette = ctx.load_instrument("p50_multi_gen3", "left") for rack in tipracks: pipette.pick_up_tip(rack["A1"]) - pipette.aspirate(10, vial["A1"].top()) - pipette.dispense(10, vial["A1"].top()) + pipette.aspirate(pipette.min_volume, vial["A1"].top()) + pipette.dispense(pipette.min_volume, vial["A1"].top()) pipette.drop_tip(home_after=False) diff --git a/hardware-testing/hardware_testing/scripts/gravimetric_rnd.py b/hardware-testing/hardware_testing/scripts/gravimetric_rnd.py index a5cec740817..46c35d0ffdd 100644 --- a/hardware-testing/hardware_testing/scripts/gravimetric_rnd.py +++ b/hardware-testing/hardware_testing/scripts/gravimetric_rnd.py @@ -7,6 +7,7 @@ GravimetricRecorder, GravimetricRecorderConfig, ) +from hardware_testing.gravimetric.scale import Scale # type: ignore[import] metadata = {"protocolName": "gravimetric-rnd", "apiLevel": "2.12"} CALIBRATE_SCALE = False @@ -26,6 +27,7 @@ def _run(is_simulating: bool) -> None: frequency=5, stable=False, ), + scale=Scale.build(simulate=is_simulating), simulate=is_simulating, ) if CALIBRATE_SCALE: diff --git a/hardware-testing/tests/hardware_testing/drivers/test_asair_sensor.py b/hardware-testing/tests/hardware_testing/drivers/test_asair_sensor.py index 7480b6665a6..55ac2f0e99f 100644 --- a/hardware-testing/tests/hardware_testing/drivers/test_asair_sensor.py +++ b/hardware-testing/tests/hardware_testing/drivers/test_asair_sensor.py @@ -23,8 +23,8 @@ def test_reading(subject: AsairSensor, connection: MagicMock) -> None: connection.read.return_value = data connection.inWaiting.return_value = len(data) assert subject.get_reading() == Reading( - 1.0, 3.0, + 1.0, ) connection.write.assert_called_once_with(b"\x01\x03\x00\x00\x00\x02\xC4\x0b")