fix(api): Prompt user with distribution error to prevent invalid disp…

…osal values (#13659)

Prevents the distribute functionality from accepting a disposal value equal to or greater than the maximum value of the instrument.

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(

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"]