feat(api): proper LiquidHandlingPropertyByVolume class and validation for setting liquid class properties

api/src/opentrons/protocol_api/_liquid_properties.py

@@ -1,5 +1,7 @@
+from copy import copy
 from dataclasses import dataclass
-from typing import Optional, Dict, Sequence
+from numpy import interp
+from typing import Optional, Dict, Sequence, Union
 
 from opentrons_shared_data.liquid_classes.liquid_class_definition import (
     AspirateProperties as SharedDataAspirateProperties,
@@ -18,9 +20,45 @@
     Coordinate,
 )
 
-# TODO replace this with a class that can extrapolate given volumes to the correct float,
-#   also figure out how we want people to be able to set this
-LiquidHandlingPropertyByVolume = Dict[str, float]
+from . import validation
+
+
+class LiquidHandlingPropertyByVolume:
+    def __init__(self, properties_by_volume: Dict[str, float]) -> None:
+        self._default = properties_by_volume.pop("default")
+        self._properties_by_volume: Dict[float, float] = {
+            float(volume): value for volume, value in properties_by_volume.items()
+        }
+
+    @property
+    def default(self) -> float:
+        return self._default
+
+    @property
+    def properties_by_volume(self) -> Dict[Union[float, str], float]:
+        props_by_volume_copy: Dict[Union[float, str], float] = copy(
+            self._properties_by_volume  # type: ignore[arg-type]
+        )
+        props_by_volume_copy["default"] = self._default
+        return props_by_volume_copy
+
+    def get_for_volume(self, volume: float) -> float:
+        validated_volume = validation.ensure_positive_float(volume)
+        # numpy interp does not automatically sort the points used for interpolation, so
+        # here we are sorting them by the keys (volume) and then using zip to get two
+        # tuples in the correct order
+        sorted_volumes, sorted_values = zip(*sorted(self._properties_by_volume.items()))
+        return float(interp(validated_volume, sorted_volumes, sorted_values))
+
+    def set_for_volume(self, volume: float, value: float) -> None:
+        validated_volume = validation.ensure_positive_float(volume)
+        self._properties_by_volume[validated_volume] = value
+
+    def delete_for_volume(self, volume: float) -> None:
+        try:
+            del self._properties_by_volume[volume]
+        except KeyError:
+            raise KeyError(f"No value set for volume {volume} uL")
 
 
 @dataclass
@@ -35,19 +73,19 @@ def enabled(self) -> bool:
 
     @enabled.setter
     def enabled(self, enable: bool) -> None:
-        # TODO insert bool validation here
-        if enable and self._duration is None:
+        validated_enable = validation.ensure_boolean(enable)
+        if validated_enable and self._duration is None:
             raise ValueError("duration must be set before enabling delay.")
-        self._enabled = enable
+        self._enabled = validated_enable
 
     @property
     def duration(self) -> Optional[float]:
         return self._duration
 
     @duration.setter
     def duration(self, new_duration: float) -> None:
-        # TODO insert positive float validation here
-        self._duration = new_duration
+        validated_duration = validation.ensure_positive_float(new_duration)
+        self._duration = validated_duration
 
 
 @dataclass
@@ -64,41 +102,41 @@ def enabled(self) -> bool:
 
     @enabled.setter
     def enabled(self, enable: bool) -> None:
