diff --git a/autoprotocol/builders.py b/autoprotocol/builders.py index 6e6d49b6..22f262f3 100644 --- a/autoprotocol/builders.py +++ b/autoprotocol/builders.py @@ -23,17 +23,53 @@ Instruction Instructions corresponding to each of the builders """ -import enum - from collections import defaultdict from collections.abc import Iterable # pylint: disable=no-name-in-module -from dataclasses import dataclass +from dataclasses import asdict from functools import reduce from numbers import Number from typing import Any, Dict, List, Optional, Union from .constants import SBS_FORMAT_SHAPES from .container import Container, Well, WellGroup +from .types.builders import ( + DispenseBuildersShakePaths, + EvaporateBuildersBlowdownParams, + EvaporateBuildersCentrifugeParams, + EvaporateBuildersValidGases, + EvaporateBuildersValidModes, + EvaporateBuildersVortexParams, + LiquidHandleBuildersDispenseModes, + LiquidHandleBuildersLiquidClasses, + LiquidHandleBuildersZDetectionMethods, + LiquidHandleBuildersZReferences, + SpectrophotometryBuildersReadPositions, + SpectrophotometryBuildersShakePaths, + SpectrophotometryBuildersZHeuristics, + SpectrophotometryBuildersZReferences, + ThermocycleBuildersValidDyes, +) +from .types.protocol import ( + ACCELERATION, + DENSITY, + FLOW_RATE, + FREQUENCY, + LENGTH, + POWER, + TEMPERATURE, + TIME, + VELOCITY, + VOLTAGE, + VOLUME, + WAVELENGTH, + DispenseColumn, + FlowCytometryChannel, + FlowCytometryChannelEmissionFilter, + FlowCytometryChannelMeasurements, + FlowCytometryChannelTriggerLogic, + FlowCytometryCollectionConditionStopCriteria, + GelPurifyBand, +) from .unit import Unit from .util import is_valid_well, parse_unit @@ -45,7 +81,7 @@ def __init__(self): self.sbs_shapes = ["SBS96", "SBS384"] @staticmethod - def _merge_param_dicts(left=None, right=None): + def _merge_param_dicts(left: Optional[dict] = None, right: Optional[dict] = None): """Finds the union of two dicts of params and checks for duplicates Parameters @@ -87,7 +123,7 @@ def _merge_param_dicts(left=None, right=None): return unique # pylint: disable=redefined-builtin - def shape(self, rows=1, columns=1, format=None): + def shape(self, rows: int = 1, columns: int = 1, format: Optional[str] = None): """ Helper function for building a shape dictionary @@ -154,21 +190,7 @@ class ThermocycleBuilders(InstructionBuilders): def __init__(self): super(ThermocycleBuilders, self).__init__() - self.valid_dyes = { - "FAM", - "SYBR", # channel 1 - "VIC", - "HEX", - "TET", - "CALGOLD540", # channel 2 - "ROX", - "TXR", - "CALRED610", # channel 3 - "CY5", - "QUASAR670", # channel 4 - "QUASAR705", # channel 5 - "FRET", # channel 6 - } + self.valid_dyes = {option.name for option in ThermocycleBuildersValidDyes} def dyes(self, **kwargs): """Helper function for creating a dye parameter @@ -206,7 +228,7 @@ def dyes(self, **kwargs): return dyes - def dyes_from_well_map(self, well_map): + def dyes_from_well_map(self, well_map: Dict[Well, str]): """Helper function for creating a dye parameter from a well_map Take a map of wells to the dyes it contains and returns a map of dyes to @@ -238,7 +260,12 @@ def dyes_from_well_map(self, well_map): return self.dyes(**dyes) @staticmethod - def melting(start=None, end=None, increment=None, rate=None): + def melting( + start: Optional[TEMPERATURE] = None, + end: Optional[TEMPERATURE] = None, + increment: Optional[TEMPERATURE] = None, + rate: Optional[TIME] = None, + ): """Helper function for creating melting parameters Generates melt curve parameters for Thermocycle Instructions. @@ -284,7 +311,7 @@ def melting(start=None, end=None, increment=None, rate=None): return {"start": start, "end": end, "increment": increment, "rate": rate} - def group(self, steps, cycles=1): + def group(self, steps: List[dict], cycles: int = 1): """ Helper function for creating a thermocycle group, which is a series of steps repeated for the number of cycles @@ -293,7 +320,7 @@ def group(self, steps, cycles=1): ---------- steps: list(ThermocycleBuilders.step) Steps to be carried out. At least one step has to be specified. - See `ThermocycleBuilders.step` for more information + See `ThermocycleBuildeThermocycleBuilders.step` for more information cycles: int, optional Number of cycles to repeat the specified steps. Defaults to 1 @@ -335,7 +362,7 @@ def reformat_gradient(**kwargs): return group_dict @staticmethod - def step(temperature, duration, read=None): + def step(temperature: TEMPERATURE, duration: TIME, read: Optional[bool] = None): """ Helper function for creating a thermocycle step. @@ -526,14 +553,14 @@ class DispenseBuilders(InstructionBuilders): def __init__(self): super(DispenseBuilders, self).__init__() - self.SHAKE_PATHS = ["landscape_linear"] + self.SHAKE_PATHS = [option.name for option in DispenseBuildersShakePaths] @staticmethod # pragma pylint: disable=unused-argument, missing-param-doc def nozzle_position( - position_x: Optional[Unit] = None, - position_y: Optional[Unit] = None, - position_z: Optional[Unit] = None, + position_x: Optional[LENGTH] = None, + position_y: Optional[LENGTH] = None, + position_z: Optional[LENGTH] = None, ): """ Generates a validated nozzle_position parameter. @@ -561,7 +588,7 @@ def nozzle_position( # pragma pylint: enable=unused-argument # pragma pylint: disable=missing-param-doc @staticmethod - def column(column: int, volume: Union[str, Unit]): + def column(column: int, volume: VOLUME) -> DispenseColumn: """ Generates a validated column parameter. @@ -576,10 +603,12 @@ def column(column: int, volume: Union[str, Unit]): Column parameter of type {"column": int, "volume": Unit} """ - return {"column": int(column), "volume": parse_unit(volume, "uL")} + return DispenseColumn( + **{"column": int(column), "volume": parse_unit(volume, "uL")} + ) # pragma pylint: disable=missing-param-doc - def columns(self, columns: List[dict]): + def columns(self, columns: List[Union[DispenseColumn, dict]]): """ Generates a validated columns parameter. @@ -602,9 +631,12 @@ def columns(self, columns: List[dict]): if not len(columns) > 0: raise ValueError("There must be at least one column specified for columns.") - column_list = [self.column(**_) for _ in columns] + column_list: List[DispenseColumn] = [ + self.column(**asdict(_) if isinstance(_, DispenseColumn) else _) + for _ in columns + ] - if len(column_list) != len(set([_["column"] for _ in column_list])): + if len(column_list) != len(set([_.column for _ in column_list])): raise ValueError( f"Column indices must be unique, but there were duplicates " f"in {column_list}." @@ -615,10 +647,10 @@ def columns(self, columns: List[dict]): # pragma pylint: disable=missing-param-doc def shake_after( self, - duration: Optional[Union[Unit, str]], - frequency: Optional[Union[Unit, str]] = None, + duration: TIME, + frequency: Optional[FREQUENCY] = None, path: Optional[str] = None, - amplitude: Optional[Union[Unit, str]] = None, + amplitude: Optional[LENGTH] = None, ): """ Generates a validated shake_after parameter. @@ -673,34 +705,25 @@ def __init__(self): "shake": self.shake_mode_params, } - self.READ_POSITIONS = ["top", "bottom"] - + self.READ_POSITIONS = [ + option.name for option in SpectrophotometryBuildersReadPositions + ] self.SHAKE_PATHS = [ - "portrait_linear", - "landscape_linear", - "cw_orbital", - "ccw_orbital", - "portrait_down_double_orbital", - "landscape_down_double_orbital", - "portrait_up_double_orbital", - "landscape_up_double_orbital", - "cw_diamond", - "ccw_diamond", + option.name for option in SpectrophotometryBuildersShakePaths + ] + self.Z_REFERENCES = [ + option.name for option in SpectrophotometryBuildersZReferences ] - - self.Z_REFERENCES = ["plate_bottom", "plate_top", "well_bottom", "well_top"] - self.Z_HEURISTICS = [ - "max_mean_read_without_saturation", - "closest_distance_without_saturation", + option.name for option in SpectrophotometryBuildersZHeuristics ] # pragma pylint: disable=missing-param-doc @staticmethod def wavelength_selection( - shortpass: Optional[Union[Unit, str]] = None, - longpass: Optional[Union[Unit, str]] = None, - ideal: Optional[Union[Unit, str]] = None, + shortpass: Optional[WAVELENGTH] = None, + longpass: Optional[WAVELENGTH] = None, + ideal: Optional[WAVELENGTH] = None, ): """ Generates a representation of a wavelength selection by either @@ -770,12 +793,12 @@ def group(self, mode, mode_params): def absorbance_mode_params( self, - wells, - wavelength, - num_flashes=None, - settle_time=None, - read_position=None, - position_z=None, + wells: Union[List[Well], WellGroup], + wavelength: WAVELENGTH, + num_flashes: Optional[int] = None, + settle_time: Optional[TIME] = None, + read_position: Optional[SpectrophotometryBuildersReadPositions] = None, + position_z: Optional[dict] = None, ): """ Parameters @@ -851,16 +874,16 @@ def absorbance_mode_params( def fluorescence_mode_params( self, - wells, - excitation, - emission, - num_flashes=None, - settle_time=None, - lag_time=None, - integration_time=None, - gain=None, - read_position=None, - position_z=None, + wells: Union[List[Well], WellGroup], + excitation: WAVELENGTH, + emission: WAVELENGTH, + num_flashes: Optional[int] = None, + settle_time: Optional[TIME] = None, + lag_time: Optional[TIME] = None, + integration_time: Optional[TIME] = None, + gain: Optional[Union[int, float]] = None, + read_position: Optional[SpectrophotometryBuildersReadPositions] = None, + position_z: Optional[dict] = None, ): """ Parameters @@ -969,13 +992,13 @@ def fluorescence_mode_params( def luminescence_mode_params( self, - wells, - num_flashes=None, - settle_time=None, - integration_time=None, - gain=None, - read_position=None, - position_z=None, + wells: Union[List[Well], WellGroup], + num_flashes: Optional[int] = None, + settle_time: Optional[TIME] = None, + integration_time: Optional[TIME] = None, + gain: Optional[int] = None, + read_position: Optional[SpectrophotometryBuildersReadPositions] = None, + position_z: Optional[dict] = None, ): """ Parameters @@ -1060,7 +1083,11 @@ def luminescence_mode_params( return mode_params def shake_mode_params( - self, duration=None, frequency=None, path=None, amplitude=None + self, + duration: Optional[TIME] = None, + frequency: Optional[FREQUENCY] = None, + path: Optional[SpectrophotometryBuildersShakePaths] = None, + amplitude: Optional[LENGTH] = None, ): """ Parameters @@ -1087,10 +1114,10 @@ def shake_mode_params( def shake_before( self, - duration: Union[str, Unit], - frequency: Optional[Union[str, Unit]] = None, - path: Optional[str] = None, - amplitude: Optional[Union[str, Unit]] = None, + duration: TIME, + frequency: Optional[FREQUENCY] = None, + path: Optional[SpectrophotometryBuildersShakePaths] = None, + amplitude: Optional[LENGTH] = None, ): """ Parameters @@ -1116,7 +1143,13 @@ def shake_before( duration=duration, frequency=frequency, path=path, amplitude=amplitude ) - def _shake(self, duration=None, frequency=None, path=None, amplitude=None): + def _shake( + self, + duration: Optional[TIME] = None, + frequency: Optional[FREQUENCY] = None, + path: Optional[SpectrophotometryBuildersShakePaths] = None, + amplitude: Optional[LENGTH] = None, + ): """ Helper method for validating shake params. """ @@ -1145,7 +1178,11 @@ def _shake(self, duration=None, frequency=None, path=None, amplitude=None): return params - def position_z_manual(self, reference=None, displacement=None): + def position_z_manual( + self, + reference: Optional[SpectrophotometryBuildersZReferences] = None, + displacement: Optional[LENGTH] = None, + ): """Helper for building position_z parameters for a manual position_z configuration @@ -1181,7 +1218,7 @@ def position_z_manual(self, reference=None, displacement=None): return {"manual": {"reference": reference, "displacement": displacement}} - def position_z_calculated(self, wells, heuristic=None): + def position_z_calculated(self, wells: List[Well], heuristic: Optional[str] = None): """Helper for building position_z parameters for a calculated position_z configuration @@ -1219,7 +1256,7 @@ def position_z_calculated(self, wells, heuristic=None): return {"calculated_from_wells": {"wells": wells, "heuristic": heuristic}} - def _position_z(self, position_z): + def _position_z(self, position_z: dict): """ Helper method for validating position_z params """ @@ -1249,18 +1286,23 @@ class LiquidHandleBuilders(InstructionBuilders): def __init__(self): super(LiquidHandleBuilders, self).__init__() - self.liquid_classes = ["air", "default", "viscous", "protein_buffer"] + self.liquid_classes = [ + option.name for option in LiquidHandleBuildersLiquidClasses + ] self.xy_max = 1 - self.z_references = [ - "well_top", - "well_bottom", - "liquid_surface", - "preceding_position", + self.z_references = [option.name for option in LiquidHandleBuildersZReferences] + self.z_detection_methods = [ + option.name for option in LiquidHandleBuildersZDetectionMethods + ] + self.dispense_modes = [ + option.name for option in LiquidHandleBuildersDispenseModes ] - self.z_detection_methods = ["capacitance", "pressure", "tracked"] - self.dispense_modes = ["air_displacement", "positive_displacement"] - def location(self, location=None, transports=None): + def location( + self, + location: Optional[Union[Well, str]] = None, + transports: Optional[List[dict]] = None, + ): """Helper for building locations Parameters @@ -1301,12 +1343,12 @@ def location(self, location=None, transports=None): def transport( self, - volume=None, - pump_override_volume=None, - flowrate=None, - delay_time=None, - mode_params=None, - density=None, + volume: Optional[VOLUME] = None, + pump_override_volume: Optional[VOLUME] = None, + flowrate: Optional[dict] = None, + delay_time: Optional[TIME] = None, + mode_params: Optional[dict] = None, + density: Optional[DENSITY] = None, ): """Helper for building transports @@ -1358,7 +1400,11 @@ def transport( @staticmethod def flowrate( - target, initial=None, cutoff=None, acceleration=None, deceleration=None + target: FLOW_RATE, + initial: Optional[FLOW_RATE] = None, + cutoff: Optional[FLOW_RATE] = None, + acceleration: Optional[ACCELERATION] = None, + deceleration: Optional[ACCELERATION] = None, ): """Helper for building flowrates @@ -1400,12 +1446,12 @@ def flowrate( def mode_params( self, - liquid_class=None, - position_x=None, - position_y=None, - position_z=None, - tip_position=None, - volume_resolution=None, + liquid_class: Optional[str] = None, + position_x: Optional[dict] = None, + position_y: Optional[dict] = None, + position_z: Optional[dict] = None, + tip_position: Optional[dict] = None, + volume_resolution: Optional[VOLUME] = None, ): """Helper for building transport mode_params @@ -1485,9 +1531,9 @@ def mode_params( @staticmethod def device_mode_params( - model=None, - chip_material=None, - nozzle=None, + model: Optional[str] = None, + chip_material: Optional[str] = None, + nozzle: Optional[str] = None, ): """Helper for building device level mode_params @@ -1545,7 +1591,9 @@ def device_mode_params( return device_mode_params @staticmethod - def move_rate(target=None, acceleration=None): + def move_rate( + target: Optional[VELOCITY] = None, acceleration: Optional[ACCELERATION] = None + ): """Helper for building move_rates Parameters @@ -1567,7 +1615,11 @@ def move_rate(target=None, acceleration=None): return {"target": target, "acceleration": acceleration} - def position_xy(self, position=None, move_rate=None): + def position_xy( + self, + position: Optional[Union[int, float]] = None, + move_rate: Optional[dict] = None, + ): """Helper for building position_x and position_y parameters Parameters @@ -1606,14 +1658,14 @@ def position_xy(self, position=None, move_rate=None): def position_z( self, - reference=None, - offset=None, - move_rate=None, - detection_method=None, - detection_threshold=None, - detection_duration=None, - detection_fallback=None, - detection=None, + reference: Optional[str] = None, + offset: Optional[LENGTH] = None, + move_rate: Optional[dict] = None, + detection_method: Optional[str] = None, + detection_threshold: Optional[Union[Unit, str]] = None, + detection_duration: Optional[TIME] = None, + detection_fallback: Optional[dict] = None, + detection: Optional[dict] = None, ): """Helper for building position_z parameters @@ -1721,7 +1773,7 @@ def position_z( } @staticmethod - def instruction_mode_params(tip_type=None): + def instruction_mode_params(tip_type: Optional[str] = None): """Helper for building instruction mode_params Parameters @@ -1738,7 +1790,14 @@ def instruction_mode_params(tip_type=None): return {"tip_type": tip_type} - def mix(self, volume, repetitions, initial_z, asp_flowrate=None, dsp_flowrate=None): + def mix( + self, + volume: VOLUME, + repetitions: int, + initial_z: dict, + asp_flowrate: Optional[dict] = None, + dsp_flowrate: Optional[dict] = None, + ): """Helper for building mix params for Transfer LiquidHandleMethods Parameters @@ -1784,7 +1843,7 @@ def mix(self, volume, repetitions, initial_z, asp_flowrate=None, dsp_flowrate=No "dsp_flowrate": dsp_flowrate, } - def blowout(self, volume, initial_z, flowrate=None): + def blowout(self, volume: VOLUME, initial_z: dict, flowrate: Optional[dict] = None): """Helper for building blowout params for LiquidHandleMethods Parameters @@ -1810,7 +1869,9 @@ def blowout(self, volume, initial_z, flowrate=None): return {"volume": volume, "initial_z": initial_z, "flowrate": flowrate} - def desired_mode(self, transports=None, mode=None): + def desired_mode( + self, transports: Optional[dict] = None, mode: Optional[str] = None + ): """Helper for selecting dispense mode based on liquid_class name For non-viscous, water-like liquid and air, the method will default to "air_displacement". To allow for more accurate aspirate and @@ -1958,7 +2019,11 @@ class PlateReaderBuilders(InstructionBuilders): """Helpers for building parameters for plate reading instructions""" def incubate_params( - self, duration, shake_amplitude=None, shake_orbital=None, shaking=None + self, + duration: TIME, + shake_amplitude: Optional[LENGTH] = None, + shake_orbital: Optional[bool] = None, + shaking: Optional[dict] = None, ): """ Create a dictionary with incubation parameters which can be used as @@ -2023,13 +2088,6 @@ def incubate_params( } -class EvaporateModeParamsMode(enum.Enum): - rotate = enum.auto() - centrifuge = enum.auto() - vortex = enum.auto() - blowdown = enum.auto() - - class EvaporateBuilders(InstructionBuilders): """ Helpers for building Evaporate instructions @@ -2037,28 +2095,21 @@ class EvaporateBuilders(InstructionBuilders): def __init__(self): super(EvaporateBuilders, self).__init__() - self.valid_modes = ["rotate", "centrifuge", "vortex", "blowdown"] - self.valid_gases = ["nitrogen", "argon", "helium"] - self.rotate_params = [ - "flask_volume", - "rotation_speed", - "vacuum_pressure", - "condenser_temperature", - ] + self.valid_modes = [option.name for option in EvaporateBuildersValidModes] + self.valid_gases = [option.name for option in EvaporateBuildersValidGases] + from autoprotocol.types.builders import EvaporateBuildersRotateParams + + self.rotate_params = [option.name for option in EvaporateBuildersRotateParams] self.centrifuge_params = [ - "spin_acceleration", - "vacuum_pressure", - "condenser_temperature", + option.name for option in EvaporateBuildersCentrifugeParams ] - self.vortex_params = [ - "vortex_speed", - "vacuum_pressure", - "condenser_temperature", + self.vortex_params = [option.name for option in EvaporateBuildersVortexParams] + self.blowdown_params = [ + option.name for option in EvaporateBuildersBlowdownParams ] - self.blowdown_params = ["gas", "blow_rate", "vortex_speed"] def get_mode_params( - self, mode: EvaporateModeParamsMode, mode_params: Dict[str, Any] + self, mode: Union[EvaporateBuildersValidModes, str], mode_params: Dict[str, Any] ): """ Checks on the validity of mode and mode_params, and @@ -2161,22 +2212,6 @@ def get_mode_params( return mode_param_output -@dataclass() -class GelPurifyBandSizeRange: - min_bp: int - max_bp: int - - -@dataclass -class GelPurifyBand: - elution_buffer: str - elution_volume: Union[str, Unit] - destination: Well - min_bp: Optional[int] - max_bp: Optional[int] - band_size_range: Optional[GelPurifyBandSizeRange] - - class GelPurifyBuilders(InstructionBuilders): """Helpers for building GelPurify instructions""" @@ -2217,18 +2252,21 @@ def extract( if not isinstance(band_list, list): band_list = [band_list] - band_list = [self.band(**i) for i in band_list] + band_list = [ + self.band(**asdict(i) if isinstance(i, GelPurifyBand) else i) + for i in band_list + ] return {"source": source, "band_list": band_list, "lane": lane, "gel": gel} def band( self, - elution_buffer, - elution_volume, - destination, - min_bp=None, - max_bp=None, - band_size_range=None, + elution_buffer: str, + elution_volume: VOLUME, + destination: Well, + min_bp: Optional[int] = None, + max_bp: Optional[int] = None, + band_size_range: Optional[dict] = None, ): """Helper for building band params for gel_purify @@ -2298,7 +2336,7 @@ class MagneticTransferBuilders(InstructionBuilders): """Helpers for building MagneticTransfer instruction parameters""" @staticmethod - def mag_dry(object, duration): + def mag_dry(object: Container, duration: TIME): """Helper for building mag_dry sub operations for MagneticTransfer Parameters @@ -2329,7 +2367,13 @@ def mag_dry(object, duration): return {"object": object, "duration": duration} @staticmethod - def mag_incubate(object, duration, magnetize, tip_position, temperature=None): + def mag_incubate( + object: Container, + duration: TIME, + magnetize: bool, + tip_position: float, + temperature: Optional[TEMPERATURE] = None, + ): """Helper for building mag_incubate sub operations for MagneticTransfer Parameters @@ -2384,7 +2428,11 @@ def mag_incubate(object, duration, magnetize, tip_position, temperature=None): @staticmethod def mag_collect( - object, cycles, pause_duration, bottom_position=None, temperature=None + object: Container, + cycles: int, + pause_duration: TIME, + bottom_position: Optional[float] = None, + temperature: Optional[TEMPERATURE] = None, ): """Helper for building mag_collect sub operations for MagneticTransfer @@ -2440,7 +2488,12 @@ def mag_collect( @staticmethod def mag_release( - object, duration, frequency, center=None, amplitude=None, temperature=None + object: Container, + duration: TIME, + frequency: FREQUENCY, + center: Optional[float] = None, + amplitude: Optional[float] = None, + temperature: Optional[TEMPERATURE] = None, ): """Helper for building mag_release sub operations for MagneticTransfer @@ -2506,13 +2559,13 @@ def mag_release( @staticmethod def mag_mix( - object, - duration, - frequency, - center=None, - amplitude=None, - magnetize=None, - temperature=None, + object: Container, + duration: TIME, + frequency: FREQUENCY, + center: Optional[float] = None, + amplitude: Optional[float] = None, + magnetize: Optional[bool] = None, + temperature: Optional[TEMPERATURE] = None, ): """Helper for building mag_mix sub operations for MagneticTransfer @@ -2586,51 +2639,6 @@ def mag_mix( } -class FlowCytometryChannelTriggerLogicEnum(enum.Enum): - and_ = enum.auto() - or_ = enum.auto() - - -@dataclass -class FlowCytometryChannelTriggerLogic: - value: FlowCytometryChannelTriggerLogicEnum - - def __post_init__(self): - trigger_modes = ("and_", "or_") - if self.value is not None and self.value not in trigger_modes: - raise ValueError(f"trigger_logic must be one of {trigger_modes}.") - - -@dataclass -class FlowCytometryChannelMeasurments: - area: Optional[bool] = None - height: Optional[bool] = None - width: Optional[bool] = None - - -@dataclass -class FlowCytometryChannelEmissionFilter: - channel_name: str - shortpass: Union[str, Unit] = None - longpass: Union[str, Unit] = None - - -@dataclass -class FlowCytometryChannel: - emission_filter: FlowCytometryChannelEmissionFilter - detector_gain: Union[str, Unit] - measurements: Optional[FlowCytometryChannelMeasurments] = None - trigger_threshold: Optional[int] = None - trigger_logic: Optional[FlowCytometryChannelTriggerLogic] = None - - -@dataclass() -class FlowCytometryCollectionConditionStopCriteria: - volume: Optional[Union[str, Unit]] = None - events: Optional[int] = None - time: Union[str, Unit] = None - - class FlowCytometryBuilders(InstructionBuilders): """ Builders for FlowCytometry instructions. @@ -2643,8 +2651,8 @@ def __init__(self): def laser( self, channels: List[FlowCytometryChannel], - excitation: Union[str, Unit] = None, - power: Union[str, Unit] = None, + excitation: Optional[WAVELENGTH] = None, + power: Optional[POWER] = None, area_scaling_factor: Optional[int] = None, ): """ @@ -2694,7 +2702,10 @@ def laser( if excitation is not None: excitation = parse_unit(excitation, "nanometers") - channels = [self.channel(**c) for c in channels] + channels = [ + self.channel(**asdict(c) if isinstance(c, FlowCytometryChannel) else c) + for c in channels + ] # Gating modes do not allow specification of excitation parameter channel_names = set( @@ -2715,8 +2726,8 @@ def laser( def channel( self, emission_filter: FlowCytometryChannelEmissionFilter, - detector_gain: Union[str, Unit], - measurements: Optional[FlowCytometryChannelMeasurments] = None, + detector_gain: VOLTAGE, + measurements: Optional[FlowCytometryChannelMeasurements] = None, trigger_threshold: Optional[int] = None, trigger_logic: Optional[FlowCytometryChannelTriggerLogic] = None, ): @@ -2755,9 +2766,17 @@ def channel( if measurements is None: measurements = self.measurements() else: - measurements = self.measurements(**measurements) + measurements = self.measurements( + **asdict(measurements) + if isinstance(measurements, FlowCytometryChannelMeasurements) + else measurements + ) - emission_filter = self.emission_filter(**emission_filter) + emission_filter = self.emission_filter( + **asdict(emission_filter) + if isinstance(emission_filter, FlowCytometryChannelEmissionFilter) + else emission_filter + ) detector_gain = parse_unit(detector_gain, "millivolts") return { @@ -2771,8 +2790,8 @@ def channel( def emission_filter( self, channel_name: str, - shortpass: Union[str, Unit] = None, - longpass: Union[str, Unit] = None, + shortpass: Optional[WAVELENGTH] = None, + longpass: Optional[WAVELENGTH] = None, ): """ Generates a dict of emission filter parameters. @@ -2853,13 +2872,13 @@ def measurements( def collection_conditions( self, - acquisition_volume: Union[str, Unit], - flowrate: Union[str, Unit], - wait_time: Union[str, Unit], + acquisition_volume: VOLUME, + flowrate: FLOW_RATE, + wait_time: TIME, mix_cycles: int, - mix_volume: Union[str, Unit], + mix_volume: VOLUME, rinse_cycles: int, - stop_criteria: Optional[FlowCytometryCollectionConditionStopCriteria], + stop_criteria: Optional[FlowCytometryCollectionConditionStopCriteria] = None, ): """ Generates a dict of collection_conditions parameters. @@ -2907,7 +2926,13 @@ def collection_conditions( if stop_criteria is None: stop_criteria = self.stop_criteria(volume=acquisition_volume) else: - stop_criteria = self.stop_criteria(**stop_criteria) + stop_criteria = self.stop_criteria( + **asdict(stop_criteria) + if isinstance( + stop_criteria, FlowCytometryCollectionConditionStopCriteria + ) + else stop_criteria + ) return { "acquisition_volume": acquisition_volume, @@ -2921,9 +2946,9 @@ def collection_conditions( @staticmethod def stop_criteria( - volume: Optional[Union[str, Unit]] = None, + volume: Optional[VOLUME] = None, events: Optional[int] = None, - time: Union[str, Unit] = None, + time: Optional[TIME] = None, ): """ Generates a dict of stop_criteria parameters. diff --git a/autoprotocol/protocol.py b/autoprotocol/protocol.py index f21204e1..899c0481 100644 --- a/autoprotocol/protocol.py +++ b/autoprotocol/protocol.py @@ -9,19 +9,114 @@ import json import warnings -from typing import Dict, List, Tuple +from collections import defaultdict +from dataclasses import asdict, dataclass, field +from numbers import Number +from typing import Any, Dict, List, Optional, Tuple, Union +from .builders import LiquidHandleBuilders from .compound import Compound from .constants import AGAR_CLLD_THRESHOLD, SPREAD_PATH -from .container import COVER_TYPES, SEAL_TYPES, Container, Well +from .container import COVER_TYPES, SEAL_TYPES, Container, Well, WellGroup from .container_type import _CONTAINER_TYPES, ContainerType -from .instruction import * # pylint: disable=unused-wildcard-import +from .informatics import AttachCompounds, Informatics +from .instruction import ( + SPE, + Absorbance, + AcousticTransfer, + Agitate, + Autopick, + CountCells, + Cover, + Dispense, + Evaporate, + FlashFreeze, + FlowAnalyze, + FlowCytometry, + Fluorescence, + GelPurify, + GelSeparate, + IlluminaSeq, + Image, + ImagePlate, + Incubate, + Instruction, + LiquidHandle, + Luminescence, + MagneticTransfer, + MeasureConcentration, + MeasureMass, + MeasureVolume, + Oligosynthesize, + Provision, + SangerSeq, + Seal, + Sonicate, + Spectrophotometry, + Spin, + Thermocycle, + Uncover, + Unseal, +) from .liquid_handle import Dispense as DispenseMethod from .liquid_handle import LiquidClass, Mix, Transfer -from .types.protocol import * # pylint: disable=unused-wildcard-import +from .types.protocol import ( + ACCELERATION, + AMOUNT_CONCENTRATION, + DENSITY, + FLOW_RATE, + FREQUENCY, + TEMPERATURE, + TIME, + VOLUME, + WAVELENGTH, + AgitateMode, + AgitateModeParams, + AgitateModeParamsBarShape, + AutopickGroup, + DispenseColumn, + DispenseNozzlePosition, + DispenseShakeAfter, + DispenseShape, + EvaporateMode, + EvaporateModeParams, + FlowAnalyzeChannel, + FlowAnalyzeColors, + FlowAnalyzeNegControls, + FlowAnalyzePosControls, + FlowAnalyzeSample, + FlowCytometryCollectionCondition, + FlowCytometryLaser, + GelPurifyExtract, + IlluminaSeqLane, + ImageExposure, + ImageMode, + IncubateShakingParams, + OligosynthesizeOligo, + PlateReaderIncubateBefore, + PlateReaderPositionZCalculated, + PlateReaderPositionZManual, + SonicateMode, + SonicateModeParamsBath, + SonicateModeParamsHorn, + SpectrophotometryShakeBefore, + SpeElute, + SpeLoadSample, + SpeParams, + ThermocycleTemperature, + ThermocycleTemperatureGradient, + TimeConstraint, + TimeConstraintFromToDict, + WellParam, +) from .types.ref import Ref, RefOpts, StorageLocation from .unit import Unit, UnitError -from .util import _check_container_type_with_shape, _validate_as_instance +from .util import ( + _check_container_type_with_shape, + _validate_as_instance, + is_valid_well, + parse_unit, +) @dataclass @@ -2222,10 +2317,10 @@ def dispense( if not isinstance(ref, Container): raise TypeError(f"ref must be a Container but it was {type(ref)}.") - columns = Dispense.builders.columns(columns) + columns: List[DispenseColumn] = Dispense.builders.columns(columns) ref_cols = list(range(ref.container_type.col_count)) - if not all(_["column"] in ref_cols for _ in columns): + if not all(_.column in ref_cols for _ in columns): raise ValueError( f"Specified dispense columns: {columns} contains a column index that is outside of the valid columns: {ref_cols} for ref: {ref}." ) @@ -2234,13 +2329,23 @@ def dispense( if flowrate is not None: flowrate = parse_unit(flowrate, "uL/s") if nozzle_position is not None: - nozzle_position = Dispense.builders.nozzle_position(**nozzle_position) + nozzle_position = Dispense.builders.nozzle_position( + **asdict(nozzle_position) + if isinstance(nozzle_position, DispenseNozzlePosition) + else nozzle_position + ) if pre_dispense is not None: pre_dispense = parse_unit(pre_dispense, "uL") if shape is not None: - shape = Dispense.builders.shape(**shape) + shape = Dispense.builders.shape( + **asdict(shape) if isinstance(shape, DispenseShape) else shape + ) if shake_after is not None: - shake_after = Dispense.builders.shake_after(**shake_after) + shake_after = Dispense.builders.shake_after( + **asdict(shake_after) + if isinstance(shake_after, DispenseShakeAfter) + else shake_after + ) nozzle_count = ( shape["rows"] * shape["columns"] if shape else _DEFAULT_NOZZLE_COUNT @@ -2255,7 +2360,7 @@ def dispense( ) for c in columns: - if c["volume"] % step_size != Unit("0:uL"): + if c.volume % step_size != Unit("0:uL"): raise ValueError( f"Dispense volume must be a multiple of the step size {step_size}, but column {c} does not meet these requirements." ) @@ -2276,7 +2381,7 @@ def dispense( self._remove_cover(reagent.container, "dispense from") # Volume accounting - total_vol_dispensed = sum([Unit(c["volume"]) for c in columns]) * row_count + total_vol_dispensed = sum([Unit(c.volume) for c in columns]) * row_count if pre_dispense is not None: total_vol_dispensed += nozzle_count * pre_dispense if reagent.volume: @@ -2300,12 +2405,12 @@ def dispense( self._remove_cover(ref, "dispense to") for c in columns: - wells = ref.wells_from(c["column"], row_count, columnwise=True) + wells = ref.wells_from(c.column, row_count, columnwise=True) for w in wells: if w.volume: - w.volume += c["volume"] + w.volume += c.volume else: - w.volume = c["volume"] + w.volume = c.volume return self._append_and_return( Dispense( @@ -2693,8 +2798,8 @@ def agitate( If ref cannot be undergo agitate mode `roll` or `invert` """ - valid_modes = AgitateMode.__dict__.get("_member_names_") - valid_bar_shapes = AgitateModeParamsBarShape.__dict__.get("_member_names_") + valid_modes = [option.name for option in AgitateMode] + valid_bar_shapes = [option.name for option in AgitateModeParamsBarShape] valid_bar_mode_params = ["wells", "bar_shape", "bar_length"] speed = parse_unit(speed) @@ -6335,7 +6440,7 @@ def image( ValueError Invalid exposure parameter supplied """ - allowed_image_modes = ImageMode.__dict__.get("_member_names_") + allowed_image_modes = [option.name for option in ImageMode] if not mode in allowed_image_modes: raise ValueError(f"image mode must be one of: {allowed_image_modes}") if num_images <= 0: diff --git a/autoprotocol/types/builders.py b/autoprotocol/types/builders.py new file mode 100644 index 00000000..00992b14 --- /dev/null +++ b/autoprotocol/types/builders.py @@ -0,0 +1,114 @@ +import enum + + +class EvaporateBuildersValidModes(enum.Enum): + rotate = enum.auto() + centrifuge = enum.auto() + vortex = enum.auto() + blowdown = enum.auto() + + +class EvaporateBuildersValidGases(enum.Enum): + nitrogen = enum.auto() + argon = enum.auto() + helium = enum.auto() + + +class EvaporateBuildersRotateParams(enum.Enum): + flask_volume = enum.auto() + rotation_speed = enum.auto() + vacuum_pressure = enum.auto() + condenser_temperature = enum.auto() + + +class EvaporateBuildersCentrifugeParams(enum.Enum): + spin_acceleration = enum.auto() + vacuum_pressure = enum.auto() + condenser_temperature = enum.auto() + + +class EvaporateBuildersVortexParams(enum.Enum): + vortex_speed = enum.auto() + vacuum_pressure = enum.auto() + condenser_temperature = enum.auto() + + +class EvaporateBuildersBlowdownParams(enum.Enum): + gas = enum.auto() + blow_rate = enum.auto() + vortex_speed = enum.auto() + + +class LiquidHandleBuildersLiquidClasses(enum.Enum): + air = enum.auto() + default = enum.auto() + viscous = enum.auto() + protein_buffer = enum.auto() + + +class LiquidHandleBuildersZReferences(enum.Enum): + well_top = enum.auto() + well_bottom = enum.auto() + liquid_surface = enum.auto() + preceding_position = enum.auto() + + +class LiquidHandleBuildersZDetectionMethods(enum.Enum): + capacitance = enum.auto() + pressure = enum.auto() + tracked = enum.auto() + + +class LiquidHandleBuildersDispenseModes(enum.Enum): + air_displacement = enum.auto() + positive_displacement = enum.auto() + + +class ThermocycleBuildersValidDyes(enum.Enum): + FAM = enum.auto() + SYBR = enum.auto() # channel 1 + VIC = enum.auto() + HEX = enum.auto() + TET = enum.auto() + CALGOLD540 = enum.auto() # channel 2 + ROX = enum.auto() + TXR = enum.auto() + CALRED610 = enum.auto() # channel 3 + CY5 = enum.auto() + QUASAR670 = enum.auto() # channel 4 + QUASAR705 = enum.auto() # channel 5 + FRET = enum.auto() # channel 6 + + +class DispenseBuildersShakePaths(enum.Enum): + landscape_linear = enum.auto() + + +class SpectrophotometryBuildersReadPositions(enum.Enum): + top = enum.auto() + bottom = enum.auto() + + +class SpectrophotometryBuildersZHeuristics(enum.Enum): + max_mean_read_without_saturation = enum.auto() + closest_distance_without_saturation = enum.auto() + + +class SpectrophotometryBuildersZReferences(enum.Enum): + plate_bottom = enum.auto() + plate_top = enum.auto() + well_bottom = enum.auto() + well_top = enum.auto() + + +class SpectrophotometryBuildersShakePaths(enum.Enum): + portrait_linear = enum.auto() + landscape_linear = enum.auto() + cw_orbital = enum.auto() + ccw_orbital = enum.auto() + portrait_down_double_orbital = enum.auto() + landscape_down_double_orbital = enum.auto() + portrait_up_double_orbital = enum.auto() + landscape_up_double_orbital = enum.auto() + cw_diamond = enum.auto() + ccw_diamond = enum.auto() diff --git a/autoprotocol/types/protocol.py b/autoprotocol/types/protocol.py index ed1f0915..5d7634ab 100644 --- a/autoprotocol/types/protocol.py +++ b/autoprotocol/types/protocol.py @@ -3,11 +3,6 @@ from dataclasses import dataclass, field from typing import Dict, List, Optional, Union -from ..builders import ( - FlowCytometryChannel, - FlowCytometryCollectionConditionStopCriteria, - GelPurifyBand, -) from ..container import Container, Well, WellGroup from ..unit import Unit @@ -28,23 +23,31 @@ WAVELENGTH = Union[str, Unit] DENSITY = Union[str, Unit] POWER = Union[str, Unit] +VOLTAGE = Union[str, Unit] + + +class HashableMixin: + """Allows class to be accessed like a dictionary""" + + def __getitem__(self, item): + return getattr(self, item) @dataclass -class AutopickGroup: +class AutopickGroup(HashableMixin): source: WellParam destination: WellParam min_abort: int = 0 @dataclass -class DispenseColumn: +class DispenseColumn(HashableMixin): column: int volume: VOLUME @dataclass -class IncubateShakingParams: +class IncubateShakingParams(HashableMixin): path: Union[str, Unit] frequency: FREQUENCY @@ -55,8 +58,8 @@ class TimeConstraintOptimizationCost(enum.Enum): exponential = enum.auto() -@dataclass(frozen=False) -class TimeConstraint: +@dataclass +class TimeConstraint(HashableMixin): from_: Union[int, Container] to: Union[int, Container] = field(default=None) less_than: Optional[TIME] = field(default=None) @@ -65,13 +68,13 @@ class TimeConstraint: optimization_cost: TimeConstraintOptimizationCost = field(default=None) -class TimeConstraintState: +class TimeConstraintState(enum.Enum): start = enum.auto() end = enum.auto() @dataclass -class TimeConstraintFromToDict: +class TimeConstraintFromToDict(HashableMixin): mark: Dict[str, Union[Container, str, int]] state: TimeConstraintState @@ -90,7 +93,7 @@ class OligosynthesizeOligoScale(enum.Enum): @dataclass -class OligosynthesizeOligo: +class OligosynthesizeOligo(HashableMixin): destination: Well sequence: str scale: OligosynthesizeOligoScale @@ -98,14 +101,13 @@ class OligosynthesizeOligo: def __post_init__(self): allowable_scales = [ - allowable.strip("_") - for allowable in OligosynthesizeOligoScale.__dict__.get("_member_names_") + allowable.name.strip("_") for allowable in OligosynthesizeOligoScale ] if self.scale not in allowable_scales: raise ValueError(f"Scale entered {self.scale} not in {allowable_scales}") - allowable_purification = OligosynthesizeOligoPurification.__dict__.get( - "_member_names_" - ) + allowable_purification = [ + option.name for option in OligosynthesizeOligoPurification + ] if self.purification not in allowable_purification: raise ValueError( f"Purification entered {self.purification} not in {allowable_purification}" @@ -113,7 +115,7 @@ def __post_init__(self): @dataclass -class IlluminaSeqLane: +class IlluminaSeqLane(HashableMixin): object: Well library_concentration: float @@ -131,13 +133,13 @@ class AgitateModeParamsBarShape(enum.Enum): @dataclass -class AgitateModeParams: +class AgitateModeParams(HashableMixin): wells: Union[List[Well], WellGroup] bar_shape: AgitateModeParamsBarShape bar_length: LENGTH def __post_init__(self): - allowable_bar_shape = AgitateModeParamsBarShape.__dict__.get("_member_names_") + allowable_bar_shape = [option.name for option in AgitateModeParamsBarShape] if self.bar_shape not in allowable_bar_shape: raise ValueError( f"bar_shape entered {self.bar_shape} not in {allowable_bar_shape}" @@ -145,27 +147,27 @@ def __post_init__(self): @dataclass -class ThermocycleTemperature: +class ThermocycleTemperature(HashableMixin): duration: TIME temperature: TEMPERATURE read: bool = field(default=False) @dataclass -class TemperatureGradient: +class TemperatureGradient(HashableMixin): top: TEMPERATURE bottom: TEMPERATURE @dataclass -class ThermocycleTemperatureGradient: +class ThermocycleTemperatureGradient(HashableMixin): duration: TIME gradient: TemperatureGradient read: bool = field(default=False) @dataclass -class PlateReaderIncubateBefore: +class PlateReaderIncubateBefore(HashableMixin): duration: TIME shake_amplitude: Optional[LENGTH] = field(default=None) shake_orbital: Optional[bool] = field(default=None) @@ -173,33 +175,94 @@ class PlateReaderIncubateBefore: @dataclass -class PlateReaderPositionZManual: +class PlateReaderPositionZManual(HashableMixin): manual: LENGTH @dataclass -class PlateReaderPositionZCalculated: +class PlateReaderPositionZCalculated(HashableMixin): calculated_from_wells: List[Well] @dataclass -class GelPurifyExtract: +class GelPurifyBandSizeRange(HashableMixin): + min_bp: int + max_bp: int + + +@dataclass +class GelPurifyBand(HashableMixin): + elution_buffer: str + elution_volume: Union[str, Unit] + destination: Well + min_bp: Optional[int] + max_bp: Optional[int] + band_size_range: Optional[GelPurifyBandSizeRange] + + +@dataclass +class GelPurifyExtract(HashableMixin): source: Well band_list: List[GelPurifyBand] lane: Optional[int] = field(default=None) gel: Optional[int] = field(default=None) +class FlowCytometryChannelTriggerLogicEnum(enum.Enum): + and_ = enum.auto() + or_ = enum.auto() + + +@dataclass +class FlowCytometryChannelTriggerLogic(HashableMixin): + value: FlowCytometryChannelTriggerLogicEnum + + def __post_init__(self): + trigger_modes = ("and_", "or_") + if self.value is not None and self.value not in trigger_modes: + raise ValueError(f"trigger_logic must be one of {trigger_modes}.") + + +@dataclass +class FlowCytometryChannelMeasurements(HashableMixin): + area: Optional[bool] = None + height: Optional[bool] = None + width: Optional[bool] = None + + +@dataclass +class FlowCytometryChannelEmissionFilter(HashableMixin): + channel_name: str + shortpass: WAVELENGTH = None + longpass: WAVELENGTH = None + + +@dataclass +class FlowCytometryChannel(HashableMixin): + emission_filter: FlowCytometryChannelEmissionFilter + detector_gain: VOLTAGE + measurements: Optional[FlowCytometryChannelMeasurements] = None + trigger_threshold: Optional[int] = None + trigger_logic: Optional[FlowCytometryChannelTriggerLogic] = None + + +@dataclass() +class FlowCytometryCollectionConditionStopCriteria(HashableMixin): + volume: Optional[VOLUME] = None + events: Optional[int] = None + time: Optional[TIME] = None + + @dataclass -class FlowCytometryLaser: +class FlowCytometryLaser(HashableMixin): channels: List[FlowCytometryChannel] - excitation: Union[str, Unit] = field(default=None) - power: Union[str, Unit] = field(default=None) + excitation: Optional[WAVELENGTH] = field(default=None) + power: Optional[POWER] = field(default=None) area_scaling_factor: Optional[int] = field(default=None) @dataclass -class FlowCytometryCollectionCondition: +class FlowCytometryCollectionCondition(HashableMixin): acquisition_volume: Union[str, Unit] flowrate: Union[str, Unit] wait_time: Union[str, Unit] @@ -210,13 +273,13 @@ class FlowCytometryCollectionCondition: @dataclass -class FlowAnalyzeChannelVoltageRange: +class FlowAnalyzeChannelVoltageRange(HashableMixin): low: Union[str, Unit] high: Union[str, Unit] @dataclass -class FlowAnalyzeChannel: +class FlowAnalyzeChannel(HashableMixin): voltage_range: FlowAnalyzeChannelVoltageRange area: bool height: bool @@ -224,7 +287,7 @@ class FlowAnalyzeChannel: @dataclass -class FlowAnalyzeNegControls: +class FlowAnalyzeNegControls(HashableMixin): well: Well volume: Union[str, Unit] channel: str @@ -232,14 +295,14 @@ class FlowAnalyzeNegControls: @dataclass -class FlowAnalyzeSample: +class FlowAnalyzeSample(HashableMixin): well: Well volume: Union[str, Unit] captured_events: int @dataclass -class FlowAnalyzeColors: +class FlowAnalyzeColors(HashableMixin): name: str emission_wavelength: Union[str, Unit] excitation_wavelength: Union[str, Unit] @@ -250,13 +313,13 @@ class FlowAnalyzeColors: @dataclass(frozen=True) -class FlowAnalyzePosControlsMinimizeBleed: +class FlowAnalyzePosControlsMinimizeBleed(HashableMixin): from_: FlowAnalyzeColors to: FlowAnalyzeColors @dataclass -class FlowAnalyzePosControls: +class FlowAnalyzePosControls(HashableMixin): well: Well volume: Union[str, Unit] channel: str @@ -265,7 +328,7 @@ class FlowAnalyzePosControls: @dataclass -class SpectrophotometryShakeBefore: +class SpectrophotometryShakeBefore(HashableMixin): duration: TIME frequency: Optional[Union[str, Unit]] = field(default=None) path: Optional[str] = field(default=None) @@ -279,7 +342,7 @@ class EvaporateModeParamsGas(enum.Enum): @dataclass -class EvaporateModeParams: +class EvaporateModeParams(HashableMixin): gas: EvaporateModeParamsGas vortex_speed: Union[str, Unit] blow_rate: Union[str, Unit] @@ -293,7 +356,7 @@ class EvaporateMode(enum.Enum): @dataclass -class SpeElute: +class SpeElute(HashableMixin): loading_flowrate: Union[str, Unit] resource_id: str settle_time: Union[str, Unit] @@ -304,7 +367,7 @@ class SpeElute: @dataclass -class SpeLoadSample: +class SpeLoadSample(HashableMixin): volume: Union[str, Unit] loading_flowrate: Union[str, Unit] settle_time: Optional[bool] @@ -316,7 +379,7 @@ class SpeLoadSample: @dataclass -class SpeParams: +class SpeParams(HashableMixin): volume: Union[str, Unit] loading_flowrate: Union[str, Unit] settle_time: Optional[bool] @@ -334,32 +397,32 @@ class ImageMode(enum.Enum): @dataclass -class ImageExposure: +class ImageExposure(HashableMixin): shutter_speed: Optional[Unit] = field(default=None) iso: Optional[float] = field(default=None) aperture: Optional[float] = field(default=None) @dataclass -class DispenseNozzlePosition: - position_x: Unit - position_y: Unit - position_z: Unit +class DispenseNozzlePosition(HashableMixin): + position_x: LENGTH + position_y: LENGTH + position_z: LENGTH @dataclass -class DispenseShape: +class DispenseShape(HashableMixin): rows: int columns: int format: str @dataclass -class DispenseShakeAfter: - duration: Optional[Union[Unit, str]] = field(default=None) - frequency: Optional[Union[Unit, str]] = field(default=None) +class DispenseShakeAfter(HashableMixin): + duration: Optional[TIME] = field(default=None) + frequency: Optional[FREQUENCY] = field(default=None) path: Optional[str] = field(default=None) - amplitude: Optional[Union[Unit, str]] = field(default=None) + amplitude: Optional[LENGTH] = field(default=None) class SonicateModeParamsBathSampleHolder(enum.Enum): @@ -369,13 +432,13 @@ class SonicateModeParamsBathSampleHolder(enum.Enum): @dataclass -class SonicateModeParamsBath: +class SonicateModeParamsBath(HashableMixin): sample_holder: SonicateModeParamsBathSampleHolder power: POWER @dataclass -class SonicateModeParamsHorn: +class SonicateModeParamsHorn(HashableMixin): duty_cycle: float power: LENGTH diff --git a/autoprotocol/unit.py b/autoprotocol/unit.py index 75a4465a..44b392a8 100644 --- a/autoprotocol/unit.py +++ b/autoprotocol/unit.py @@ -197,11 +197,18 @@ def __new__(cls, value, units=None): return value # Automatically parse String if no units provided - if not units and isinstance(value, str): - try: - value, units = value.split(":") - except ValueError as e: - raise UnitStringError(value) from e + if not units: + if isinstance(value, str): + try: + value, units = value.split(":") + except ValueError as e: + raise UnitStringError(value) from e + elif isinstance(value, dict): + try: + value, units = value["value"], value["units"] + except ValueError as e: + raise UnitUnitsError(value) from e + try: return super(Unit, cls).__new__(cls, Decimal(str(value)), units) except (ValueError, InvalidOperation) as e: @@ -211,8 +218,9 @@ def __new__(cls, value, units=None): def __post_init__(self): super(Unit, self).__init__() - self.units = self._units.__str__() + self.value = float(self.magnitude) self.unit = self._units.__str__() + self.units = self._units.__str__() def __str__(self, ndigits=12): """ diff --git a/autoprotocol/version.py b/autoprotocol/version.py index 41b62e8d..47e07f2e 100644 --- a/autoprotocol/version.py +++ b/autoprotocol/version.py @@ -1,2 +1,2 @@ """Maintains current version of package""" -__version__ = "10.1.1" +__version__ = "10.2.0" diff --git a/docs/changelog.rst b/docs/changelog.rst index 78499603..54bf8860 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -1,6 +1,9 @@ ========= Changelog ========= +* :release: `10.2.0 <2023-02-24>` +* :support:`385` Further updates for dataclass compatibility + * :release: `10.1.1 <2023-02-23>` * :feature: `378` Adjust `96-deep-kf` and `96-v-kf` container types `true_max_vol` to 1.8mL and added validations in `MagBuilders` class such that plates being put onto KF diff --git a/test/builder_test.py b/test/builder_test.py index 566e4e8b..fb08ca9f 100644 --- a/test/builder_test.py +++ b/test/builder_test.py @@ -13,6 +13,7 @@ Spectrophotometry, Thermocycle, ) +from autoprotocol.types.protocol import DispenseColumn from autoprotocol.unit import Unit, UnitError @@ -56,7 +57,7 @@ class TestDispenseBuilders(object): def test_column(self): for column in self.columns_reference: - assert column == Dispense.builders.column(**column) + assert DispenseColumn(**column) == Dispense.builders.column(**column) with pytest.raises(TypeError): Dispense.builders.column(0, 5) @@ -66,7 +67,7 @@ def test_column(self): def test_columns(self): cols = Dispense.builders.columns(self.columns_reference) - assert cols == self.columns_reference + assert cols == [DispenseColumn(**c) for c in self.columns_reference] with pytest.raises(TypeError): Dispense.builders.columns([{"column": 0}])