From bf85a29a3aee39e3f96629ec005e5c0a2a331311 Mon Sep 17 00:00:00 2001 From: CaseyBatten Date: Tue, 3 Oct 2023 12:11:15 -0400 Subject: [PATCH] fix(api): Prompt user with distribution error to prevent invalid disposal values (#13659) Prevents the distribute functionality from accepting a disposal value equal to or greater than the maximum value of the instrument. --- .../protocols/advanced_control/transfers.py | 38 ++++++++++ .../advanced_control/test_transfers.py | 75 +++++++++++++++++++ 2 files changed, 113 insertions(+) diff --git a/api/src/opentrons/protocols/advanced_control/transfers.py b/api/src/opentrons/protocols/advanced_control/transfers.py index 00f4a1ae02e..8f513a3afff 100644 --- a/api/src/opentrons/protocols/advanced_control/transfers.py +++ b/api/src/opentrons/protocols/advanced_control/transfers.py @@ -485,6 +485,11 @@ def _plan_transfer(self): """ # reform source target lists sources, dests = self._extend_source_target_lists(self._sources, self._dests) + self._check_valid_volume_parameters( + disposal_volume=self._strategy.disposal_volume, + air_gap=self._strategy.air_gap, + max_volume=self._instr.max_volume, + ) plan_iter = self._expand_for_volume_constraints( self._volumes, zip(sources, dests), @@ -571,6 +576,13 @@ def _plan_distribute(self): .. Dispense air gap -> ...* """ + + self._check_valid_volume_parameters( + disposal_volume=self._strategy.disposal_volume, + air_gap=self._strategy.air_gap, + max_volume=self._instr.max_volume, + ) + # TODO: decide whether default disposal vol for distribute should be # pipette min_vol or should we leave it to being 0 by default and # recommend users to specify a disposal vol when using distribute. @@ -633,6 +645,10 @@ def _expand_for_volume_constraints( """Split a sequence of proposed transfers if necessary to keep each transfer under the given max volume. """ + # A final defense against an infinite loop. + # Raising a proper exception with a helpful message is left to calling code, + # because it has more context about what the user is trying to do. + assert max_volume > 0 for volume, target in zip(volumes, targets): while volume > max_volume * 2: yield max_volume, target @@ -680,6 +696,12 @@ def _plan_consolidate(self): *.. Aspirate -> Air gap -> Touch tip ->.. .. Aspirate -> .....* """ + # TODO: verify if _check_valid_volume_parameters should be re-enabled here + # self._check_valid_volume_parameters( + # disposal_volume=self._strategy.disposal_volume, + # air_gap=self._strategy.air_gap, + # max_volume=self._instr.max_volume, + # ) plan_iter = self._expand_for_volume_constraints( # todo(mm, 2021-03-09): Is it right to use _instr.max_volume here? # Why don't we account for tip max volume, disposal volume, or air @@ -843,6 +865,22 @@ def _map_volume(i): return [_map_volume(i) for i in range(total)] + def _check_valid_volume_parameters( + self, disposal_volume: float, air_gap: float, max_volume: float + ): + if air_gap >= max_volume: + raise ValueError( + "The air gap must be less than the maximum volume of the pipette" + ) + elif disposal_volume >= max_volume: + raise ValueError( + "The disposal volume must be less than the maximum volume of the pipette" + ) + elif disposal_volume + air_gap >= max_volume: + raise ValueError( + "The sum of the air gap and disposal volume must be less than the maximum volume of the pipette" + ) + def _check_valid_well_list(self, well_list, id, old_well_list): if self._api_version >= APIVersion(2, 2) and len(well_list) < 1: raise RuntimeError( diff --git a/api/tests/opentrons/protocols/advanced_control/test_transfers.py b/api/tests/opentrons/protocols/advanced_control/test_transfers.py index 71237286af7..21f6c30dae9 100644 --- a/api/tests/opentrons/protocols/advanced_control/test_transfers.py +++ b/api/tests/opentrons/protocols/advanced_control/test_transfers.py @@ -796,6 +796,81 @@ def test_all_options(_instr_labware): assert xfer_plan_list == exp1 +def test_invalid_air_gap_disposal_sum_distribute(_instr_labware): + _instr_labware["ctx"].home() + lw1 = _instr_labware["lw1"] + lw2 = _instr_labware["lw2"] + # Supply the disposal volume assessment with the instrument followed by a disposal value equal to the max volume + + options = tx.TransferOptions( + transfer=tx.Transfer( + disposal_volume=_instr_labware["instr"].max_volume / 2, + air_gap=_instr_labware["instr"].max_volume / 2, + ) + ) + with pytest.raises(ValueError): + plan = tx.TransferPlan( + 700, + lw1.columns()[0][0], + lw2.rows()[0][1:3], + _instr_labware["instr"], + max_volume=_instr_labware["instr"].hw_pipette["max_volume"], + api_version=_instr_labware["ctx"].api_version, + mode="distribute", + options=options, + ) + # Exhaust the iterator in case it raises the expected exception lazily. + list(plan) + + +def test_invalid_air_gap_distribute(_instr_labware): + _instr_labware["ctx"].home() + lw1 = _instr_labware["lw1"] + lw2 = _instr_labware["lw2"] + # Supply the disposal volume assessment with the instrument followed by a disposal value equal to the max volume + + options = tx.TransferOptions( + transfer=tx.Transfer(air_gap=_instr_labware["instr"].max_volume) + ) + with pytest.raises(ValueError): + plan = tx.TransferPlan( + 700, + lw1.columns()[0][0], + lw2.rows()[0][1:3], + _instr_labware["instr"], + max_volume=_instr_labware["instr"].hw_pipette["max_volume"], + api_version=_instr_labware["ctx"].api_version, + mode="distribute", + options=options, + ) + # Exhaust the iterator in case it raises the expected exception lazily. + list(plan) + + +def test_invalid_disposal_volume_distribute(_instr_labware): + _instr_labware["ctx"].home() + lw1 = _instr_labware["lw1"] + lw2 = _instr_labware["lw2"] + # Supply the disposal volume assessment with the instrument followed by a disposal value equal to the max volume + + options = tx.TransferOptions( + transfer=tx.Transfer(disposal_volume=_instr_labware["instr"].max_volume) + ) + with pytest.raises(ValueError): + plan = tx.TransferPlan( + 700, + lw1.columns()[0][0], + lw2.rows()[0][1:3], + _instr_labware["instr"], + max_volume=_instr_labware["instr"].hw_pipette["max_volume"], + api_version=_instr_labware["ctx"].api_version, + mode="distribute", + options=options, + ) + # Exhaust the iterator in case it raises the expected exception lazily. + list(plan) + + def test_oversized_distribute(_instr_labware): _instr_labware["ctx"].home() lw1 = _instr_labware["lw1"]