fix(api): Addition of distribution error to prevent invalid disposal values

api/src/opentrons/protocols/advanced_control/transfers.py

@@ -484,6 +484,11 @@ def _plan_transfer(self):
             -> Blow out -> Touch tip -> Drop tip*
         """
         # reform source target lists
+        self._check_valid_volume_parameters(
+            self._strategy.disposal_volume,
+            self._strategy.air_gap,
+            self._instr._core.get_working_volume(),
+        )
         sources, dests = self._extend_source_target_lists(self._sources, self._dests)
         plan_iter = self._expand_for_volume_constraints(
             self._volumes,
@@ -571,6 +576,13 @@ def _plan_distribute(self):
                .. Dispense air gap -> ...*
 
         """
+
+        self._check_valid_volume_parameters(
+            self._strategy.disposal_volume,
+            self._strategy.air_gap,
+            self._instr._core.get_working_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,7 @@ def _expand_for_volume_constraints(
         """Split a sequence of proposed transfers if necessary to keep each
         transfer under the given max volume.
         """
+        assert max_volume > 0
         for volume, target in zip(volumes, targets):
             while volume > max_volume * 2:
                 yield max_volume, target
@@ -680,6 +693,11 @@ def _plan_consolidate(self):
                *.. Aspirate -> Air gap -> Touch tip ->..
                .. Aspirate -> .....*
         """
+        self._check_valid_volume_parameters(
+            self._strategy.disposal_volume,
+            self._strategy.air_gap,
+            self._instr._core.get_working_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 +861,20 @@ def _map_volume(i):
 
         return [_map_volume(i) for i in range(total)]
 
+    def _check_valid_volume_parameters(self, disposal_volume, air_gap, working_volume):
+        if air_gap >= working_volume:
+            raise ValueError(
+                "The air gap must be less than the working volume of the pipette"
+            )
+        elif disposal_volume >= working_volume:
+            raise ValueError(
+                "The disposal volume must be less than the working volume of the pipette"
+            )
+        elif disposal_volume + air_gap >= working_volume:
+            raise ValueError(
+                "The sum of the air gap and disposal volume must be less than the working 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,83 @@ 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()
+    options = options._replace(
+        transfer=options.transfer._replace(
+            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,
+        )
+        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()
+    options = options._replace(
+        transfer=options.transfer._replace(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,
+        )
+        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()
+    options = options._replace(
+        transfer=options.transfer._replace(
+            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,
+        )
+        list(plan)
+
+
 def test_oversized_distribute(_instr_labware):
     _instr_labware["ctx"].home()
     lw1 = _instr_labware["lw1"]