-        # TODO insert bool validation here
-        if enable and (
+        validated_enable = validation.ensure_boolean(enable)
+        if validated_enable and (
             self._z_offset is None or self._mm_to_edge is None or self._speed is None
         ):
             raise ValueError(
                 "z_offset, mm_to_edge and speed must be set before enabling touch tip."
             )
-        self._enabled = enable
+        self._enabled = validated_enable
 
     @property
     def z_offset(self) -> Optional[float]:
         return self._z_offset
 
     @z_offset.setter
     def z_offset(self, new_offset: float) -> None:
-        # TODO validation for float
-        self._z_offset = new_offset
+        validated_offset = validation.ensure_float(new_offset)
+        self._z_offset = validated_offset
 
     @property
     def mm_to_edge(self) -> Optional[float]:
         return self._mm_to_edge
 
     @mm_to_edge.setter
     def mm_to_edge(self, new_mm: float) -> None:
-        # TODO validation for float
-        self._z_offset = new_mm
+        validated_mm = validation.ensure_float(new_mm)
+        self._z_offset = validated_mm
 
     @property
     def speed(self) -> Optional[float]:
         return self._speed
 
     @speed.setter
     def speed(self, new_speed: float) -> None:
-        # TODO insert positive float validation here
-        self._speed = new_speed
+        validated_speed = validation.ensure_positive_float(new_speed)
+        self._speed = validated_speed
 
 
 @dataclass
@@ -114,28 +152,28 @@ def enabled(self) -> bool:
 
     @enabled.setter
     def enabled(self, enable: bool) -> None:
-        # TODO insert bool validation here
-        if enable and (self._repetitions is None or self._volume is None):
+        validated_enable = validation.ensure_boolean(enable)
+        if validated_enable and (self._repetitions is None or self._volume is None):
             raise ValueError("repetitions and volume must be set before enabling mix.")
-        self._enabled = enable
+        self._enabled = validated_enable
 
     @property
     def repetitions(self) -> Optional[int]:
         return self._repetitions
 
     @repetitions.setter
     def repetitions(self, new_repetitions: int) -> None:
-        # TODO validations for positive int
-        self._repetitions = new_repetitions
+        validated_repetitions = validation.ensure_positive_int(new_repetitions)
+        self._repetitions = validated_repetitions
 
     @property
     def volume(self) -> Optional[float]:
         return self._volume
 
     @volume.setter
     def volume(self, new_volume: float) -> None:
-        # TODO validations for volume float
-        self._volume = new_volume
+        validated_volume = validation.ensure_positive_float(new_volume)
+        self._volume = validated_volume
 
 
 @dataclass
@@ -151,20 +189,19 @@ def enabled(self) -> bool:
 
     @enabled.setter
     def enabled(self, enable: bool) -> None:
-        # TODO insert bool validation here
-        if enable and (self._location is None or self._flow_rate is None):
+        validated_enable = validation.ensure_boolean(enable)
+        if validated_enable and (self._location is None or self._flow_rate is None):
             raise ValueError(
                 "location and flow_rate must be set before enabling blowout."
             )
-        self._enabled = enable
+        self._enabled = validated_enable
 
     @property
     def location(self) -> Optional[BlowoutLocation]:
         return self._location
 
     @location.setter
     def location(self, new_location: str) -> None:
-        # TODO blowout location validation
         self._location = BlowoutLocation(new_location)
 
     @property
@@ -173,8 +210,8 @@ def flow_rate(self) -> Optional[float]:
 
     @flow_rate.setter
     def flow_rate(self, new_flow_rate: float) -> None:
-        # TODO validations for positive float
-        self._flow_rate = new_flow_rate
+        validated_flow_rate = validation.ensure_positive_float(new_flow_rate)
+        self._flow_rate = validated_flow_rate
 
 
 @dataclass
@@ -191,7 +228,6 @@ def position_reference(self) -> PositionReference:
 
     @position_reference.setter
     def position_reference(self, new_position: str) -> None:
-        # TODO validation for position reference
         self._position_reference = PositionReference(new_position)
 
     @property
@@ -200,17 +236,17 @@ def offset(self) -> Coordinate:
 
     @offset.setter
     def offset(self, new_offset: Sequence[float]) -> None:
-        # TODO validate valid coordinates
-        self._offset = Coordinate(x=new_offset[0], y=new_offset[1], z=new_offset[2])
+        x, y, z = validation.validate_coordinates(new_offset)
+        self._offset = Coordinate(x=x, y=y, z=z)
 
     @property
     def speed(self) -> float:
         return self._speed
 
     @speed.setter
     def speed(self, new_speed: float) -> None:
-        # TODO insert positive float validation here
-        self._speed = new_speed
+        validated_speed = validation.ensure_positive_float(new_speed)
+        self._speed = validated_speed
 
     @property
     def delay(self) -> DelayProperties:
@@ -276,7 +312,6 @@ def position_reference(self) -> PositionReference:
 
     @position_reference.setter
     def position_reference(self, new_position: str) -> None:
-        # TODO validation for position reference
         self._position_reference = PositionReference(new_position)
 
     @property
@@ -285,8 +320,8 @@ def offset(self) -> Coordinate:
 
     @offset.setter
     def offset(self, new_offset: Sequence[float]) -> None:
-        # TODO validate valid coordinates
-        self._offset = Coordinate(x=new_offset[0], y=new_offset[1], z=new_offset[2])
+        x, y, z = validation.validate_coordinates(new_offset)
+        self._offset = Coordinate(x=x, y=y, z=z)
 
     @property
     def flow_rate_by_volume(self) -> LiquidHandlingPropertyByVolume:
@@ -310,8 +345,8 @@ def pre_wet(self) -> bool:
 
     @pre_wet.setter
     def pre_wet(self, new_setting: bool) -> None:
-        # TODO boolean validation
-        self._pre_wet = new_setting
+        validated_setting = validation.ensure_boolean(new_setting)
+        self._pre_wet = validated_setting
 
     @property
     def retract(self) -> RetractAspirate:
@@ -362,8 +397,6 @@ def disposal_by_volume(self) -> LiquidHandlingPropertyByVolume:
         return self._disposal_by_volume
 
 
-# TODO (spp, 2024-10-17): create PAPI-equivalent types for all the properties
-#  and have validation on value updates with user-facing error messages
 @dataclass
 class TransferProperties:
     _aspirate: AspirateProperties
@@ -461,7 +494,9 @@ def _build_retract_aspirate(
         _position_reference=retract_aspirate.positionReference,
         _offset=retract_aspirate.offset,
         _speed=retract_aspirate.speed,
-        _air_gap_by_volume=retract_aspirate.airGapByVolume,
+        _air_gap_by_volume=LiquidHandlingPropertyByVolume(
+            retract_aspirate.airGapByVolume
+        ),
         _touch_tip=_build_touch_tip_properties(retract_aspirate.touchTip),
         _delay=_build_delay_properties(retract_aspirate.delay),
     )
@@ -474,7 +509,9 @@ def _build_retract_dispense(
         _position_reference=retract_dispense.positionReference,
         _offset=retract_dispense.offset,
         _speed=retract_dispense.speed,
-        _air_gap_by_volume=retract_dispense.airGapByVolume,
+        _air_gap_by_volume=LiquidHandlingPropertyByVolume(
+            retract_dispense.airGapByVolume
+        ),
         _blowout=_build_blowout_properties(retract_dispense.blowout),
         _touch_tip=_build_touch_tip_properties(retract_dispense.touchTip),
         _delay=_build_delay_properties(retract_dispense.delay),
@@ -489,7 +526,9 @@ def build_aspirate_properties(
         _retract=_build_retract_aspirate(aspirate_properties.retract),
         _position_reference=aspirate_properties.positionReference,
         _offset=aspirate_properties.offset,
-        _flow_rate_by_volume=aspirate_properties.flowRateByVolume,
+        _flow_rate_by_volume=LiquidHandlingPropertyByVolume(
+            aspirate_properties.flowRateByVolume
+        ),
         _pre_wet=aspirate_properties.preWet,
         _mix=_build_mix_properties(aspirate_properties.mix),
         _delay=_build_delay_properties(aspirate_properties.delay),
@@ -504,9 +543,13 @@ def build_single_dispense_properties(
         _retract=_build_retract_dispense(single_dispense_properties.retract),
         _position_reference=single_dispense_properties.positionReference,
         _offset=single_dispense_properties.offset,
-        _flow_rate_by_volume=single_dispense_properties.flowRateByVolume,
+        _flow_rate_by_volume=LiquidHandlingPropertyByVolume(
+            single_dispense_properties.flowRateByVolume
+        ),
         _mix=_build_mix_properties(single_dispense_properties.mix),
-        _push_out_by_volume=single_dispense_properties.pushOutByVolume,
+        _push_out_by_volume=LiquidHandlingPropertyByVolume(
+            single_dispense_properties.pushOutByVolume
+        ),
         _delay=_build_delay_properties(single_dispense_properties.delay),
     )
 
@@ -521,9 +564,15 @@ def build_multi_dispense_properties(
         _retract=_build_retract_dispense(multi_dispense_properties.retract),
         _position_reference=multi_dispense_properties.positionReference,
         _offset=multi_dispense_properties.offset,
-        _flow_rate_by_volume=multi_dispense_properties.flowRateByVolume,
-        _conditioning_by_volume=multi_dispense_properties.conditioningByVolume,
-        _disposal_by_volume=multi_dispense_properties.disposalByVolume,
+        _flow_rate_by_volume=LiquidHandlingPropertyByVolume(
+            multi_dispense_properties.flowRateByVolume
+        ),
+        _conditioning_by_volume=LiquidHandlingPropertyByVolume(
+            multi_dispense_properties.conditioningByVolume
+        ),
+        _disposal_by_volume=LiquidHandlingPropertyByVolume(
+            multi_dispense_properties.disposalByVolume
+        ),
         _delay=_build_delay_properties(multi_dispense_properties.delay),
     )
 

api/src/opentrons/protocol_api/validation.py

@@ -11,7 +11,7 @@
     NamedTuple,
     TYPE_CHECKING,
 )
-
+from math import isinf, isnan
 from typing_extensions import TypeGuard
 
 from opentrons_shared_data.labware.labware_definition import LabwareRole
@@ -483,3 +483,45 @@ def validate_location(
         if well is not None
         else PointTarget(location=target_location, in_place=in_place)
     )
+
+
+def ensure_boolean(value: bool) -> bool:
+    """Ensure value is a boolean."""
+    if not isinstance(value, bool):
+        raise ValueError("Value must be a boolean.")
+    return value
+
+
+def ensure_float(value: Union[int, float]) -> float:
+    """Ensure value is a float (or an integer) and return it as a float."""
+    if not isinstance(value, (int, float)):
+        raise ValueError("Value must be a floating point number.")
+    return float(value)
+
+
+def ensure_positive_float(value: Union[int, float]) -> float:
+    """Ensure value is a positive and real float value."""
+    float_value = ensure_float(value)
+    if isnan(float_value) or isinf(float_value):
+        raise ValueError("Value must be a defined, non-infinite number.")
+    if float_value < 0:
+        raise ValueError("Value must be a positive float.")
+    return float_value
+
+
+def ensure_positive_int(value: int) -> int:
+    """Ensure value is a positive integer."""
+    if not isinstance(value, int):
+        raise ValueError("Value must be an integer.")
+    if value < 0:
+        raise ValueError("Value must be a positive integer.")
+    return value
+
+
+def validate_coordinates(value: Sequence[float]) -> Tuple[float, float, float]:
+    """Ensure value is a valid sequence of 3 floats and return a tuple of 3 floats."""
+    if len(value) != 3:
+        raise ValueError("Coordinates must be a sequence of length 3")
+    if not all(isinstance(v, (float, int)) for v in value):
+        raise ValueError("All values in coordinates must be floats.")
+    return float(value[0]), float(value[1]), float(value[2])