Source code for autoprotocol.protocol

"""
Module containing the main `Protocol` object and associated functions

    :copyright: 2021 by The Autoprotocol Development Team, see AUTHORS
        for more details.
    :license: BSD, see LICENSE for more details

"""
import json
import warnings

from collections import defaultdict
from dataclasses import 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, WellGroup
from .container_type import _CONTAINER_TYPES, ContainerType
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 import asdict
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,
    _validate_liha_shape,
    is_valid_well,
    parse_unit,
)


[docs]@dataclass class Protocol: refs: Optional[Dict[str, Ref]] = None instructions: List[Instruction] = field(default_factory=list) propagate_properties: bool = False time_constraints: List[TimeConstraint] = field(default_factory=list) """ A Protocol is a sequence of instructions to be executed, and a set of containers on which those instructions act. Parameters ---------- refs : list(Ref) Pre-existing refs that the protocol should be populated with. instructions : list(Instruction) Pre-existing instructions that the protocol should be populated with. propagate_properties : bool, optional Whether liquid handling operations should propagate aliquot properties from source to destination wells. time_constraints : List(time_constraints) Pre-existing time_constraints that the protocol should be populated with. Examples -------- Initially, a Protocol has an empty sequence of instructions and no referenced containers. To add a reference to a container, use the ref() method, which returns a Container. .. code-block:: python p = Protocol() my_plate = p.ref("my_plate", id="ct1xae8jabbe6", cont_type="96-pcr", storage="cold_4") To add instructions to the protocol, use the helper methods in this class .. code-block:: python p.transfer(source=my_plate.well("A1"), dest=my_plate.well("B4"), volume="50:microliter") p.thermocycle(my_plate, groups=[ { "cycles": 1, "steps": [ { "temperature": "95:celsius", "duration": "1:hour" }] }]) Autoprotocol Output: .. code-block:: json { "refs": { "my_plate": { "id": "ct1xae8jabbe6", "store": { "where": "cold_4" } } }, "instructions": [ { "groups": [ { "transfer": [ { "volume": "50.0:microliter", "to": "my_plate/15", "from": "my_plate/0" } ] } ], "op": "pipette" }, { "volume": "10:microliter", "dataref": null, "object": "my_plate", "groups": [ { "cycles": 1, "steps": [ { "duration": "1:hour", "temperature": "95:celsius" } ] } ], "op": "thermocycle" } ] } """ def __post_init__(self): if not self.refs: self.refs: Dict[str, Ref] = {} def __repr__(self): return f"Protocol({self.__dict__})"
[docs] def container_type(self, shortname: str): """ Convert a ContainerType shortname into a ContainerType object. Parameters ---------- shortname : str String representing one of the ContainerTypes in the _CONTAINER_TYPES dictionary. Returns ------- ContainerType Returns a Container type object corresponding to the shortname passed to the function. If a ContainerType object is passed, that same ContainerType is returned. Raises ------ ValueError If an unknown ContainerType shortname is passed as a parameter. """ if isinstance(shortname, ContainerType): return shortname elif shortname in _CONTAINER_TYPES: return _CONTAINER_TYPES[shortname] else: raise ValueError( f"Unknown container type {shortname}" f"(known types={str(_CONTAINER_TYPES.keys())})" )
# pragma pylint: disable=redefined-builtin
[docs] def ref( self, name: str, id: Optional[str] = None, cont_type: Optional[Union[str, ContainerType]] = None, storage: Optional[str] = None, discard: Optional[bool] = None, cover: Optional[str] = None, properties: Optional[Dict[str, str]] = None, ctx_properties: Optional[Dict[str, str]] = None, ): """ Add a Ref object to the dictionary of Refs associated with this protocol and return a Container with the id, container type and storage or discard conditions specified. Example Usage: .. code-block:: python p = Protocol() # ref a new container (no id specified) sample_ref_1 = p.ref("sample_plate_1", cont_type="96-pcr", discard=True) # ref an existing container with a known id sample_ref_2 = p.ref("sample_plate_2", id="ct1cxae33lkj", cont_type="96-pcr", storage="ambient") Autoprotocol Output: .. code-block:: json { "refs": { "sample_plate_1": { "new": "96-pcr", "discard": true }, "sample_plate_2": { "id": "ct1cxae33lkj", "store": { "where": "ambient" } } }, "instructions": [] } Parameters ---------- name : str name of the container/ref being created. id : str, optional id of the container being created, from your organization's inventory on http://secure.transcriptic.com. Strings representing ids begin with "ct". cont_type : str or ContainerType container type of the Container object that will be generated. storage : Enum({"ambient", "cold_20", "cold_4", "warm_37"}), optional temperature the container being referenced should be stored at after a run is completed. Either a storage condition must be specified or discard must be set to True. discard : bool, optional if no storage condition is specified and discard is set to True, the container being referenced will be discarded after a run. cover: str, optional name of the cover which will be on the container/ref properties : dict, optional mapping of key value properties associated to the Container ctx_properties : dict, optional mapping of key value properties associated to the Container Returns ------- Container Container object generated from the id and container type provided Raises ------ RuntimeError If a container previously referenced in this protocol (existent in refs section) has the same name as the one specified. RuntimeError If no container type is specified. RuntimeError If no valid storage or discard condition is specified. """ if name in self.refs.keys(): raise RuntimeError( "Two containers within the same protocol cannot have the same " "name." ) opts: RefOpts = RefOpts() # Check container type try: cont_type = self.container_type(cont_type) if id and cont_type: opts.id = id elif cont_type: opts.new = cont_type.shortname except ValueError as e: raise RuntimeError( f"{cont_type} is not a recognized container type." ) from e if storage: opts.store = StorageLocation(**{"where": storage}) elif discard and not storage: opts.discard = discard else: raise RuntimeError( "You must specify either a valid storage condition or set " "discard=True for a Ref." ) if cover: opts.cover = cover container = Container( id=id, container_type=cont_type, name=name, storage=storage if storage else None, cover=cover if cover else None, properties=properties, ctx_properties=ctx_properties, ) self.refs[name] = Ref(name, opts, container) return container
# pragma pylint: enable=redefined-builtin
[docs] def add_time_constraint( self, from_dict: TimeConstraintFromToDict, to_dict: TimeConstraintFromToDict, less_than: Optional[TIME] = None, more_than: Optional[TIME] = None, mirror: bool = False, ideal: Optional[TIME] = None, optimization_cost: Optional[str] = None, ): """Constraint the time between two instructions Add time constraints from `from_dict` to `to_dict`. Time constraints guarantee that the time from the `from_dict` to the `to_dict` is less than or greater than some specified duration. Care should be taken when applying time constraints as constraints may make some protocols impossible to schedule or run. Though autoprotocol orders instructions in a list, instructions do not need to be run in the order they are listed and instead depend on the preceding dependencies. Time constraints should be added with such limitations in mind. Constraints are directional; use `mirror=True` if the time constraint should be added in both directions. Note that mirroring is only applied to the less_than constraint, as the more_than constraint implies both a minimum delay betweeen two timing points and also an explicit ordering between the two timing points. Ideal time constraints are sometimes helpful for ensuring that a certain set of operations happen within some specified time. This can be specified by using the `ideal` parameter. There is an optional `optimization_cost` parameter associated with `ideal` time constraints for specifying the penalization system used for calculating deviations from the `ideal` time. When left unspecified, the `optimization_cost` function defaults to linear. Please refer to the ASC for more details on how this is implemented. Example Usage: .. code-block:: python plate_1 = protocol.ref("plate_1", id=None, cont_type="96-flat", discard=True) plate_2 = protocol.ref("plate_2", id=None, cont_type="96-flat", discard=True) protocol.cover(plate_1) time_point_1 = protocol.get_instruction_index() protocol.cover(plate_2) time_point_2 = protocol.get_instruction_index() protocol.add_time_constraint( {"mark": plate_1, "state": "start"}, {"mark": time_point_1, "state": "end"}, less_than = "1:minute") protocol.add_time_constraint( {"mark": time_point_2, "state": "start"}, {"mark": time_point_1, "state": "start"}, less_than = "1:minute", mirror=True) # Ideal time constraint protocol.add_time_constraint( {"mark": time_point_1, "state": "start"}, {"mark": time_point_2, "state": "end"}, ideal = "30:second", optimization_cost = "squared") Autoprotocol Output: .. code-block:: json { "refs": { "plate_1": { "new": "96-flat", "discard": true }, "plate_2": { "new": "96-flat", "discard": true } }, "time_constraints": [ { "to": { "instruction_end": 0 }, "less_than": "1.0:minute", "from": { "ref_start": "plate_1" } }, { "to": { "instruction_start": 0 }, "less_than": "1.0:minute", "from": { "instruction_start": 1 } }, { "to": { "instruction_start": 1 }, "less_than": "1.0:minute", "from": { "instruction_start": 0 } }, { "from": { "instruction_start": 0 }, "to": { "instruction_end": 1 }, "ideal": { "value": "5:minute", "optimization_cost": "squared" } } ], "instructions": [ { "lid": "standard", "object": "plate_1", "op": "cover" }, { "lid": "standard", "object": "plate_2", "op": "cover" } ] } Parameters ---------- from_dict: dict Dictionary defining the initial time constraint condition. Composed of keys: "mark" and "state" mark: int or Container instruction index of container state: "start" or "end" specifies either the start or end of the "mark" point to_dict: dict Dictionary defining the end time constraint condition. Specified in the same format as from_dict less_than: str or Unit, optional max time between from_dict and to_dict more_than: str or Unit, optional min time between from_dict and to_dict mirror: bool, optional choice to mirror the from and to positions when time constraints should be added in both directions (only applies to the less_than constraint) ideal: str or Unit, optional ideal time between from_dict and to_dict optimization_cost: Enum({"linear", "squared", "exponential"}), optional cost function used for calculating the penalty for missing the `ideal` timing Raises ------ ValueError If an instruction mark is less than 0 TypeError If mark is not container or integer TypeError If state not in ['start', 'end'] TypeError If any of `ideal`, `more_than`, `less_than` is not a Unit of the 'time' dimension KeyError If `to_dict` or `from_dict` does not contain 'mark' KeyError If `to_dict` or `from_dict` does not contain 'state' ValueError If time is less than '0:second' ValueError If `optimization_cost` is specified but `ideal` is not ValueError If `more_than` is greater than `less_than` ValueError If `ideal` is smaller than `more_than` or greater than `less_than` RuntimeError If `from_dict` and `to_dict` are equal RuntimeError If from_dict["marker"] and to_dict["marker"] are equal and from_dict["state"] = "end" """ inst_string = "instruction_" cont_string = "ref_" state_strings = ["start", "end"] keys = [] # Move the 4th param to mirror if the caller used the syntax # add_time_constraint(a, b, 1:minute, True) if type(more_than) == bool: mirror = more_than more_than = None # Validate input types def validate_timing(constraint): if constraint is not None: constraint = parse_unit(constraint, "minute") if constraint < Unit(0, "second"): raise ValueError( f"The timing constraint {constraint} cannot be " "less than '0:second'" ) return constraint more_than = validate_timing(more_than) less_than = validate_timing(less_than) ideal = validate_timing(ideal) if ideal and optimization_cost is None: optimization_cost = "linear" if optimization_cost is not None: if ideal is None: raise ValueError( "'optimization_cost' can only be specified if 'ideal'" "is also specified" ) ACCEPTED_COST_FUNCTIONS = ["linear", "squared", "exponential"] if optimization_cost not in ACCEPTED_COST_FUNCTIONS: raise ValueError( f"'optimization_cost': {optimization_cost} has to be a " f"member of {ACCEPTED_COST_FUNCTIONS}" ) if more_than and less_than and more_than > less_than: raise ValueError( f"'more_than': {more_than} cannot be greater than 'less_than': " f"{less_than}" ) if ideal and more_than and ideal < more_than: raise ValueError( f"'ideal': {ideal} cannot be smaller than 'more_than': " f"{more_than}" ) if ideal and less_than and ideal > less_than: raise ValueError( f"'ideal': {ideal} cannot be greater than 'less_than': " f"{less_than}" ) for m in [from_dict, to_dict]: if "mark" in m: if isinstance(m["mark"], Container): k = cont_string elif isinstance(m["mark"], int): k = inst_string if m["mark"] < 0: raise ValueError( f"The instruction 'mark' in {m} must be greater " f"than and equal to 0" ) else: raise TypeError(f"The 'mark' in {m} must be Container or Integer") else: raise KeyError(f"The {m} dict must contain `mark`") if "state" in m: if m["state"] in state_strings: k += m["state"] else: raise TypeError( f"The 'state' in {m} must be in " f"{', '.join(state_strings)}" ) else: raise KeyError(f"The {m} dict must contain 'state'") keys.append(k) if from_dict["mark"] == to_dict["mark"]: if from_dict["state"] == to_dict["state"]: raise RuntimeError( f"The from_dict: {from_dict} and to_dict: {to_dict} are " f"the same" ) if from_dict["state"] == "end": raise RuntimeError( f"The from_dict: {from_dict} cannot come before the " f"to_dict {to_dict}" ) from_time_point = {keys[0]: from_dict["mark"]} to_time_point = {keys[1]: to_dict["mark"]} if less_than is not None: self.time_constraints += [ {"from": from_time_point, "to": to_time_point, "less_than": less_than} ] if mirror: self.add_time_constraint(to_dict, from_dict, less_than, mirror=False) if more_than is not None: self.time_constraints += [ {"from": from_time_point, "to": to_time_point, "more_than": more_than} ] if ideal is not None: ideal_dict = dict(value=ideal) if optimization_cost is not None: ideal_dict["optimization_cost"] = optimization_cost self.time_constraints += [ {"from": from_time_point, "to": to_time_point, "ideal": ideal_dict} ]
[docs] def get_instruction_index(self): """Get index of the last appended instruction Example Usage: .. code-block:: python p = Protocol() plate_1 = p.ref("plate_1", id=None, cont_type="96-flat", discard=True) p.cover(plate_1) time_point_1 = p.get_instruction_index() # time_point_1 = 0 Raises ------ ValueError If an instruction index is less than 0 Returns ------- int Index of the preceding instruction """ instruction_index = len(self.instructions) - 1 if instruction_index < 0: raise ValueError("Instruction index less than 0") return instruction_index
def _append_and_return(self, instructions: Union[Instruction, List[Instruction]]): """ Append instruction(s) to the Protocol list and returns the Instruction(s). The other functions on Protocol() should be used in lieu of doing this directly. Example Usage: .. code-block:: python p = Protocol() p._append_and_return( Incubate("sample_plate", "ambient", "1:hour") ) Autoprotocol Output: .. code-block:: none "instructions": [ { "duration": "1:hour", "where": "ambient", "object": "sample_plate", "shaking": false, "op": "incubate" } ] Parameters ---------- instructions : Instruction or list(Instruction) Instruction object(s) to be appended. Returns ------- Instruction or list(Instruction) Instruction object(s) to be appended and returned """ if isinstance(instructions, list): self.instructions.extend(instructions) else: self.instructions.append(instructions) return instructions
[docs] def batch_containers( self, containers: List[Container], batch_in: bool = True, batch_out: bool = False, ): """ Batch containers such that they all enter or exit together. Example Usage: .. code-block:: python plate_1 = protocol.ref("p1", None, "96-pcr", storage="cold_4") plate_2 = protocol.ref("p2", None, "96-pcr", storage="cold_4") protocol.batch_containers([plate_1, plate_2]) Autoprotocol Output: .. code-block:: json { "refs": { "p1": { "new": "96-pcr", "store": { "where": "cold_4" } }, "p2": { "new": "96-pcr", "store": { "where": "cold_4" } } }, "time_constraints": [ { "from": { "ref_start": "p1" }, "less_than": "0:second", "to": { "ref_start": "p2" } }, { "from": { "ref_start": "p1" }, "more_than": "0:second", "to": { "ref_start": "p2" } } ] } Parameters ---------- containers : list(Container) Containers to batch batch_in : bool, optional Batch the entry of containers, default True batch_out: bool, optional Batch the exit of containers, default False Raises ------ TypeError If containers is not a list TypeError If containers is not a list of Container object """ time = Unit(0, "second") if not isinstance(containers, list): raise TypeError("batch_containers containers must be a list") if not all(isinstance(cont, Container) for cont in containers): raise TypeError("batch_containers containers must be a list of containers.") if not batch_in and not batch_out or len(containers) < 2: warnings.warn("batch_containers is used but has no effect") reference_container = containers[0] remainder_containers = containers[1:] states = [] if batch_in: states.append("start") if batch_out: states.append("end") for container in remainder_containers: for state in states: from_dict = {"mark": reference_container, "state": state} to_dict = {"mark": container, "state": state} self.add_time_constraint( from_dict=from_dict, to_dict=to_dict, less_than=time, more_than=time )
[docs] def as_dict(self): """ Return the entire protocol as a dictionary. Example Usage: .. code-block:: python from autoprotocol.protocol import Protocol import json p = Protocol() sample_ref_2 = p.ref("sample_plate_2", id="ct1cxae33lkj", cont_type="96-pcr", storage="ambient") p.seal(sample_ref_2) p.incubate(sample_ref_2, "warm_37", "20:minute") print json.dumps(p.as_dict(), indent=2) Autoprotocol Output: .. code-block:: json { "refs": { "sample_plate_2": { "id": "ct1cxae33lkj", "store": { "where": "ambient" } } }, "instructions": [ { "object": "sample_plate_2", "op": "seal" }, { "duration": "20:minute", "where": "warm_37", "object": "sample_plate_2", "shaking": false, "op": "incubate" } ] } Returns ------- dict dict with keys "refs" and "instructions" and optionally "time_constraints" and "outs", each of which contain the "refified" contents of their corresponding Protocol attribute. Raises ------ RuntimeError If either refs or instructions attribute is empty """ outs = defaultdict(lambda: defaultdict(dict)) # pragma pylint: disable=protected-access for n, ref in self.refs.items(): # assign any storage or discard condition changes to ref if ref.opts.store: ref.opts.store.where = ref.container.storage if not ref.container.storage and not ref.opts.discard: ref.opts.discard = True ref.opts.store = None elif ref.container.storage and ref.opts.discard: ref.opts.store = StorageLocation(**{"where": ref.container.storage}) ref.opts.discard = None if ref.container.properties: outs[n]["properties"] = ref.container.properties if ref.container.ctx_properties: outs[n]["contextual_custom_properties"] = ref.container.ctx_properties for well in ref.container._wells: if well.name or len(well.properties) > 0: if well.name: outs[n][str(well.index)]["name"] = well.name if len(well.properties) > 0: outs[n][str(well.index)]["properties"] = well.properties if well.ctx_properties: outs[n][str(well.index)][ "contextual_custom_properties" ] = well.ctx_properties # pragma pylint: enable=protected-access if outs: setattr(self, "outs", json.loads(json.dumps(outs))) prop_list = [ a for a in dir(self) if not a.startswith("__") and not callable(getattr(self, a)) ] explicit_props = ["outs", "refs", "instructions"] # attributes that are always serialized. optional_props = ["time_constraints"] # optional attributes that are serialized only when there are values for prop in optional_props: if getattr(self, prop): explicit_props.append(prop) return { attr: self._refify(getattr(self, attr)) for attr in prop_list if attr in explicit_props }
# pylint: disable=protected-access # pylint: disable=no-member
[docs] def liquid_handle_dispense( self, source: Union[Well, List[Well], List[Tuple[Well, int]]], destination: Union[WellParam, List[WellGroup]], volume: Union[VOLUME, List[VOLUME]], rows: int = 8, columns: int = 1, method: DispenseMethod = DispenseMethod, liquid: LiquidClass = LiquidClass, device: Optional[str] = "x_tempest_chip", model: Optional[str] = None, chip_material: Optional[str] = None, nozzle: Optional[str] = None, diaphragm: Optional[int] = None, nozzle_size: Optional[Union[Unit, str]] = None, tubing: Optional[str] = None, z_drop: Optional[Union[Unit, str]] = None, viscosity: Optional[str] = None, ): """Generates a liquid_handle dispense Dispenses liquid from a source well to a group of destination wells. Notes ----- Some liquidClass parameters including volume calibration and flowrate are not yet supported. There are currently no plans to support liquid level detection and its corresponding thresholds for this mode. Parameters ---------- source : Well or list(Well) or list(tuple(Well, int)) Well(s) to transfer liquid from. If the source is given as a list of (Well, int) tuples (ie: source=[(Well, 2), (Well, 5)]) the Well is the source well and the int is the number of dispense chips to use from the specified source. destination : Well or WellGroup or list(Well) or list(WellGroup) Well(s) to transfer liquid to. If specifying more than a Well or WellGroup the list of destinations must match the number of chips specified in the source tuple. volume : str or Unit or list(str) or list(Unit) Volume(s) of liquid to be transferred from source well to destination wells. The number of volumes specified must correspond to the number of destination wells. If specifying more than one volume to be used for all dispenses, the list of volumes must match the length of destinations. rows : int, optional Number of rows to be concurrently transferred columns : int, optional Number of columns to be concurrently transferred liquid : LiquidClass or list(LiquidClass), optional Type(s) of liquid to be dispensed. This affects aspirate and dispense behavior including flowrates and physical movements. If the number of Dispense classes specified is more than one then number specified must match the length of sources. method : Dispense or list(Dispense), optional Integrates with the specified liquid to define a set of physical movements. If the number of Dispense classes specified is more than one then number specified must match the length of sources. device: str Device used for liquid_handle_dispense execution model : str, optional Tempest chip model, currently only support "high_volume". chip_material : str, optional Tempest chip material, support "silicone" and "pfe", default is "silicon". nozzle : str, optional Tempest nozzle type, currently only support "standard". The three chip parameters: model, chip_material, and nozzle will be used in liquid handle mode_params to allow tempest chip specification. diaphragm : Optional[int] any integer between 0 and 100, inclusive nozzle_size : Unit[Length], optional one of "0.1:mm", "0.2:mm", "0.5:mm". Default is "0.1:mm". tubing: Optional[String] one of "LV", "HV", "P200", "P1000". Default is "LV" z_drop: Optional[Unit[Length]] within the range: 0:mm - 100:mm, inclusive. Default is "0:mm" viscosity: Optional[String] one of "1", "2-5", "6-10", "11-20", "21-25". Default is "1". Returns ------- LiquidHandle Returns a :py:class:`autoprotocol.instruction.LiquidHandle` instruction created from the specified parameters. Raises ------ ValueError if source is not a Well or list((Well, int)) ValueError if destination and volumes are not of the same length ValueError if model is not high_volume ValueError if nozzle is not standard ValueError if chip_material is not in silicone or pfe TypeError if volume is not one of the defined allowable types Examples -------- A single volume dispense to a single column .. code-block:: python from autoprotocol import Protocol p = Protocol() source = p.ref("source", cont_type="conical-50", discard=True) destination = p.ref("destination", cont_type="96-flat", discard=True) p.liquid_handle_dispense( source=source.well(0), destination=destination.well(0), volume="5:ul" ) A single volume dispense to a whole 384 well plate .. code-block:: python from autoprotocol import Protocol p = Protocol() source = p.ref("source", cont_type="conical-50", discard=True) destination = p.ref("destination", cont_type="384-flat", discard=True) p.liquid_handle_dispense( source=source.well(0), destination=destination.wells_from(0, 48), volume="5:ul" ) Dispensing a volume gradient across a plate .. code-block:: python from autoprotocol import Unit from autoprotocol import Protocol p = Protocol() source = p.ref("source", cont_type="conical-50", discard=True) destination = p.ref("destination", cont_type="96-flat", discard=True) p.liquid_handle_dispense( source=source.well(0), destination=destination.wells_from(0, 12), volume=[Unit(_, "uL") for _ in range(1, 13)] ) Using multiple chips in a single source tube. The following example will dispense 100:microliters to the first 3 columns of the destination plate. .. code-block:: python from autoprotocol import Unit from autoprotocol import Protocol p = Protocol() source = p.ref("source", cont_type="micro-2.0", discard=True).well(0) destination = p.ref("destination", cont_type="96-flat", discard=True) intake_hoses = 3 number_dispense_columns = 5 source: List[Tuple[Well, int]] = [(source, intake_hoses)] destination = [destination.wells_from(0, num_dispense_columns)] self.protocol.liquid_handle_dispense( source=source, destination=destination, volume='100:microliter', ) See Also -------- :py:class:`autoprotocol.liquid_handle.Dispense` Base liquid handling method for dispense operations """ default_num_dispense_chips_in_source: int = 1 default_max_num_dispense_chips: int = 12 remaining_num_chips_to_specify = default_max_num_dispense_chips _validate_liha_shape(device, {"rows": rows, "columns": columns}) def format_source_well(well: Well, num_chips: int) -> Tuple[Well, int]: return (well, num_chips) # Format parameters into lists that map params by index if isinstance(source, list): allowable_source_types = {Well, tuple} specified_source_types = {type(i) for i in source} unallowed_source_types = specified_source_types.difference( allowable_source_types ) if unallowed_source_types: raise ValueError( f"The specified list of sources contains invalid source " f"types {unallowed_source_types}" ) for i, s in enumerate(source): if isinstance(s, Well): source[i]: Tuple[Well, int] = format_source_well( s, default_num_dispense_chips_in_source ) remaining_num_chips_to_specify -= ( default_num_dispense_chips_in_source ) elif isinstance(s, tuple): sw, chip_num = s if not isinstance(sw, Well) or not isinstance(chip_num, int): raise ValueError( f"The specified value {s} in source[{i}] is not a (Well, int)" ) if chip_num > default_max_num_dispense_chips: raise ValueError( f"The number of chips specified {chip_num}" f" in source[{i}] is greater than the " f"max number of chips available {default_max_num_dispense_chips}" ) remaining_num_chips_to_specify -= chip_num else: raise ValueError( f"The specified value {s} in source[{i}] is not a (Well, int) or Well" ) elif isinstance(source, Well): source: List[Tuple[Well, int]] = [ format_source_well(source, default_num_dispense_chips_in_source) ] remaining_num_chips_to_specify -= default_num_dispense_chips_in_source else: raise ValueError( f"Source must be either " f"Well or list(Well) or list((Well, int)) but received {source}" ) if remaining_num_chips_to_specify < 0: raise ValueError( f"The number of chips configured for the " f"instruction {default_max_num_dispense_chips - remaining_num_chips_to_specify}" f" is greater than the number of chips available {default_max_num_dispense_chips}" ) # Check destinations match the number of sources len_src = len(source) def configure_dest(destinations) -> List[WellGroup]: final_destinations = [] if isinstance(destinations, Well): final_destinations.append(convert_to_wellgroup(destinations)) elif isinstance(destinations, list): for dest in destinations: wg = convert_to_wellgroup(dest) final_destinations.append(wg) if len_src == 1: final_destinations = [convert_to_wellgroup(final_destinations)] src_list = [] for _ in range(len(final_destinations)): src_list.append(source[0]) elif isinstance(destinations, WellGroup): final_destinations.append(destinations) else: raise TypeError("Destinations must be of type well, wellgroup, list.") if len(final_destinations) == 1: final_destinations = [final_destinations[0] for _ in range(len_src)] elif len_src == 1: src_list = [] for _ in range(len(final_destinations)): src_list.append(source[0]) elif len(final_destinations) != len_src: raise ValueError( f"If multiple destinations are specified, destinations({len(final_destinations)}) and" f" sources({len_src}) must be of equal length." ) if not all([isinstance(wg, WellGroup) for wg in final_destinations]): raise ValueError( f"not everything in the dest list of wellgroups is a wellgroup... {final_destinations}" ) return final_destinations def convert_to_wellgroup(dest) -> WellGroup: if isinstance(dest, Well): ret = WellGroup([dest]) elif isinstance(dest, list): ret = WellGroup([]) for item in dest: if isinstance(item, Well): ret.append(item) elif isinstance(item, WellGroup): ret.extend(item) else: raise TypeError( f"Destination {item}({type(item)}) is not a well or wellgroup." ) elif not isinstance(dest, WellGroup): raise TypeError( f"Destination {dest}({type(dest)}) is not a well or wellgroup." ) else: ret = dest return ret destination = configure_dest(destinations=destination) def equalize_lengths( vols, dest: Union[List[WellGroup], WellGroup] ) -> Union[List[Unit], List[List[Unit]]]: if isinstance(dest, list): if isinstance(vols, list): vols_list = vols if len(vols_list) == len(dest): for i in range(len(vols_list)): vols_list[i] = equalize_lengths(vols_list[i], dest[i]) return vols_list elif len(vols_list) == 1: vols_list = [vols_list for _ in dest] else: raise ValueError( f"If the volumes argument is a list(yours" f" was length {len(vols_list)}), it must " f"be length 1 or the same length as the" f" length of sources(length {len_src})." ) elif isinstance(vols, (Unit, str)): vols_list = [parse_unit(vols, "uL") for _ in dest] else: raise ValueError( f"Acceptable volumes argument types are:" f"Unit, List[Unit], or List[List[Unit]]." f"{vols}(type {type(vols)}) is not one " f"of those types." ) elif isinstance(dest, WellGroup): if isinstance(vols, (Unit, str)): return [parse_unit(vols) for _ in dest] elif isinstance(vols, list): if len(vols) == len(dest): return vols elif len(vols) == 1 and isinstance(vols[0], Unit): return [vols[0] for _ in dest] else: raise ValueError( f"Your list of volumes lists is" f" improperly nested({vols[0]} should" f" be a Unit) or one of the lists" f"(length {vols}) is not of the same" f" length as its corresponding source" f"(length {len(dest)})." ) # vols list will always be defined by now, or the function will have # raised an error because destinations will always be a list of wellgroups return equalize_lengths(vols_list, dest) volume: List[List[Unit]] = equalize_lengths(vols=volume, dest=destination) # Format LiquidClasses if not isinstance(liquid, list): lc = _validate_as_instance(liquid, LiquidClass) liquid: List[LiquidClass] = [lc] * len_src else: len_liquid = len(liquid) if len_liquid != len_src: raise ValueError( f"The number of LiquidClass(s) {len_liquid} and sources {len_src} do not match " ) elif not all(_validate_as_instance(lc, LiquidClass) for lc in liquid): raise ValueError( f"liquid list {liquid} did not contain all of type LiquidClass" ) # Format DispenseMethods if not isinstance(method, list): dm = _validate_as_instance(method, DispenseMethod) method: List[DispenseMethod] = [dm] * len_src else: len_method = len(method) if len_method != len_src: raise ValueError( f"The number of DispenseMethod(s) {len_method} and sources {len_src} do not match " ) elif not all(_validate_as_instance(dm, DispenseMethod) for dm in method): raise ValueError( f"method list {method} did not contain all of type DispenseMethod" ) # Set the DispenseMethod's liquid class for dm, lc in zip(method, liquid): # The index of the method and liquid classes should line up with # source, destination, and volumes you wish to effect. The name of # the liquid class will be applied to the transports of the mappings. if not dm._liquid: # Do not set a liquid class if the dispense method has one dm._liquid = lc transport_locations = [] shape = LiquidHandle.builders.shape(rows, columns, None) for i, (source_location, num_dispense_chips) in enumerate(source): dispense_volumes: List[Unit] = volume[i] destination_wg: WellGroup = destination[i] try: sum_dispense_volumes: Unit = ( sum(dispense_volumes) if dispense_volumes else Unit("0:microliter") ) except TypeError as e: raise TypeError(dispense_volumes) from e total_volume_dispensed: Unit = Unit("0:microliter") # Aspirate from source transport_locations.append( LiquidHandle.builders.location( location=source_location, transports=( method[i]._aspirate_transports(sum_dispense_volumes) * num_dispense_chips ), ) ) # there should be the same number of volumes as destinations total_volume_dispensed += sum( [rows * columns * vol for vol in dispense_volumes] ) # Prime from source if method[i].prime: transport_locations.append( LiquidHandle.builders.location( location=source_location, transports=(method[i]._prime_transports() * num_dispense_chips), ) ) # Predispense from source if method[i].predispense: transport_locations.append( LiquidHandle.builders.location( location=None, transports=( method[i]._predispense_transports() * num_dispense_chips ), ) ) total_volume_dispensed += ( method[i].get_predispense_volume() * rows * columns * num_dispense_chips ) # Dispense from source to destination # for each src_tuple-dest_wg-vol_list triplicate # for each destination well in the dest_wg # there will always be the same number of wells # in the well group as there are volumes in the list for j in range(len(destination_wg)): vol = dispense_volumes[j] destination_well = destination_wg[j] transport_locations.append( LiquidHandle.builders.location( location=destination_well, transports=(method[i]._dispense_transports(vol)), ) ) # Update destination column volumes with consideration of the num chips specified for w in destination_well.container.wells_from_shape( destination_well.index, shape ): w.add_volume(vol) # Update source volume source_location.add_volume(-total_volume_dispensed) device_mode_params = LiquidHandleBuilders.device_mode_params( device=device, model=model, chip_material=chip_material, nozzle=nozzle, diaphragm=diaphragm, nozzle_size=nozzle_size, tubing=tubing, z_drop=z_drop, viscosity=viscosity, ) return self._append_and_return( LiquidHandle( locations=transport_locations, shape=shape, mode="dispense", mode_params=device_mode_params, ) )
[docs] def store(self, container: Container, condition: str): """ Manually adjust the storage destiny for a container used within this protocol. Parameters ---------- container : Container Container used within this protocol condition : str New storage destiny for the specified Container Raises ------ TypeError If container argument is not a Container object RuntimeError If the container passed is not already present in self.refs """ if not condition or condition == "discard": condition = None if not isinstance(container, Container): raise TypeError("Protocol.store() can only be used on a Container object.") container.storage = condition r = self.refs.get(container.name) if not r: raise RuntimeError( "That container does not exist in the refs for this protocol." ) if "discard" in r.opts: r.opts.pop("discard") if condition: r.opts["store"] = {"where": str(condition)} else: r.opts.pop("store") r.opts["discard"] = True
[docs] def acoustic_transfer( self, source: WellParam, dest: WellParam, volume: VOLUME, one_source: bool = False, droplet_size: VOLUME = "25:nanoliter", ): """ Specify source and destination wells for transferring liquid via an acoustic liquid handler. Droplet size is usually device-specific. Example Usage: .. code-block:: python p.acoustic_transfer( echo.wells(0,1).set_volume("12:nanoliter"), plate.wells_from(0,5), "4:nanoliter", one_source=True) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ { "transfer": [ { "volume": "0.004:microliter", "to": "plate/0", "from": "echo_plate/0" }, { "volume": "0.004:microliter", "to": "plate/1", "from": "echo_plate/0" }, { "volume": "0.004:microliter", "to": "plate/2", "from": "echo_plate/0" }, { "volume": "0.004:microliter", "to": "plate/3", "from": "echo_plate/1" }, { "volume": "0.004:microliter", "to": "plate/4", "from": "echo_plate/1" } ] } ], "droplet_size": "25:microliter", "op": "acoustic_transfer" } ] Parameters ---------- source : Well or WellGroup or list(Well) Well or wells to transfer liquid from. If multiple source wells are supplied and one_source is set to True, liquid will be transferred from each source well specified as long as it contains sufficient volume. Otherwise, the number of source wells specified must match the number of destination wells specified and liquid will be transferred from each source well to its corresponding destination well. dest : Well or WellGroup or list(Well) Well or WellGroup to which to transfer liquid. The number of destination wells must match the number of source wells specified unless one_source is set to True. volume : str or Unit or list The volume(s) of liquid to be transferred from source wells to destination wells. Volume can be specified as a single string or Unit, or can be given as a list of volumes. The length of a list of volumes must match the number of destination wells given unless the same volume is to be transferred to each destination well. one_source : bool, optional Specify whether liquid is to be transferred to destination wells from a group of wells all containing the same substance. droplet_size : str or Unit, optional Volume representing a droplet_size. The volume of each `transfer` group should be a multiple of this volume. Returns ------- AcousticTransfer Returns the :py:class:`autoprotocol.instruction.AcousticTransfer` instruction created from the specified parameters Raises ------ TypeError Incorrect input types, e.g. source/dest are not Well or WellGroup or list of Well; or container_type does not have 'acoustic_transfer' capability. RuntimeError Incorrect length for source and destination RuntimeError Transfer volume not being a multiple of droplet size RuntimeError Insufficient volume in source wells """ # Check valid well inputs if not is_valid_well(source): raise TypeError("Source must be of type Well, list of Wells, or WellGroup.") if not is_valid_well(dest): raise TypeError( "Destination (dest) must be of type Well, list of Wells, or " "WellGroup." ) # Check valid container_type capability wells = [] if isinstance(source, Well): wells.append(source) elif isinstance(source, WellGroup): wells = source.wells elif isinstance(source, list): wells = source for well in wells: if "acoustic_transfer" not in well.container.container_type.capabilities: raise TypeError( f"AcousticTransfer: Source does not have 'acoustic_transfer' " f"capability for well '{well.index}' of container name " f"'{well.container.name}' with container ID '{well.container.id}' " f"and container type '{well.container.container_type.shortname}'!" ) transfers = [] source = WellGroup(source) dest = WellGroup(dest) len_source = len(source.wells) len_dest = len(dest.wells) droplet_size = Unit(droplet_size) max_decimal_places = 12 # for rounding after floating point arithmetic # Auto-generate well-group if only 1 well specified and using >1 source if not one_source: if len_dest > 1 and len_source == 1: source = WellGroup(source.wells * len_dest) len_source = len(source.wells) if len_dest == 1 and len_source > 1: dest = WellGroup(dest.wells * len_source) len_dest = len(dest.wells) if len_source != len_dest: raise RuntimeError( "To transfer liquid from one well or multiple wells " "containing the same source, set one_source to True. To " "transfer liquid from multiple wells to a single " "destination well, specify only one destination well. " "Otherwise, you must specify the same number of source and " "destination wells to do a one-to-one transfer." ) # Auto-generate list from single volume, check if list length matches if isinstance(volume, str) or isinstance(volume, Unit): if len_dest == 1 and not one_source: volume = [Unit(volume).to("ul")] * len_source else: volume = [Unit(volume).to("ul")] * len_dest elif isinstance(volume, list) and len(volume) == len_dest: volume = list(map(lambda x: Unit(x).to("ul"), volume)) else: raise RuntimeError( "Unless the same volume of liquid is being transferred to each " "destination well, each destination well must have a " "corresponding volume in the form of a list." ) vol_errors = [] for vol_d in volume: if not round(vol_d / droplet_size, max_decimal_places) % 1 == 0: vol_errors.append(vol_d) if len(vol_errors) > 0: raise RuntimeError( f"Transfer volume has to be a multiple of the droplet size ({droplet_size}).This is not true for the following volumes: {vol_errors}" ) # Ensure enough volume in single well to transfer to all dest wells if one_source: try: source_vol = [s.available_volume() for s in source.wells] if sum([a for a in volume]) > sum([a for a in source_vol]): raise RuntimeError( "There is not enough volume in the source well(s) " "specified to complete the transfers." ) if len_source >= len_dest and all( i > j for i, j in zip(source_vol, volume) ): sources = source.wells[:len_dest] destinations = dest.wells volumes = volume else: sources = [] source_counter = 0 destinations = [] volumes = [] s = source.wells[source_counter] vol = s.available_volume() for idx, d in enumerate(dest.wells): vol_d = volume[idx] while vol_d > Unit("0:microliter"): if vol > vol_d: sources.append(s) destinations.append(d) volumes.append(vol_d) vol -= vol_d vol = round(vol, max_decimal_places) vol_d -= vol_d vol_d = round(vol_d, max_decimal_places) else: sources.append(s) destinations.append(d) vol = int(vol / droplet_size) * droplet_size volumes.append(vol) vol_d -= vol vol_d = round(vol_d, max_decimal_places) source_counter += 1 if source_counter < len_source: s = source.wells[source_counter] vol = s.available_volume() source = WellGroup(sources) dest = WellGroup(destinations) volume = volumes except (ValueError, AttributeError, TypeError) as e: raise RuntimeError( "When transferring liquid from multiple wells containing " "the same substance to multiple other wells, each source " "Well must have a volume attribute (aliquot) associated " "with it." ) from e for s, d, v in list(zip(source.wells, dest.wells, volume)): self._remove_cover(s.container, "acoustic_transfer") self._remove_cover(d.container, "acoustic_transfer") xfer = {"from": s, "to": d, "volume": v} # Volume accounting if d.volume: d.volume += v else: d.volume = v if s.volume: s.volume -= v if v > Unit(0, "microliter"): transfers.append(xfer) if self.propagate_properties: d.add_properties(s.properties) if not transfers: raise RuntimeError( "At least one transfer must have a nonzero transfer volume." ) for x in transfers: x["volume"] = round(x["volume"].to("nl"), max_decimal_places) return self._append_and_return( AcousticTransfer([{"transfer": transfers}], droplet_size) )
[docs] def illuminaseq( self, flowcell: str, lanes: List[IlluminaSeqLane], sequencer: str, mode: str, index: str, library_size: int, dataref: str, cycles: Optional[str] = None, ): """ Load aliquots into specified lanes for Illumina sequencing. The specified aliquots should already contain the appropriate mix for sequencing and require a library concentration reported in ng/uL. Example Usage: .. code-block:: python p = Protocol() sample_wells = p.ref( "test_plate", None, "96-pcr", discard=True).wells_from(0, 8) p.illuminaseq( "PE", [ {"object": sample_wells[0], "library_concentration": 1.0}, {"object": sample_wells[1], "library_concentration": 5.32}, {"object": sample_wells[2], "library_concentration": 54}, {"object": sample_wells[3], "library_concentration": 20}, {"object": sample_wells[4], "library_concentration": 23}, {"object": sample_wells[5], "library_concentration": 23}, {"object": sample_wells[6], "library_concentration": 21}, {"object": sample_wells[7], "library_concentration": 62} ], "hiseq", "rapid", 'none', 250, "my_illumina") Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "my_illumina", "index": "none", "lanes": [ { "object": "test_plate/0", "library_concentration": 1 }, { "object": "test_plate/1", "library_concentration": 5.32 }, { "object": "test_plate/2", "library_concentration": 54 }, { "object": "test_plate/3", "library_concentration": 20 }, { "object": "test_plate/4", "library_concentration": 23 }, { "object": "test_plate/5", "library_concentration": 23 }, { "object": "test_plate/6", "library_concentration": 21 }, { "object": "test_plate/7", "library_concentration": 62 } ], "flowcell": "PE", "mode": "mid", "sequencer": "hiseq", "library_size": 250, "op": "illumina_sequence" } ] Parameters ---------- flowcell : str Flowcell designation: "SR" or " "PE" lanes : list(dict) .. code-block:: none "lanes": [ { "object": aliquot, Well, "library_concentration": decimal, // ng/uL }, {...}] sequencer : str Sequencer designation: "miseq", "hiseq" or "nextseq" mode : str Mode designation: "rapid", "mid" or "high" index : str Index designation: "single", "dual" or "none" library_size: int Library size expressed as an integer of basepairs dataref : str Name of sequencing dataset that will be returned. cycles : Enum({"read_1", "read_2", "index_1", "index_2"}) Parameter specific to Illuminaseq read-length or number of sequenced bases. Refer to the ASC for more details Returns ------- IlluminaSeq Returns the :py:class:`autoprotocol.instruction.IlluminaSeq` instruction created from the specified parameters Raises ------ TypeError If index and dataref are not of type str. TypeError If library_concentration is not a number. TypeError If library_size is not an integer. ValueError If flowcell, sequencer, mode, index are not of type a valid option. ValueError If number of lanes specified is more than the maximum lanes of the specified type of sequencer. KeyError Invalid keys specified for cycles parameter """ valid_flowcells = ["PE", "SR"] # currently available sequencers, modes and max number of lanes and # cycles valid_sequencers = { "miseq": {"max_lanes": 1, "modes": ["high"], "max_cycles_read": 600}, "hiseq": { "max_lanes": 8, "modes": ["high", "rapid"], "max_cycles_read": 500, }, "nextseq": { "max_lanes": 4, "modes": ["high", "mid"], "max_cycles_read": 300, }, } valid_indices = ["single", "dual", "none"] valid_cycles = ["index_1", "index_2", "read_1", "read_2"] max_cycles_ind = 12 if flowcell not in valid_flowcells: raise ValueError( f"Illumina sequencing flowcell type must be one " f"of: {', '.join(valid_flowcells)}." ) if sequencer not in valid_sequencers.keys(): raise ValueError( f"Illumina sequencer must be one of: " f"{', '.join(valid_sequencers.keys())}." ) if not isinstance(lanes, list): raise TypeError("Illumina sequencing lanes must be a list(dict)") for l in lanes: if not isinstance(l, dict): raise TypeError("Illumina sequencing lanes must be a list(dict)") if not all(k in l.keys() for k in ["object", "library_concentration"]): raise TypeError( "Each Illumina sequencing lane must contain an " "'object' and a 'library_concentration'" ) if not isinstance(l["object"], Well): raise TypeError( "Each Illumina sequencing object must be of " "type Well" ) if not isinstance(l["library_concentration"], (float, int)): raise TypeError( "Each Illumina sequencing " "library_concentration must be a number." ) if len(lanes) > valid_sequencers[sequencer]["max_lanes"]: raise ValueError( f"The type of sequencer selected ({sequencer}) only has {valid_sequencers[sequencer]['max_lanes']} lane(s). You specified {len(lanes)}. Please submit additional Illumina Sequencing instructions." ) if mode not in valid_sequencers[sequencer]["modes"]: raise ValueError( f"The type of sequencer selected ({sequencer}) has valid modes: {', '.join(valid_sequencers[sequencer]['modes'])}.You specified: {mode}." ) if index not in valid_indices: raise ValueError( f"Illumina sequencing index must be one of: " f"{', '.join(valid_indices)}." ) if not isinstance(dataref, str): raise TypeError(f"dataref: {dataref}, must be a string") if not isinstance(library_size, int): raise TypeError(f"library_size: {library_size}, must be an integer.") if cycles: if not isinstance(cycles, dict): raise TypeError("Cycles must be a dict.") if not all(c in valid_cycles for c in cycles.keys()): raise KeyError( f"Valid cycle parameters are: " f"{', '.join(valid_cycles)}" ) if "read_1" not in cycles.keys(): raise ValueError("If specifying cycles, 'read_1' must be designated.") if flowcell == "SR" and "read_2" in cycles.keys(): raise ValueError("SR does not have a second read: 'read_2'.") if not all(isinstance(i, int) for i in cycles.values()): raise ValueError("Cycles must be specified as an integer.") for read in ["read_1", "read_2"]: if cycles.get(read): if cycles[read] > valid_sequencers[sequencer]["max_cycles_read"]: raise ValueError( f"The maximum number of cycles for {read} is {valid_sequencers[sequencer]['max_cycles_read']}." ) for ind in ["index_1", "index_2"]: if cycles.get(ind): if cycles[ind] > max_cycles_ind: raise ValueError( f"The maximum number of cycles for {ind} is " f"{max_cycles_ind}." ) # set index 1 and 2 to default 0 if not otherwise specified else: cycles[ind] = 0 return self._append_and_return( IlluminaSeq( flowcell, lanes, sequencer, mode, index, library_size, dataref, cycles ) )
# pylint: disable=redefined-builtin
[docs] def sangerseq( self, cont: Union[Container, str], wells: WellParam, dataref: str, type: str = "standard", primer: Optional[Container] = None, ): """ Send the indicated wells of the container specified for Sanger sequencing. The specified wells should already contain the appropriate mix for sequencing, including primers and DNA according to the instructions provided by the vendor. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.sangerseq(sample_plate, sample_plate.wells_from(0,5).indices(), "seq_data_022415") Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "seq_data_022415", "object": "sample_plate", "wells": [ "A1", "A2", "A3", "A4", "A5" ], "op": "sanger_sequence" } ] Parameters ---------- cont : Container or str Container with well(s) that contain material to be sequenced. wells : list(Well) or WellGroup or Well WellGroup of wells to be measured or a list of well references in the form of ["A1", "B1", "C5", ...] dataref : str Name of sequencing dataset that will be returned. type: Enum({"standard", "rca"}) Sanger sequencing type primer : Container, optional Tube containing sufficient primer for all RCA reactions. This field will be ignored if you specify the sequencing type as "standard". Tube containing sufficient primer for all RCA reactions Returns ------- SangerSeq Returns the :py:class:`autoprotocol.instruction.SangerSeq` instruction created from the specified parameters Raises ------ RuntimeError No primer location specified for rca sequencing type ValueError Wells belong to more than one container TypeError Invalid input type for wells """ seq_type = type.lower() if seq_type == "rca" and not primer: raise RuntimeError( "You must specify the location of primer for " "RCA sequencing reactions." ) if isinstance(wells, Well): wells = WellGroup(wells) if isinstance(wells, WellGroup): container = set([w.container for w in wells]) if len(container) > 1: raise ValueError("All wells need to be on one container for SangerSeq") wells = [str(w.index) for w in wells] if not isinstance(wells, list): raise TypeError( "Unknown input. SangerSeq wells accepts either a" "Well, a WellGroup, or a list of well indices" ) return self._append_and_return(SangerSeq(cont, wells, dataref, type, primer))
[docs] def dispense( self, ref: Container, reagent: Union[str, Well], columns: List[DispenseColumn], is_resource_id: bool = False, step_size: VOLUME = "5:uL", flowrate: Optional[FLOW_RATE] = None, nozzle_position: Optional[DispenseNozzlePosition] = None, pre_dispense: Optional[VOLUME] = None, shape: Optional[DispenseShape] = None, shake_after: Optional[DispenseShakeAfter] = None, ): """ Dispense specified reagent to specified columns. Example Usage: .. code-block:: python from autoprotocol.liquid_handle.liquid_handle_builders import * from autoprotocol.instructions import Dispense from autoprotocol import Protocol p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.dispense(sample_plate, "water", Dispense.builders.columns( [Dispense.builders.column(0, "10:uL"), Dispense.builders.column(1, "20:uL"), Dispense.builders.column(2, "30:uL"), Dispense.builders.column(3, "40:uL"), Dispense.builders.column(4, "50:uL") ]) ) p.dispense( sample_plate, "water", Dispense.builders.columns( [Dispense.builders.column(0, "10:uL")] ), Dispense.builders.nozzle_position( position_x=Unit("1:mm"), position_y=Unit("2:mm"), position_z=Unit("20:mm") ), shape_builder( rows=8, columns=1, format="SBS96" ) ) Autoprotocol Output: .. code-block:: none "instructions": [ { "reagent": "water", "object": "sample_plate", "columns": [ { "column": 0, "volume": "10:microliter" }, { "column": 1, "volume": "20:microliter" }, { "column": 2, "volume": "30:microliter" }, { "column": 3, "volume": "40:microliter" }, { "column": 4, "volume": "50:microliter" } ], "op": "dispense" }, { "reagent": "water", "object": "sample_plate", "columns": [ { "column": 0, "volume": "10:microliter" } ], "nozzle_position" : { "position_x" : "1:millimeter", "position_y" : "2:millimeter", "position_z" : "20:millimeter" }, "shape" : { "rows" : 8, "columns" : 1, "format" : "SBS96" } "op": "dispense" }, ] Parameters ---------- ref : Container Container for reagent to be dispensed to. reagent : str or well Reagent to be dispensed. Use a string to specify the name or resource_id (see below) of the reagent to be dispensed. Alternatively, use a well to specify that the dispense operation must be executed using a specific aliquot as the dispense source. columns : list(dict("column": int, "volume": str/Unit)) Columns to be dispensed to, in the form of a list(dict) specifying the column number and the volume to be dispensed to that column. Columns are expressed as integers indexed from 0. [{"column": <column num>, "volume": <volume>}, ...] is_resource_id : bool, optional If true, interprets reagent as a resource ID step_size : str or Unit, optional Specifies that the dispense operation must be executed using a peristaltic pump with the given step size. Note that the volume dispensed in each column must be an integer multiple of the step_size. Currently, step_size must be one of 5 uL, 0.5 uL, 0.05 uL. If set to None, will use vendor specified defaults. flowrate : str or Unit, optional The rate at which liquid is dispensed into the ref in units of volume/time. nozzle_position : dict, optional A dict represent nozzle offsets from the bottom middle of the plate's wells. see Dispense.builders.nozzle_position; specified as {"position_x": Unit, "position_y": Unit, "position_z": Unit}. pre_dispense : str or Unit, optional The volume of reagent to be dispensed per-nozzle into waste immediately prior to dispensing into the ref. shape: dict, optional The shape of the dispensing head to be used for the dispense. See liquid_handle_builders.shape_builder; specified as {"rows": int, "columns": int, "format": str} with format being a valid SBS format. shake_after: dict, optional Parameters that specify how a plate should be shaken at the very end of the instruction execution. See Dispense.builders.shake_after. Returns ------- Dispense Returns the :py:class:`autoprotocol.instruction.Dispense` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. ref is not of type Container ValueError Columns specified is invalid for this container type ValueError Invalid step-size given ValueError Invalid pre-dispense volume """ _VALID_STEP_SIZES = [Unit(5, "uL"), Unit(0.5, "uL"), Unit(0.05, "uL")] _DEFAULT_NOZZLE_COUNT = 8 if not isinstance(ref, Container): raise TypeError(f"ref must be a Container but it was {type(ref)}.") 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): raise ValueError( f"Specified dispense columns: {columns} contains a column index that is outside of the valid columns: {ref_cols} for ref: {ref}." ) # pre-evaluate all parameters before making any changes to the Protocol if flowrate is not None: flowrate = parse_unit(flowrate, "uL/s") if nozzle_position is not None: 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( **asdict(shape) if isinstance(shape, DispenseShape) else shape ) if shake_after is not None: 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 ) if step_size is not None: step_size = parse_unit(step_size, "uL") if step_size not in _VALID_STEP_SIZES: raise ValueError( f"specified step_size was {step_size} but it must be in " f"{_VALID_STEP_SIZES}" ) for c in columns: 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." ) if pre_dispense is not None: invalid_pre_dispense_range = pre_dispense < 2 * step_size if invalid_pre_dispense_range and pre_dispense != Unit(0, "uL"): raise ValueError( f"Dispense pre_dispense must either be 0:uL or at least 2 * step_size: {step_size} but it was {pre_dispense}." ) if pre_dispense % step_size != Unit(0, "uL"): raise ValueError( f"Dispense pre_dispense must be a multiple of step_size: {step_size} but it was {pre_dispense}." ) row_count = ref.container_type.row_count() if isinstance(reagent, Well): self._remove_cover(reagent.container, "dispense from") # Volume accounting 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: reagent.volume -= total_vol_dispensed else: reagent.volume = -total_vol_dispensed reagent, resource_id, reagent_source = None, None, reagent else: if not isinstance(reagent, str): raise TypeError( f"reagent: {reagent} must be a Well or string but it was: " f"{type(reagent)}." ) if is_resource_id: reagent, resource_id, reagent_source = None, reagent, None else: # reagent remains unchanged resource_id, reagent_source = None, None self._remove_cover(ref, "dispense to") for c in columns: wells = ref.wells_from(c.column, row_count, columnwise=True) for w in wells: if w.volume: w.volume += c.volume else: w.volume = c.volume return self._append_and_return( Dispense( object=ref, columns=[asdict(c) for c in columns], reagent=reagent, resource_id=resource_id, reagent_source=reagent_source, step_size=step_size, flowrate=flowrate, nozzle_position=nozzle_position, pre_dispense=pre_dispense, shape=shape, shake_after=shake_after, ) )
[docs] def dispense_full_plate( self, ref: Container, reagent: Union[str, Well], volume: VOLUME, is_resource_id: bool = False, step_size: VOLUME = "5:uL", flowrate: Optional[FLOW_RATE] = None, nozzle_position: Optional[DispenseNozzlePosition] = None, pre_dispense: Optional[VOLUME] = None, shape: Optional[DispenseShape] = None, shake_after: Optional[DispenseShakeAfter] = None, ): """ Dispense the specified amount of the specified reagent to every well of a container using a reagent dispenser. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.dispense_full_plate(sample_plate, "water", "100:microliter") Autoprotocol Output: .. code-block:: none "instructions": [ { "reagent": "water", "object": "sample_plate", "columns": [ { "column": 0, "volume": "100:microliter" }, { "column": 1, "volume": "100:microliter" }, { "column": 2, "volume": "100:microliter" }, { "column": 3, "volume": "100:microliter" }, { "column": 4, "volume": "100:microliter" }, { "column": 5, "volume": "100:microliter" }, { "column": 6, "volume": "100:microliter" }, { "column": 7, "volume": "100:microliter" }, { "column": 8, "volume": "100:microliter" }, { "column": 9, "volume": "100:microliter" }, { "column": 10, "volume": "100:microliter" }, { "column": 11, "volume": "100:microliter" } ], "op": "dispense" } ] Parameters ---------- ref : Container Container for reagent to be dispensed to. reagent : str or Well Reagent to be dispensed. Use a string to specify the name or resource_id (see below) of the reagent to be dispensed. Alternatively, use a well to specify that the dispense operation must be executed using a specific aliquot as the dispense source. volume : Unit or str Volume of reagent to be dispensed to each well is_resource_id : bool, optional If true, interprets reagent as a resource ID step_size : str or Unit, optional Specifies that the dispense operation must be executed using a peristaltic pump with the given step size. Note that the volume dispensed in each column must be an integer multiple of the step_size. Currently, step_size must one of 5 uL, 0.5 uL, 0.05 uL. If set to None, will use vendor specified defaults. flowrate : str or Unit, optional The rate at which liquid is dispensed into the ref in units of volume/time. nozzle_position : dict, optional A dict represent nozzle offsets from the bottom middle of the plate's wells. see Dispense.builders.nozzle_position; specified as {"position_x": Unit, "position_y": Unit, "position_z": Unit}. pre_dispense : str or Unit, optional The volume of reagent to be dispensed per-nozzle into waste immediately prior to dispensing into the ref. shape: dict, optional The shape of the dispensing head to be used for the dispense. See liquid_handle_builders.shape_builder; specified as {"rows": int, "columns": int, "format": str} with format being a valid SBS format. shake_after: dict, optional Parameters that specify how a plate should be shaken at the very end of the instruction execution. See Dispense.builders.shake_after. Returns ------- Dispense Returns the :py:class:`autoprotocol.instruction.Dispense` instruction created from the specified parameters """ columns = Dispense.builders.columns( [ {"column": col, "volume": volume} for col in range(ref.container_type.col_count) ] ) return self.dispense( ref, reagent, columns, is_resource_id, step_size, flowrate, nozzle_position, pre_dispense, shape, shake_after, )
[docs] def spin( self, ref: Container, acceleration: str, duration: TIME, flow_direction: Optional[str] = None, spin_direction: Optional[List[str]] = None, ): """ Apply acceleration to a container. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.spin(sample_plate, "1000:g", "20:minute", flow_direction="outward") Autoprotocol Output: .. code-block:: none "instructions": [ { "acceleration": "1000:g", "duration": "20:minute", "flow_direction": "outward", "spin_direction": [ "cw", "ccw" ] "object": "sample_plate", "op": "spin" } ] Parameters ---------- ref : Container The container to be centrifuged. acceleration: str Acceleration to be applied to the plate, in units of `g` or `meter/second^2`. duration: str or Unit Length of time that acceleration should be applied. flow_direction: str Specifies the direction contents will tend toward with respect to the container. Valid directions are "inward" and "outward", default value is "inward". spin_direction: list(str) A list of "cw" (clockwise), "cww" (counterclockwise). For each element in the list, the container will be spun in the stated direction for the set "acceleration" and "duration". Default values are derived from the "flow_direction" parameter. If "flow_direction" is "outward", then "spin_direction" defaults to ["cw", "ccw"]. If "flow_direction" is "inward", then "spin_direction" defaults to ["cw"]. Returns ------- Spin Returns the :py:class:`autoprotocol.instruction.Spin` instruction created from the specified parameters Raises ------ TypeError If ref to spin is not of type Container. TypeError If spin_direction or flow_direction are not properly formatted. ValueError If spin_direction or flow_direction do not have appropriate values. """ if flow_direction is not None and flow_direction not in ["inward", "outward"]: raise ValueError( "The specified value for flow_direction was not valid. If " "specifying, please choose either 'inward' or 'outward'" ) default_directions = {"inward": ["cw"], "outward": ["cw", "ccw"]} if spin_direction is None and flow_direction: spin_direction = default_directions[flow_direction] if spin_direction is not None: if not isinstance(spin_direction, list): raise TypeError("Spin directions must be in the form of a list.") if len(spin_direction) == 0: raise ValueError( "Spin direction must be a list containing at least one " "spin direction ('cw', 'ccw')" ) if spin_direction and not all(s in ["cw", "ccw"] for s in spin_direction): raise ValueError("Spin directions must be 'cw' or 'ccw'.") try: duration = Unit(duration) except ValueError as e: raise ValueError(f"Duration must be a unit. {e}") from e if not isinstance(ref, Container): raise TypeError("Ref must be of type Container.") if not flow_direction or flow_direction == "inward": self._add_cover(ref, "inward spin") elif flow_direction == "outward": self._remove_cover(ref, "outward spin") return self._append_and_return( Spin(ref, acceleration, duration, flow_direction, spin_direction) )
[docs] def agitate( self, ref: Container, mode: AgitateMode, speed: ACCELERATION, duration: TIME, temperature: Optional[TEMPERATURE] = None, mode_params: Optional[AgitateModeParams] = None, ): """ Agitate a container in a specific condition for a given duration. If temperature is not specified, container is agitated at ambient temperature by default. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("test pcr plate", id=None, cont_type="96-pcr", storage="cold_4") p.agitate( ref = plate, mode="vortex", speed="1000:rpm", duration="5:minute", temperature="25:celsius" ) Autoprotocol Output: .. code-block:: none "instructions" : [ { "object": "test pcr plate", "mode": "vortex", "speed": "1000:rpm", "duration": "5:minute", "temperature": "25:celsius", "op": "agitate" } ] Parameters ---------- ref : Container Container to be agitated mode : str Mode by which to agitate container speed : str or Unit Speed at which to agitate container duration : str or Unit Specify the duration to agitate for temperature : Unit or str, optional Specify target temperature to agitate container at. Defaults to ambient mode_params : dict, optional Dictionary containing mode params for agitation modes Returns ------- Agitate returns a :py:class:`autoprotocol.instruction.Agitate` instruction created from the specified parameters Raises ------ ValueError If ref provided is not of type Container ValueError If speed is less than 0 rpm ValueError if duration is less than 0 minutes ValueError If `mode_params` not specified for mode `stir_bar` ValueError If valid keys for `mode_params` used for `stir_bar` are not included ValueError If wells specified in `mode_params` are not in the same container ValueError If `bar_shape` is not valid ValueError If `bar_length` is less than 0 millimeter ValueError If `mode` used does not require `mode_params` TypeError If ref cannot be undergo agitate mode `roll` or `invert` """ 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) temperature = parse_unit(temperature, "celsius") if temperature else None duration = parse_unit(duration, "minute") if not isinstance(ref, Container): raise ValueError("Ref is not of type Container.") if speed <= Unit("0:rpm"): raise ValueError(f"Speed: {speed} must be more than 0 rpm.") if duration <= Unit("0:minute"): raise ValueError(f"Duration: {duration} must be longer than 0 minutes.") if mode not in valid_modes: raise ValueError(f"Agitate mode must be one of {valid_modes}") if mode == "stir_bar": if mode_params is None: raise ValueError( "Dictionary `mode_params` must be specified for the " "mode `stir_bar`" ) else: if isinstance(mode_params, dict): try: mode_params = AgitateModeParams(**mode_params) except: raise ValueError( f"mode_params {mode_params.keys()} to not match {valid_bar_mode_params}" ) wells = WellGroup(mode_params.wells) container = set([w.container for w in wells]) shape = mode_params.bar_shape length = parse_unit(mode_params.bar_length, "millimeter") if len(container) > 1: raise ValueError( "All wells need to be on the same container for Agitate" ) if shape not in valid_bar_shapes: raise ValueError(f"Param `bar_shape` must be one of {valid_bar_shapes}") if length <= Unit(0, "millimeter"): raise ValueError( "Params `bar_length` must be greater than 0 millimeter" ) elif mode != "stir_bar" and mode_params: raise ValueError(f"Mode {mode} does not need mode_params specified") elif mode in ["invert", "roll"] and not ref.container_type.is_tube: raise TypeError(f"Specified container {ref} cannot be inverted or rolled.") return self._append_and_return( Agitate(ref, mode, speed, duration, temperature, mode_params) )
[docs] def thermocycle( self, ref: Container, groups: List[Union[ThermocycleTemperature, ThermocycleTemperatureGradient]], volume: Optional[VOLUME] = "10:microliter", dataref: Optional[str] = None, dyes: Optional[Dict[str, str]] = None, melting_start: Optional[TEMPERATURE] = None, melting_end: Optional[TEMPERATURE] = None, melting_increment: Optional[TEMPERATURE] = None, melting_rate: Optional[TEMPERATURE] = None, lid_temperature: Optional[TEMPERATURE] = None, ): """ Append a Thermocycle instruction to the list of instructions, with groups is a list(dict) in the form of: .. code-block:: python "groups": [{ "cycles": integer, "steps": [ { "duration": duration, "temperature": temperature, "read": boolean // optional (default false) }, { "duration": duration, "gradient": { "top": temperature, "bottom": temperature }, "read": boolean // optional (default false) } ] }], Thermocycle can also be used for either conventional or row-wise gradient PCR as well as qPCR. Refer to the examples below for details. Example Usage: To thermocycle a container according to the protocol: * 1 cycle: * 95 degrees for 5 minutes * 30 cycles: * 95 degrees for 30 seconds * 56 degrees for 20 seconds * 72 degrees for 30 seconds * 1 cycle: * 72 degrees for 10 minutes * 1 cycle: * 4 degrees for 30 seconds * all cycles: Lid temperature at 97 degrees .. code-block:: python from instruction import Thermocycle p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") # a plate must be sealed before it can be thermocycled p.seal(sample_plate) p.thermocycle( sample_plate, [ Thermocycle.builders.group( steps=[ Thermocycle.builders.step("95:celsius", "5:minute") ] ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step("95:celsius", "30:s"), Thermocycle.builders.step("56:celsius", "20:s"), Thermocycle.builders.step("72:celsius", "20:s"), ], cycles=30 ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step("72:celsius", "10:minute") ] ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step("4:celsius", "30:s") ] ) ], lid_temperature="97:celsius" ) Autoprotocol Output: .. code-block:: none "instructions": [ { "object": "sample_plate", "op": "seal" }, { "volume": "10:microliter", "dataref": null, "object": "sample_plate", "groups": [ { "cycles": 1, "steps": [ { "duration": "5:minute", "temperature": "95:celsius" } ] }, { "cycles": 30, "steps": [ { "duration": "30:second", "temperature": "95:celsius" }, { "duration": "20:second", "temperature": "56:celsius" }, { "duration": "20:second", "temperature": "72:celsius" } ] }, { "cycles": 1, "steps": [ { "duration": "10:minute", "temperature": "72:celsius" } ] }, { "cycles": 1, "steps": [ { "duration": "30:second", "temperature": "4:celsius" } ] } ], "op": "thermocycle" } ] To gradient thermocycle a container according to the protocol: * 1 cycle: * 95 degrees for 5 minutes * 30 cycles: * 95 degrees for 30 seconds Top Row: * 65 degrees for 20 seconds Bottom Row: * 55 degrees for 20 seconds * 72 degrees for 30 seconds * 1 cycle: * 72 degrees for 10 minutes .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") # a plate must be sealed before it can be thermocycled p.seal(sample_plate) p.thermocycle( sample_plate, [ Thermocycle.builders.group( steps=[ Thermocycle.builders.step("95:celsius", "5:minute") ] ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step("95:celsius", "30:s"), Thermocycle.builders.step( {"top": "65:celsius", "bottom": "55:celsius"}, "20:s" ), Thermocycle.builders.step("72:celsius", "20:s"), ], cycles=30 ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step("72:celsius", "10:minute") ] ) ] ) To conduct a qPCR, at least one dye type and the dataref field has to be specified. The example below uses SYBR dye and the following temperature profile: * 1 cycle: * 95 degrees for 3 minutes * 40 cycles: * 95 degrees for 10 seconds * 60 degrees for 30 seconds (Read during extension) .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") # a plate must be sealed before it can be thermocycled p.seal(sample_plate) p.thermocycle( sample_plate, [ Thermocycle.builders.group( steps=[ Thermocycle.builders.step("95:celsius", "3:minute") ] ), Thermocycle.builders.group( steps=[ Thermocycle.builders.step( "95:celsius", "10:second", read=False ), Thermocycle.builders.step( "95:celsius", "10:second", read=True ) ], cycles=40 ) ], dataref = "my_qpcr_data", dyes = {"SYBR": sample_plate.all_wells().indices()} ) Parameters ---------- ref : Container Container to be thermocycled. groups : list(dict) List of thermocycling instructions formatted as above volume : str or Unit, optional Volume contained in wells being thermocycled dataref : str, optional Name of dataref representing read data if performing qPCR dyes : dict, optional Dictionary mapping dye types to the wells they're used in melting_start: str or Unit, optional Temperature at which to start the melting curve. melting_end: str or Unit, optional Temperature at which to end the melting curve. melting_increment: str or Unit, optional Temperature by which to increment the melting curve. Accepted increment values are between 0.1 and 9.9 degrees celsius. melting_rate: str or Unit, optional Specifies the duration of each temperature step in the melting curve. lid_temperature: str or Unit, optional Specifies the lid temperature throughout the duration of the thermocycling instruction Returns ------- Thermocycle Returns the :py:class:`autoprotocol.instruction.Thermocycle` instruction created from the specified parameters Raises ------ AttributeError If groups are not properly formatted TypeError If ref to thermocycle is not of type Container. ValueError Container specified cannot be thermocycled ValueError Lid temperature is not within bounds """ if not isinstance(ref, Container): raise TypeError("Ref must be of type Container.") if "thermocycle" not in ref.container_type.capabilities: raise ValueError( f"Container '{ref.name}' type '{ref.container_type.shortname}', cannot be thermocycled." ) groups = [Thermocycle.builders.group(**_) for _ in groups] dyes = Thermocycle.builders.dyes(**(dyes or {})) melting = Thermocycle.builders.melting( melting_start, melting_end, melting_increment, melting_rate ) # Constants are currently based off the Biorad thermocyclers, and # assumes that they are generally reflective of other thermocyclers _MIN_LID_TEMP = Unit("30:celsius") _MAX_LID_TEMP = Unit("110:celsius") if lid_temperature is not None: lid_temperature = parse_unit(lid_temperature) if not (_MIN_LID_TEMP <= lid_temperature <= _MAX_LID_TEMP): raise ValueError( f"Lid temperature {lid_temperature} has to be within " f"[{_MIN_LID_TEMP}, {_MAX_LID_TEMP}]" ) self._add_seal(ref, "thermocycle") return self._append_and_return( Thermocycle( object=ref, groups=groups, volume=volume, dataref=dataref, dyes=dyes, melting=melting, lid_temperature=lid_temperature, ) )
[docs] def incubate( self, ref: Union[Container, str], where: str, duration: TIME, shaking: bool = False, co2: float = 0, uncovered: bool = False, target_temperature: Optional[TEMPERATURE] = None, shaking_params: Optional[IncubateShakingParams] = None, ): """ Move plate to designated thermoisolater or ambient area for incubation for specified duration. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") # a plate must be sealed/covered before it can be incubated p.seal(sample_plate) p.incubate(sample_plate, "warm_37", "1:hour", shaking=True) Autoprotocol Output: .. code-block:: none "instructions": [ { "object": "sample_plate", "op": "seal" }, { "duration": "1:hour", "where": "warm_37", "object": "sample_plate", "shaking": true, "op": "incubate", "co2_percent": 0 } ] Parameters ---------- ref : Ref or str The container to be incubated where : Enum({"ambient", "warm_37", "cold_4", "cold_20", "cold_80"}) Temperature at which to incubate specified container duration : Unit or str Length of time to incubate container shaking : bool, optional Specify whether or not to shake container if available at the specified temperature co2 : Number, optional Carbon dioxide percentage uncovered: bool, optional Specify whether the container should be uncovered during incubation target_temperature : Unit or str, optional Specify a target temperature for a device (eg. an incubating block) to reach during the specified duration. shaking_params: dict, optional Specify "path" and "frequency" of shaking parameters to be used with compatible devices (eg. thermoshakes) Returns ------- Incubate Returns the :py:class:`autoprotocol.instruction.Incubate` instruction created from the specified parameters Raises ------ TypeError Invalid input types given, e.g. ref is not of type Container RuntimeError Incubating uncovered in a location which is shaking """ if not isinstance(ref, Container): raise TypeError("Ref needs to be of type Container") allowed_uncovered = ["ambient"] if uncovered and (where not in allowed_uncovered or shaking): raise RuntimeError( f"If incubating uncovered, location must be in " f"{', '.join(allowed_uncovered)} and not shaking." ) if not isinstance(co2, Number): raise TypeError("co2 must be a number.") if target_temperature: target_temperature = parse_unit(target_temperature, "celsius") if not uncovered: self._add_cover(ref, "incubate") return self._append_and_return( Incubate( ref, where, duration, shaking, co2, target_temperature, shaking_params ) )
[docs] def absorbance( self, ref: Union[str, Container], wells: WellParam, wavelength: WAVELENGTH, dataref: str, flashes: int = 25, incubate_before: Optional[PlateReaderIncubateBefore] = None, temperature: Optional[TEMPERATURE] = None, settle_time: Optional[TIME] = None, ): """ Read the absorbance for the indicated wavelength for the indicated wells. Append an Absorbance instruction to the list of instructions for this Protocol object. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.absorbance(sample_plate, sample_plate.wells_from(0,12), "600:nanometer", "test_reading", flashes=50) Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "test_reading", "object": "sample_plate", "wells": [ "A1", "A2", "A3", "A4", "A5", "A6", "A7", "A8", "A9", "A10", "A11", "A12" ], "num_flashes": 50, "wavelength": "600:nanometer", "op": "absorbance" } ] Parameters ---------- ref : str or Container Object to execute the absorbance read on wells : list(Well) or WellGroup or Well WellGroup of wells to be measured or a list of well references in the form of ["A1", "B1", "C5", ...] wavelength : str or Unit wavelength of light absorbance to be read for the indicated wells dataref : str name of this specific dataset of measured absorbances flashes : int, optional number of flashes for the read temperature: str or Unit, optional set temperature to heat plate reading chamber settle_time: Unit, optional the time before the start of the measurement, defaults to vendor specifications incubate_before: dict, optional parameters for incubation before executing the plate read See Also :meth:`Absorbance.builders.incubate_params` Returns ------- Absorbance Returns the :py:class:`autoprotocol.instruction.Absorbance` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. wells given is of type Well, WellGroup or list of wells ValueError Wells specified are not from the same container ValueError Settle time has to be greater than 0 UnitError Settle time is not of type Unit """ wells = WellGroup(wells) container = set([w.container for w in wells]) if len(container) > 1: raise ValueError( "All wells need to be on the same container for Absorbance" ) wells = [str(w.index) for w in wells] if incubate_before: Absorbance.builders.incubate_params(**incubate_before) if settle_time: try: settle_time = Unit(settle_time) if settle_time < Unit(0, "second"): raise ValueError( "'settle_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'settle_time' must be of type Unit.") from e return self._append_and_return( Absorbance( ref, wells, wavelength, dataref, flashes, incubate_before, temperature, settle_time, ) )
[docs] def fluorescence( self, ref: Union[str, Container], wells: WellParam, excitation: WAVELENGTH, emission: WAVELENGTH, dataref: str, flashes: Optional[int] = 25, temperature: Optional[TEMPERATURE] = None, gain: Optional[float] = None, incubate_before: Optional[PlateReaderIncubateBefore] = None, detection_mode: Optional[str] = None, position_z: Optional[ Union[PlateReaderPositionZCalculated, PlateReaderPositionZManual] ] = None, settle_time: Optional[TIME] = None, lag_time: Optional[TIME] = None, integration_time: Optional[str] = None, ): """ Read the fluoresence for the indicated wavelength for the indicated wells. Append a Fluorescence instruction to the list of instructions for this Protocol object. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.fluorescence(sample_plate, sample_plate.wells_from(0,12), excitation="587:nanometer", emission="610:nanometer", dataref="test_reading") Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "test_reading", "excitation": "587:nanometer", "object": "sample_plate", "emission": "610:nanometer", "wells": [ "A1", "A2", "A3", "A4", "A5", "A6", "A7", "A8", "A9", "A10", "A11", "A12" ], "num_flashes": 25, "op": "fluorescence" } ] Parameters ---------- ref : str or Container Container to plate read. wells : list(Well) or WellGroup or Well WellGroup of wells to be measured or a list of well references in the form of ["A1", "B1", "C5", ...] excitation : str or Unit Wavelength of light used to excite the wells indicated emission : str or Unit Wavelength of light to be measured for the indicated wells dataref : str Name of this specific dataset of measured fluoresence flashes : int, optional Number of flashes. temperature: str or Unit, optional set temperature to heat plate reading chamber gain: float, optional float between 0 and 1, multiplier, gain=0.2 of maximum signal amplification incubate_before: dict, optional parameters for incubation before executing the plate read See Also :meth:`Fluorescence.builders.incubate_params` detection_mode: str, optional set the detection mode of the optics, ["top", "bottom"], defaults to vendor specified defaults. position_z: dict, optional distance from the optics to the surface of the plate transport, only valid for "top" detection_mode and vendor capabilities. Specified as either a set distance - "manual", OR calculated from a WellGroup - "calculated_from_wells". Only one position_z determination may be specified .. code-block:: none position_z = { "manual": Unit - OR - "calculated_from_wells": [] } settle_time: Unit, optional the time before the start of the measurement, defaults to vendor specifications lag_time: Unit, optional time between flashes and the start of the signal integration, defaults to vendor specifications integration_time: Unit, optional duration of the signal recording, per Well, defaults to vendor specifications Examples -------- position_z: .. code-block:: none position_z = { "calculated_from_wells": ["plate/A1", "plate/A2"] } -OR- position_z = { "manual": "20:micrometer" } Returns ------- Fluorescence Returns the :py:class:`autoprotocol.instruction.Fluorescence` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. wells given is of type Well, WellGroup or list of wells ValueError Wells specified are not from the same container ValueError Settle time, integration time or lag time has to be greater than 0 UnitError Settle time, integration time, lag time or position z is not of type Unit ValueError Unknown value given for `detection_mode` ValueError Position z specified for non-top detection mode KeyError For position_z, only `manual` and `calculated_from_wells` is allowed NotImplementedError Specifying `calculated_from_wells` as that has not been implemented yet """ wells = WellGroup(wells) container = set([w.container for w in wells]) if len(container) > 1: raise ValueError( "All wells need to be on the same container for Fluorescence" ) wells = [str(w.index) for w in wells] if gain is not None and not (0 <= gain <= 1): raise ValueError( f"fluorescence gain set to {gain} must be between 0 and 1, " f"inclusive" ) if incubate_before: Fluorescence.builders.incubate_params(**incubate_before) valid_detection_modes = ["top", "bottom"] if detection_mode and detection_mode not in valid_detection_modes: raise ValueError( f"Unknown value for 'detection_mode'. Must be one of " f"{valid_detection_modes}." ) if detection_mode == "bottom" and position_z: raise ValueError( "position_z is only valid for 'top' detection_mode " "measurements." ) if settle_time: try: settle_time = Unit(settle_time) if settle_time < Unit(0, "second"): raise ValueError( "'settle_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'settle_time' must be of type Unit.") from e if lag_time: try: lag_time = Unit(lag_time) if lag_time < Unit(0, "second"): raise ValueError( "'lag_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'lag_time' must be of type Unit.") from e if integration_time: try: integration_time = Unit(integration_time) if integration_time < Unit(0, "second"): raise ValueError( "'integration_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'integration_time' must be of type Unit.") from e if position_z: valid_pos_z = ["manual", "calculated_from_wells"] if not isinstance(position_z, dict): raise TypeError("'position_z' must be of type dict.") if len(position_z.keys()) > 1: raise ValueError( "'position_z' can only have one mode of calculation " "specified: 'manual' or 'calculated_from_wells'." ) for k in position_z.keys(): if k not in valid_pos_z: raise KeyError( f"'position_z' keys can only be 'manual' or 'calculated_from_wells'. '{k}' is not a recognized key." ) if "manual" in position_z.keys(): try: manual = Unit(position_z["manual"]) if manual < Unit(0, "micrometer"): raise ValueError( "'manual' z_position must be a length equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'manual' position_z must be of type Unit.") from e if "calculated_from_wells" in position_z.keys(): # blocking calculated_from_wells until fully implemented # remove below RunTimeError to release feature raise NotImplementedError( "This feature, 'calculated_from_wells', " "has not been implemented yet. Please use " "'position_z':{'manual': 'set_value'} to " "specify a position_z." ) # pragma pylint: disable=unreachable, unused-variable z_ws = position_z["calculated_from_wells"] if isinstance(z_ws, Well): z_ws = [z_ws] position_z["calculated_from_wells"] = z_ws elif isinstance(z_ws, list): if not all(isinstance(w, Well) for w in z_ws): raise TypeError( "All elements in list must be wells for " "position_z, 'calculated_from_wells'." ) else: raise TypeError( "Wells specified for 'calculated_from_wells' " "must be Well, list of wells, WellGroup." ) # check z_ws against container/ref for measurement # if ref is Container if isinstance(ref, Container): try: valid_z_ws = ref.wells(z_ws) except ValueError as e: raise ValueError( "Well indices specified for " "'calculated_from_wells' must " "be valid wells of the ref'd " "container." ) from e # pragma pylint: enable=unreachable, unused-variable return self._append_and_return( Fluorescence( ref, wells, excitation, emission, dataref, flashes, incubate_before, temperature, gain, detection_mode, position_z, settle_time, lag_time, integration_time, ) )
[docs] def luminescence( self, ref: Union[str, Container], wells: WellParam, dataref: str, incubate_before: Union[PlateReaderIncubateBefore] = None, temperature: Optional[TEMPERATURE] = None, settle_time: Optional[TIME] = None, integration_time: Optional[TIME] = None, ): """ Read luminescence of indicated wells. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.luminescence(sample_plate, sample_plate.wells_from(0,12), "test_reading") Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "test_reading", "object": "sample_plate", "wells": [ "A1", "A2", "A3", "A4", "A5", "A6", "A7", "A8", "A9", "A10", "A11", "A12" ], "op": "luminescence" } ] Parameters ---------- ref : str or Container Container to plate read. wells : list(Well) or WellGroup or Well WellGroup of wells to be measured or a list of well references in the form of ["A1", "B1", "C5", ...] dataref : str Name of this dataset of measured luminescence readings. temperature: str or Unit, optional set temperature to heat plate reading chamber settle_time: Unit, optional the time before the start of the measurement, defaults to vendor specifications incubate_before: dict, optional parameters for incubation before executing the plate read See Also :meth:`Absorbance.builders.incubate_params` integration_time: Unit, optional duration of the signal recording, per Well, defaults to vendor specifications Returns ------- Luminescence Returns the :py:class:`autoprotocol.instruction.Luminescence` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. wells given is of type Well, WellGroup or list of wells ValueError Wells specified are not from the same container ValueError Settle time or integration time has to be greater than 0 UnitError Settle time or integration time is not of type Unit """ wells = WellGroup(wells) container = set([w.container for w in wells]) if len(container) > 1: raise ValueError( "All wells need to be on the same container for Luminescence" ) wells = [str(w.index) for w in wells] if incubate_before: Luminescence.builders.incubate_params(**incubate_before) if settle_time: try: settle_time = Unit(settle_time) if settle_time < Unit(0, "second"): raise ValueError( "'settle_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'settle_time' must be of type Unit.") from e if integration_time: try: integration_time = Unit(integration_time) if integration_time < Unit(0, "second"): raise ValueError( "'integration_time' must be a time equal " "to or greater than 0." ) except UnitError as e: raise UnitError("'integration_time' must be of type Unit.") from e return self._append_and_return( Luminescence( ref, wells, dataref, incubate_before, temperature, settle_time, integration_time, ) )
[docs] def gel_separate( self, wells: WellParam, volume: VOLUME, matrix: str, ladder: str, duration: TIME, dataref: str, ): """ Separate nucleic acids on an agarose gel. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.gel_separate(sample_plate.wells_from(0,12), "10:microliter", "agarose(8,0.8%)", "ladder1", "11:minute", "genotyping_030214") Autoprotocol Output: .. code-block:: none "instructions": [ { "dataref": "genotyping_030214", "matrix": "agarose(8,0.8%)", "volume": "10:microliter", "ladder": "ladder1", "objects": [ "sample_plate/0", "sample_plate/1", "sample_plate/2", "sample_plate/3", "sample_plate/4", "sample_plate/5", "sample_plate/6", "sample_plate/7", "sample_plate/8", "sample_plate/9", "sample_plate/10", "sample_plate/11" ], "duration": "11:minute", "op": "gel_separate" } ] Parameters ---------- wells : list(Well) or WellGroup or Well List of wells or WellGroup containing wells to be separated on gel. volume : str or Unit Volume of liquid to be transferred from each well specified to a lane of the gel. matrix : str Matrix (gel) in which to gel separate samples ladder : str Ladder by which to measure separated fragment size duration : str or Unit Length of time to run current through gel. dataref : str Name of this set of gel separation results. Returns ------- GelSeparate Returns the :py:class:`autoprotocol.instruction.GelSeparate` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. wells given is of type Well, WellGroup or list of wells ValueError Specifying more wells than the number of available lanes in the selected matrix """ # Check valid well inputs if not is_valid_well(wells): raise TypeError( "Wells must be of type Well, list of Wells, or " "WellGroup." ) if isinstance(wells, Well): wells = WellGroup(wells) for w in wells: self._remove_cover(w.container, "gel separate") max_well = int(matrix.split("(", 1)[1].split(",", 1)[0]) if len(wells) > max_well: raise ValueError( "Number of wells is greater than available" "lanes in matrix" ) return self._append_and_return( GelSeparate(wells, volume, matrix, ladder, duration, dataref) )
[docs] def gel_purify( self, extracts: List[GelPurifyExtract], volume: VOLUME, matrix: str, ladder: str, dataref: str, ): """ Separate nucleic acids on an agarose gel and purify according to parameters. If gel extract lanes are not specified, they will be sequentially ordered and purified on as many gels as necessary. Each element in extracts specifies a source loaded in a single lane of gel with a list of bands that will be purified from that lane. If the same source is to be run on separate lanes, a new dictionary must be added to extracts. It is also possible to add an element to extract with a source but without a list of bands. In that case, the source will be run in a lane without extraction. Example Usage: .. code-block:: python p = Protocol() sample_wells = p.ref("test_plate", None, "96-pcr", discard=True).wells_from(0, 8) extract_wells = [p.ref("extract_" + str(i.index), None, "micro-1.5", storage="cold_4").well(0) for i in sample_wells] extracts = [make_gel_extract_params( w, make_band_param( "TE", "5:microliter", 80, 79, extract_wells[i])) for i, w in enumerate(sample_wells)] p.gel_purify(extracts, "10:microliter", "size_select(8,0.8%)", "ladder1", "gel_purify_example") Autoprotocol Output: For extracts[0] .. code-block:: none { "band_list": [ { "band_size_range": { "max_bp": 80, "min_bp": 79 }, "destination": Well(Container(extract_0), 0, None), "elution_buffer": "TE", "elution_volume": "Unit(5.0, 'microliter')" } ], "gel": None, "lane": None, "source": Well(Container(test_plate), 0, None) } Parameters ---------- extracts: list(dict) List of gel extraction parameters See Also :meth:`GelPurify.builders.extract` volume: str or Unit Volume of liquid to be transferred from each well specified to a lane of the gel. matrix: str Matrix (gel) in which to gel separate samples ladder: str Ladder by which to measure separated fragment size dataref: str Name of this set of gel separation results. Returns ------- GelPurify Returns the :py:class:`autoprotocol.instruction.GelPurify` instruction created from the specified parameters Raises ------- RuntimeError If matrix is not properly formatted. AttributeError If extract parameters are not a list of dictionaries. KeyError If extract parameters do not contain the specified parameter keys. ValueError If min_bp is greater than max_bp. ValueError If extract destination is not of type Well. ValueError If extract elution volume is not of type Unit ValueError if extract elution volume is not greater than 0. RuntimeError If gel extract lanes are set for some but not all extract wells. RuntimeError If all samples do not fit on single gel type. TypeError If lane designated for gel extracts is not an integer. RuntimeError If designated lane index is outside lanes within the gel. RuntimeError If lanes not designated and number of extracts not equal to number of samples. """ from itertools import groupby from operator import itemgetter try: max_well = int(matrix.split("(", 1)[1].split(",", 1)[0]) except (AttributeError, IndexError) as e: raise RuntimeError("Matrix specified is not properly formatted.") from e volume = Unit(volume) if volume <= Unit("0:microliter"): raise ValueError(f"Volume: {volume}, must be greater than 0:microliter") if not isinstance(ladder, str): raise TypeError(f"Ladder: {ladder}, must be a string") if not isinstance(dataref, str): raise TypeError(f"Datref: {dataref}, must be a string") if not isinstance(extracts, list): extracts = [extracts] extracts = [GelPurify.builders.extract(**_) for _ in extracts] gel_set = [e["gel"] for e in extracts] lane_set = [e["lane"] for e in extracts] if None in gel_set: if any(gel_set): raise RuntimeError( "If one extract has gel set, all extracts must have " "gel set" ) else: if None in lane_set: if any(lane_set): raise RuntimeError( "If one extract has lane set, all extracts must " "have lane set" ) else: for i, e in enumerate(extracts): e["gel"] = i // max_well e["lane"] = i % max_well else: for e in extracts: e["gel"] = 0 sort_key = itemgetter("gel") parsed_extracts = [ list(grp) for key, grp in groupby(sorted(extracts, key=sort_key), key=sort_key) ] instructions = [] for pe in parsed_extracts: lane_set = [e["lane"] for e in pe] if len(lane_set) > max_well: raise RuntimeError( f"The gel is not large enough to accomodate all lanes: gel has {max_well} wells, {len(lane_set)} lanes specified" ) if None in lane_set: if any(lane_set): raise RuntimeError( "If one extract has lane set, all extracts must have " "lane set" ) else: for i, e in enumerate(pe): e["lane"] = i % max_well lane_set = [e["lane"] for e in pe] if sorted(lane_set) != list(range(0, max(lane_set) + 1)): raise RuntimeError("Lanes must be contiguous, unique, and start from 0") if len(parsed_extracts) > 1: dataref_gel = f"{dataref}_{pe[0]['gel']}" else: dataref_gel = dataref pe = sorted(pe, key=itemgetter("lane")) samples = [e["source"] for e in pe] pe_unpacked = [] for e in pe: for b in e["band_list"]: ext = b ext["lane"] = e["lane"] pe_unpacked.append(ext) instructions.append( self._append_and_return( GelPurify(samples, volume, matrix, ladder, dataref_gel, pe_unpacked) ) ) return instructions
[docs] def seal( self, ref: Container, type: Optional[str] = None, mode: Optional[str] = None, temperature: Optional[TEMPERATURE] = None, duration: Optional[TEMPERATURE] = None, ): """ Seal indicated container using the automated plate sealer. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") p.seal(sample_plate, mode="thermal", temperature="160:celsius") Autoprotocol Output: .. code-block:: none "instructions": [ { "object": "sample_plate", "type": "ultra-clear", "mode": "thermal", "mode_params": { "temperature": "160:celsius" } "op": "seal" } ] Parameters ---------- ref : Container Container to be sealed type : str, optional Seal type to be used, such as "ultra-clear" or "foil". mode: str, optional Sealing method to be used, such as "thermal" or "adhesive". Defaults to None, which is interpreted sensibly based on the execution environment. temperature: Unit or str, optional Temperature at which to melt the sealing film onto the ref. Only applicable to thermal sealing; not respected if the sealing mode is adhesive. If unspecified, thermal sealing temperature defaults correspond with manufacturer-recommended or internally-optimized values for the target container type. Applies only to thermal sealing. duration: Unit or str, optional Duration for which to press the (heated, if thermal) seal down on the ref. Defaults to manufacturer-recommended or internally- optimized seal times for the target container type. Currently applies only to thermal sealing. Returns ------- Seal Returns the :py:class:`autoprotocol.instruction.Seal` instruction created from the specified parameters Raises ------ TypeError If ref is not of type Container. RuntimeError If container type does not have `seal` capability. RuntimeError If seal is not a valid seal type. RuntimeError If the sealing mode is invalid, or incompatible with the given ref RuntimeError If thermal sealing params (temperature and/or duration) are specified alongside an adhesive sealing mode. RuntimeError If specified thermal sealing parameters are invalid RuntimeError If container is already covered with a lid. """ SEALING_MODES = ["thermal", "adhesive"] if not isinstance(ref, Container): raise TypeError("Container to seal must be of type Container.") if "seal" not in ref.container_type.capabilities: raise RuntimeError( f"Container '{ref.name}' type '{ref.container_type.shortname}'," f" cannot be sealed." ) if type is None: type = ref.container_type.seal_types[0] if type not in SEAL_TYPES: raise RuntimeError(f"{type} is not a valid seal type") if ref.is_covered(): raise RuntimeError("A container cannot be sealed over a lid.") if not (mode is None or mode in SEALING_MODES): raise RuntimeError(f"{mode} is not a valid sealing mode") if temperature is not None or duration is not None: if mode == "adhesive": raise RuntimeError( "Thermal sealing parameters `temperature` and `duration` " "are incompatible with the chosen adhesive sealing mode." ) mode = "thermal" mode_params = dict() if temperature is not None: mode_params["temperature"] = parse_unit(temperature, "celsius") if duration is not None: mode_params["duration"] = parse_unit(duration, "second") else: mode_params = None if not ref.is_sealed(): ref.cover = type return self._append_and_return(Seal(ref, type, mode, mode_params))
[docs] def unseal(self, ref: Container): """ Remove seal from indicated container using the automated plate unsealer. Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-pcr", storage="warm_37") # a plate must be sealed to be unsealed p.seal(sample_plate) p.unseal(sample_plate) Autoprotocol Output: .. code-block:: none "instructions": [ { "object": "sample_plate", "op": "seal", "type": "ultra-clear" }, { "object": "sample_plate", "op": "unseal" } ] Parameters ---------- ref : Container Container to be unsealed. Returns ------- Unseal Returns the :py:class:`autoprotocol.instruction.Unseal` instruction created from the specified parameters Raises ------ TypeError If ref is not of type Container. RuntimeError If container is covered with a lid not a seal. """ if not isinstance(ref, Container): raise TypeError("Container to unseal must be of type Container.") if ref.is_covered(): raise RuntimeError( "A container with a cover cannot be unsealed, " "use the instruction uncover." ) if ref.is_sealed(): ref.cover = None unseal_inst = Unseal(ref) return self._append_and_return(unseal_inst)
[docs] def cover( self, ref: Container, lid: Optional[str] = None, retrieve_lid: Optional[bool] = None, ): """ Place specified lid type on specified container Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") p.cover(sample_plate, lid="universal") Autoprotocol Output: .. code-block:: none "instructions": [ { "lid": "universal", "object": "sample_plate", "op": "cover" } ] Parameters ---------- ref : Container Container to be convered. lid : str, optional Type of lid to cover the container. Must be a valid lid type for the container type. retrieve_lid: bool, optional Flag to retrieve lid from previously stored location (see uncover). Returns ------- Cover Returns the :py:class:`autoprotocol.instruction.Cover` instruction created from the specified parameters Raises ------ TypeError If ref is not of type Container. RuntimeError If container type does not have `cover` capability. RuntimeError If lid is not a valid lid type. RuntimeError If container is already sealed with a seal. TypeError If retrieve_lid is not a boolean. """ if not isinstance(ref, Container): raise TypeError("Container to cover must be of type Container.") if "cover" not in ref.container_type.capabilities: raise RuntimeError( f"Container '{ref.name}' type '{ref.container_type.shortname}'," f" cannot be covered." ) if lid is None: lid = ref.container_type.cover_types[0] if lid not in COVER_TYPES: raise RuntimeError(f"{lid} is not a valid lid type") if ref.is_sealed(): raise RuntimeError("A container cannot be covered over a seal.") if retrieve_lid is not None and not isinstance(retrieve_lid, bool): raise TypeError("Cover: retrieve_lid must be of type bool") if not ref.is_covered(): ref.cover = lid return self._append_and_return(Cover(ref, lid, retrieve_lid))
[docs] def uncover(self, ref: Container, store_lid: Optional[bool] = None): """ Remove lid from specified container Example Usage: .. code-block:: python p = Protocol() sample_plate = p.ref("sample_plate", None, "96-flat", storage="warm_37") # a plate must have a cover to be uncovered p.cover(sample_plate, lid="universal") p.uncover(sample_plate) Autoprotocol Output: .. code-block:: none "instructions": [ { "lid": "universal", "object": "sample_plate", "op": "cover" }, { "object": "sample_plate", "op": "uncover" } ] Parameters ---------- ref : Container Container to remove lid. store_lid: bool, optional Flag to store the uncovered lid. Returns ------- Uncover Returns the :py:class:`autoprotocol.instruction.Uncover` instruction created from the specified parameters Raises ------ TypeError If ref is not of type Container. RuntimeError If container is sealed with a seal not covered with a lid. TypeError If store_lid is not a boolean. """ if not isinstance(ref, Container): raise TypeError("Container to uncover must be of type Container.") if ref.is_sealed(): raise RuntimeError( "A container with a seal cannot be uncovered, " "use the instruction unseal." ) if store_lid is not None and not isinstance(store_lid, bool): raise TypeError("Uncover: store_lid must be of type bool") if ref.is_covered(): ref.cover = None return self._append_and_return(Uncover(ref, store_lid))
[docs] def flow_cytometry( self, dataref: str, samples: WellParam, lasers: List[FlowCytometryLaser], collection_conditions: FlowCytometryCollectionCondition, width_threshold: Optional[Union[int, float]] = None, window_extension: Optional[Union[int, float]] = None, remove_coincident_events: Optional[bool] = None, ): """ A non-ambiguous set of parameters for performing flow cytometry. Parameters ---------- dataref : str Name of dataset that will be returned. samples : list(Well) or Well or WellGroup Wells to be analyzed lasers : list(dict) See FlowCytometryBuilders.laser. collection_conditions : dict See FlowCytometryBuilders.collection_conditions. width_threshold : int or float, optional Threshold to determine width measurement. window_extension : int or float, optional Front and rear window extension. remove_coincident_events : bool, optional Remove coincident events. Returns ------- FlowCytometry Returns a :py:class:`autoprotocol.instruction.FlowCytometry` instruction created from the specified parameters. Raises ------ TypeError If `lasers` is not of type list. TypeError If `samples` is not of type Well, list of Well, or WellGroup. TypeError If `width_threshold` is not a number. TypeError If `window_extension` is not a number. TypeError If `remove_coincident_events` is not of type bool. Examples -------- Example flow cytometry protocol .. code-block:: python p = Protocol() plate = p.ref("sample-plate", cont_type="384-flat", discard=True) lasers = [FlowCytometry.builders.laser( excitation="405:nanometers", channels=[ FlowCytometry.builders.channel( emission_filter=FlowCytometry.builders.emission_filter( channel_name="VL1", shortpass="415:nanometers", longpass="465:nanometers" ), detector_gain="10:millivolts" ) ] )] collection_conds = FlowCytometry.builders.collection_conditions( acquisition_volume="5.0:ul", flowrate="12.5:ul/min", wait_time="10:seconds", mix_cycles=10, mix_volume="10:ul", rinse_cycles=10 ) p.flow_cytometry("flow-1234", plate.wells_from(0, 3), lasers, collection_conds) Autoprotocol Output: .. code-block:: json { "op": "flow_cytometry", "dataref": "flow-1234", "samples": [ "sample-plate/0", "sample-plate/1", "sample-plate/2" ], "lasers": [ { "excitation": "405:nanometer", "channels": [ { "emission_filter": { "channel_name": "VL1", "shortpass": "415:nanometer", "longpass": "465:nanometer" }, "detector_gain": "10:millivolt" } ] } ], "collection_conditions": { "acquisition_volume": "5:microliter", "flowrate": "12.5:microliter/minute", "stop_criteria": { "volume": "5:microliter" }, "wait_time": "10:second", "mix_cycles": 10, "mix_volume": "10:microliter", "rinse_cycles": 10 } } """ if not isinstance(lasers, list): raise TypeError("lasers must be of type list.") if not is_valid_well(samples): raise TypeError( "samples must be of type Well, list of Well, " "or WellGroup." ) if width_threshold is not None: if not isinstance(width_threshold, Number): raise TypeError("width_threshold must be a number.") if window_extension is not None: if not isinstance(window_extension, Number): raise TypeError("window_extension must be a number.") if remove_coincident_events is not None: if not isinstance(remove_coincident_events, bool): raise TypeError("remove_coincident_events must be of type " "bool.") lasers = [ FlowCytometry.builders.laser( channels=l.channels, excitation=l.excitation, power=l.power, area_scaling_factor=l.area_scaling_factor, ) for l in lasers ] collection_conditions = FlowCytometry.builders.collection_conditions( acquisition_volume=collection_conditions.acquisition_volume, flowrate=collection_conditions.flowrate, wait_time=collection_conditions.wait_time, mix_cycles=collection_conditions.mix_cycles, mix_volume=collection_conditions.mix_volume, rinse_cycles=collection_conditions.rinse_cycles, stop_criteria=collection_conditions.stop_criteria, ) return self._append_and_return( FlowCytometry( dataref, samples, lasers, collection_conditions, width_threshold, window_extension, remove_coincident_events, ) )
[docs] def flow_analyze( self, dataref: str, FSC: FlowAnalyzeChannel, SSC: FlowAnalyzeChannel, neg_controls: List[FlowAnalyzeNegControls], samples: List[FlowAnalyzeSample], colors: Optional[List[FlowAnalyzeColors]] = None, pos_controls: Optional[List[FlowAnalyzePosControls]] = None, ): """ Perform flow cytometry. The instruction will be executed within the voltage range specified for each channel, optimized for the best sample separation/distribution that can be achieved within these limits. The vendor will specify the device that this instruction is executed on and which excitation and emission spectra are available. At least one negative control is required, which will be used to define data acquisition parameters as well as to determine any autofluorescent properties for the sample set. Additional negative positive control samples are optional. Positive control samples will be used to optimize single color signals and, if desired, to minimize bleed into other channels. For each sample this instruction asks you to specify the `volume` and/or `captured_events`. Vendors might also require `captured_events` in case their device does not support volumetric sample intake. If both conditions are supported, the vendor will specify if data will be collected only until the first one is met or until both conditions are fulfilled. Example Usage: .. code-block:: python p = Protocol() dataref = "test_ref" FSC = {"voltage_range": {"low": "230:volt", "high": "280:volt"}, "area": True, "height": True, "weight": False} SSC = {"voltage_range": {"low": "230:volt", "high": "280:volt"}, "area": True, "height": True, "weight": False} neg_controls = {"well": "well0", "volume": "100:microliter", "captured_events": 5, "channel": "channel0"} samples = [ { "well": "well0", "volume": "100:microliter", "captured_events": 9 } ] p.flow_analyze(dataref, FSC, SSC, neg_controls, samples, colors=None, pos_controls=None) Autoprotocol Output: .. code-block:: json { "channels": { "FSC": { "voltage_range": { "high": "280:volt", "low": "230:volt" }, "area": true, "height": true, "weight": false }, "SSC": { "voltage_range": { "high": "280:volt", "low": "230:volt" }, "area": true, "height": true, "weight": false } }, "op": "flow_analyze", "negative_controls": { "channel": "channel0", "well": "well0", "volume": "100:microliter", "captured_events": 5 }, "dataref": "test_ref", "samples": [ { "well": "well0", "volume": "100:microliter", "captured_events": 9 } ] } Parameters ---------- dataref : str Name of flow analysis dataset generated. FSC : dict Dictionary containing FSC channel parameters in the form of: .. code-block:: none { "voltage_range": { "low": "230:volt", "high": "280:volt" }, "area": true, //default: true "height": true, //default: true "weight": false //default: false } SSC : dict Dictionary of SSC channel parameters in the form of: .. code-block:: none { "voltage_range": { "low": <voltage>, "high": <voltage>" }, "area": true, //default: true "height": true, //default: false "weight": false //default: false } neg_controls : list(dict) List of negative control wells in the form of: .. code-block:: none { "well": well, "volume": volume, "captured_events": integer, // optional, default infinity "channel": [channel_name] } at least one negative control is required. samples : list(dict) List of samples in the form of: .. code-block:: none { "well": well, "volume": volume, "captured_events": integer // optional, default infinity } at least one sample is required colors : list(dict), optional Optional list of colors in the form of: .. code-block:: none [ { "name": "FitC", "emission_wavelength": "495:nanometer", "excitation_wavelength": "519:nanometer", "voltage_range": { "low": <voltage>, "high": <voltage> }, "area": true, //default: true "height": false, //default: false "weight": false //default: false }, ... ] pos_controls : list(dict), optional Optional list of positive control wells in the form of: .. code-block:: none [ { "well": well, "volume": volume, "captured_events": integer, // default: infinity "channel": [channel_name], "minimize_bleed": [{ // optional "from": color, "to": [color] }, ... ] Returns ------- FlowAnalyze Returns the :py:class:`autoprotocol.instruction.FlowAnalyze` instruction created from the specified parameters Raises ------ TypeError If inputs are not of the correct type. UnitError If unit inputs are not properly formatted. AssertionError If required parameters are missing. ValueError If volumes are not correctly formatted or present. """ sources = [] controls = [] if not isinstance(samples, list): raise TypeError("Samples must be of type list.") else: sources.extend(samples) if not isinstance(neg_controls, list): raise TypeError("Neg_controls must be of type list.") else: sources.extend(neg_controls) controls.extend(neg_controls) if pos_controls and not isinstance(pos_controls, list): raise TypeError("Pos_controls must be of type list.") elif pos_controls: sources.extend(pos_controls) controls.extend(neg_controls) for s in sources: if not isinstance(s.get("well"), Well): raise TypeError( "The well for each sample or control must " "be of type Well." ) try: Unit(s.get("volume")) except (ValueError, TypeError) as e: raise ValueError( f"Each sample or control must indicate a " f"volume of type unit. {e}" ) from e if s.get("captured_events") and not isinstance( s.get("captured_events"), int ): raise TypeError( "captured_events is optional, if given it" " must be of type integer." ) for c in controls: if not isinstance(c.get("channel"), list): raise TypeError( "Channel must be a list of strings " "indicating the colors/channels that this" " control is to be used for." ) if c.get("minimize_bleed") and not isinstance( c.get("minimize_bleed"), list ): raise TypeError("Minimize_bleed must be of type list.") elif c.get("minimize_bleed"): for b in c["minimize_bleed"]: if not isinstance(b, dict): raise TypeError( "Minimize_bleed must be a list of " "dictonaries. Dictonary was not found" ) else: if not b.get("from"): raise ValueError( "Minimize_bleed dictonaries must" " have a key `from`" ) else: if not isinstance(b["from"], str): raise TypeError( "Minimize_bleed `from` must " "have a string as value" ) if not b.get("to"): raise ValueError( "Minimize_bleed dictonaries must" " have a key `to`" ) else: if not isinstance(b["to"], list): raise ValueError( "Minimize_bleed `to` must " "have a list as value" ) else: for t in b["to"]: if not isinstance(t, str): raise TypeError( "Minimize_bleed `to` " "list must contain " "strings." ) assert FSC and SSC, "You must include parameters for FSC and SSC " "channels." channels = {} channels["FSC"] = FSC channels["SSC"] = SSC for c in channels.values(): if not isinstance(c, dict): raise TypeError("Each channel must be of type dict.") assert c["voltage_range"], ( "You must include a voltage_range for" " each channel." ) assert c["voltage_range"]["high"], ( "You must include an upper " "limit for the volage range" "in each channel." ) assert c["voltage_range"]["low"], ( "You must include a lower " "limit for the volage range " "in each channel." ) if colors: if not isinstance(colors, list): raise TypeError("Colors must be of type list.") else: for c in colors: if not isinstance(c, dict): raise TypeError("Colors must contain elements of " "type dict.") else: if not c.get("name") or not isinstance(c.get("name"), str): raise TypeError( "Each color must have a `name` " "that is of type string." ) if c.get("emission_wavelength"): try: Unit(c.get("emission_wavelength")) except (UnitError) as e: raise UnitError( "Each `emission_wavelength` " "must be of type unit." ) from e else: raise ValueError( "Each color must have an " "`emission_wavelength`." ) if c.get("excitation_wavelength"): try: Unit(c.get("excitation_wavelength")) except (UnitError) as e: raise UnitError( "Each `excitation_wavelength` " "must be of type unit." ) from e else: raise ValueError( "Each color must have an " "`excitation_wavelength`." ) for s in sources: self._remove_cover(s["well"].container, "flow_analyze") return self._append_and_return( FlowAnalyze(dataref, FSC, SSC, neg_controls, samples, colors, pos_controls) )
[docs] def oligosynthesize(self, oligos: List[OligosynthesizeOligo]): """ Specify a list of oligonucleotides to be synthesized and a destination for each product. Example Usage: .. code-block:: python oligo_1 = p.ref("oligo_1", None, "micro-1.5", discard=True) p.oligosynthesize([{"sequence": "CATGGTCCCCTGCACAGG", "destination": oligo_1.well(0), "scale": "25nm", "purification": "standard"}]) Autoprotocol Output: .. code-block:: none "instructions": [ { "oligos": [ { "destination": "oligo_1/0", "sequence": "CATGGTCCCCTGCACAGG", "scale": "25nm", "purification": "standard" } ], "op": "oligosynthesize" } ] Parameters ---------- oligos : list(dict) List of oligonucleotides to synthesize. Each dictionary should contain the oligo's sequence, destination, scale and purification .. code-block:: none [ { "destination": "my_plate/A1", "sequence": "GATCRYMKSWHBVDN", // - standard IUPAC base codes // - IDT also allows rX (RNA), mX (2' O-methyl RNA), and // X*/rX*/mX* (phosphorothioated) // - they also allow inline annotations for // modifications, // e.g. "GCGACTC/3Phos/" for a 3' phosphorylation // e.g. "aggg/iAzideN/cgcgc" for an // internal modification "scale": "25nm" | "100nm" | "250nm" | "1um", "purification": "standard" | "page" | "hplc", // default: standard }, ... ] Returns ------- Oligosynthesize Returns the :py:class:`autoprotocol.instruction.Oligosynthesize` instruction created from the specified parameters """ return self._append_and_return( Oligosynthesize([OligosynthesizeOligo(**o) for o in oligos]) )
[docs] def autopick( self, pick_groups: List[AutopickGroup], criteria: Optional[Dict[str, Any]] = None, dataref: str = "autopick", ): """ Pick colonies from the agar-containing location(s) specified in `sources` to the location(s) specified in `dests` in highest to lowest rank order until there are no more colonies available. If fewer than min_abort pickable colonies have been identified from the location(s) specified in `sources`, the run will stop and no further instructions will be executed. Example Usage: Autoprotocol Output: Parameters ---------- pick_groups: List of AutopickGroups sources : Well or WellGroup or list(Well) or list(WellGroup) or list(list(Well)) Reference wells containing agar and colonies to pick destinations : Well or WellGroup or list(Well) or list(WellGroup) or list(list(Well)) List of destination(s) for picked colonies min_abort : int Total number of colonies that must be detected in the aggregate list of `from` wells to avoid aborting the entire run. Value of 0 prevents aborting regardless of amount detected. criteria : dict Dictionary of autopicking criteria. dataref: str Name of dataset to save the picked colonies to Returns ------- Autopick Returns the :py:class:`autoprotocol.instruction.Autopick` instruction created from the specified parameters Raises ------ TypeError Invalid input types for sources and dests ValueError Source wells are not all from the same container """ groups = [] for group in pick_groups: groups.append(self.__process_pick_group(group)) # Current device requirement is that all autopick group sources are from the same container if len(set([s.container for pick in groups for s in pick["from"]])) > 1: raise ValueError( "All source wells for autopick must exist " "on the same container" ) for group in groups: for s in group["from"]: self._remove_cover(s.container, "autopick") for d in group["to"]: self._remove_cover(d.container, "autopick") criteria = {} if criteria is None else criteria return self._append_and_return(Autopick(groups, criteria, dataref))
def __process_pick_group( self, pick_group: AutopickGroup ) -> Dict[str, Union[WellGroup, int]]: if not isinstance(pick_group, AutopickGroup): raise TypeError( "Autopick groups must use provided AutopickGroup dataclass." ) # Check valid well inputs if not is_valid_well(pick_group.source): raise TypeError( "Source must be of type Well, list of Wells, or " "WellGroup." ) if not is_valid_well(pick_group.destination): raise TypeError( "Destinations (dests) must be of type Well, " "list of Wells, or WellGroup." ) pick = {} pick["from"] = WellGroup(pick_group.source) pick["to"] = WellGroup(pick_group.destination) pick["min_abort"] = pick_group.min_abort return pick
[docs] def mag_dry( self, head: str, container: Container, duration: TIME, new_tip: bool = False, new_instruction: bool = False, ): """ Dry beads with magnetized tips above and outside a container for a set time. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("plate_0", None, "96-pcr", storage="cold_20") p.mag_dry("96-pcr", plate, "30:minute", new_tip=False, new_instruction=False) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ [ { "dry": { "duration": "30:minute", "object": "plate_0" } } ] ], "magnetic_head": "96-pcr", "op": "magnetic_transfer" } ] Parameters ---------- head : str Magnetic head to use for the magnetic bead transfers container : Container Container to dry beads above duration : str or Unit Time for drying new_tip : bool Specify whether to use a new tip to complete the step new_instruction: bool Specify whether to create a new magnetic_transfer instruction Returns ------- MagneticTransfer Returns the :py:class:`autoprotocol.instruction.MagneticTransfer` instruction created from the specified parameters """ mag = MagneticTransfer.builders.mag_dry(object=container, duration=duration) self._remove_cover(container, "mag_dry") return self._add_mag(mag, head, new_tip, new_instruction, "dry")
[docs] def mag_incubate( self, head: str, container: Container, duration: TIME, magnetize: bool = False, tip_position: float = 1.5, temperature: Optional[TEMPERATURE] = None, new_tip: bool = False, new_instruction: bool = False, ): """ Incubate the container for a set time with tips set at `tip_position`. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("plate_0", None, "96-pcr", storage="cold_20") p.mag_incubate("96-pcr", plate, "30:minute", magnetize=False, tip_position=1.5, temperature=None, new_tip=False) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ [ { "incubate": { "duration": "30:minute", "tip_position": 1.5, "object": "plate_0", "magnetize": false, "temperature": null } } ] ], "magnetic_head": "96-pcr", "op": "magnetic_transfer" } ] Parameters ---------- head : str Magnetic head to use for the magnetic bead transfers container : Container Container to incubate beads duration : str or Unit Time for incubation magnetize : bool Specify whether to magnetize the tips tip_position : float Position relative to well height that tips are held temperature: str or Unit, optional Temperature heat block is set at new_tip : bool Specify whether to use a new tip to complete the step new_instruction: bool Specify whether to create a new magnetic_transfer instruction Returns ------- MagneticTransfer Returns the :py:class:`autoprotocol.instruction.MagneticTransfer` instruction created from the specified parameters """ mag = MagneticTransfer.builders.mag_incubate( object=container, duration=duration, magnetize=magnetize, tip_position=tip_position, temperature=temperature, ) self._remove_cover(container, "mag_incubate") return self._add_mag(mag, head, new_tip, new_instruction, "incubate")
[docs] def mag_collect( self, head: str, container: Container, cycles: int, pause_duration: TIME, bottom_position: float = 0.0, temperature: TEMPERATURE = None, new_tip: bool = False, new_instruction: bool = False, ): """ Collect beads from a container by cycling magnetized tips in and out of the container with an optional pause at the bottom of the insertion. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("plate_0", None, "96-pcr", storage="cold_20") p.mag_collect("96-pcr", plate, 5, "30:second", bottom_position= 0.0, temperature=None, new_tip=False, new_instruction=False) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ [ { "collect": { "bottom_position": 0, "object": "plate_0", "temperature": null, "cycles": 5, "pause_duration": "30:second" } } ] ], "magnetic_head": "96-pcr", "op": "magnetic_transfer" } ] Parameters ---------- head : str Magnetic head to use for the magnetic bead transfers container : Container Container to incubate beads cycles: int Number of cycles to raise and lower tips pause_duration : str or Unit Time tips are paused in bottom position each cycle bottom_position : float Position relative to well height that tips are held during pause temperature: str or Unit Temperature heat block is set at new_tip : bool Specify whether to use a new tip to complete the step new_instruction: bool Specify whether to create a new magnetic_transfer instruction Returns ------- MagneticTransfer Returns the :py:class:`autoprotocol.instruction.MagneticTransfer` instruction created from the specified parameters """ mag = MagneticTransfer.builders.mag_collect( object=container, cycles=cycles, pause_duration=pause_duration, bottom_position=bottom_position, temperature=temperature, ) self._remove_cover(container, "mag_collect") return self._add_mag(mag, head, new_tip, new_instruction, "collect")
[docs] def mag_release( self, head: str, container: Container, duration: TIME, frequency: FREQUENCY, center: float = 0.5, amplitude: float = 0.5, temperature: Optional[TEMPERATURE] = None, new_tip: bool = False, new_instruction: bool = False, ): """ Release beads into a container by cycling tips in and out of the container with tips unmagnetized. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("plate_0", None, "96-pcr", storage="cold_20") p.mag_release("96-pcr", plate, "30:second", "60:hertz", center=0.75, amplitude=0.25, temperature=None, new_tip=False, new_instruction=False) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ [ { "release": { "center": 0.75, "object": "plate_0", "frequency": "2:hertz", "amplitude": 0.25, "duration": "30:second", "temperature": null } } ] ], "magnetic_head": "96-pcr", "op": "magnetic_transfer" } ] Parameters ---------- head : str Magnetic head to use for the magnetic bead transfers container : Container Container to incubate beads duration : str or Unit Total time for this sub-operation frequency : str or Unit Cycles per second (hertz) that tips are raised and lowered center : float Position relative to well height where oscillation is centered amplitude : float Distance relative to well height to oscillate around "center" temperature: str or Unit Temperature heat block is set at new_tip : bool Specify whether to use a new tip to complete the step new_instruction: bool Specify whether to create a new magnetic_transfer instruction Returns ------- MagneticTransfer Returns the :py:class:`autoprotocol.instruction.MagneticTransfer` instruction created from the specified parameters """ mag = MagneticTransfer.builders.mag_release( object=container, duration=duration, frequency=frequency, center=center, amplitude=amplitude, temperature=temperature, ) self._remove_cover(container, "mag_release") return self._add_mag(mag, head, new_tip, new_instruction, "release")
[docs] def mag_mix( self, head: str, container: Container, duration: TIME, frequency: FREQUENCY, center: float = 0.5, amplitude: float = 0.5, magnetize: bool = False, temperature: Optional[TEMPERATURE] = None, new_tip: bool = False, new_instruction: bool = False, ): """ Mix beads in a container by cycling tips in and out of the container. Example Usage: .. code-block:: python p = Protocol() plate = p.ref("plate_0", None, "96-pcr", storage="cold_20") p.mag_mix("96-pcr", plate, "30:second", "60:hertz", center=0.75, amplitude=0.25, magnetize=True, temperature=None, new_tip=False, new_instruction=False) Autoprotocol Output: .. code-block:: none "instructions": [ { "groups": [ [ { "mix": { "center": 0.75, "object": "plate_0", "frequency": "2:hertz", "amplitude": 0.25, "duration": "30:second", "magnetize": true, "temperature": null } } ] ], "magnetic_head": "96-pcr", "op": "magnetic_transfer" } ] Parameters ---------- head : str Magnetic head to use for the magnetic bead transfers container : Container Container to incubate beads duration : str or Unit Total time for this sub-operation frequency : str or Unit Cycles per second (hertz) that tips are raised and lowered center : float Position relative to well height where oscillation is centered amplitude : float Distance relative to well height to oscillate around "center" magnetize : bool Specify whether to magnetize the tips temperature: str or Unit Temperature heat block is set at new_tip : bool Specify whether to use a new tip to complete the step new_instruction: bool Specify whether to create a new magnetic_transfer instruction Returns ------- MagneticTransfer Returns the :py:class:`autoprotocol.instruction.MagneticTransfer` instruction created from the specified parameters """ mag = MagneticTransfer.builders.mag_mix( object=container, duration=duration, frequency=frequency, center=center, amplitude=amplitude, magnetize=magnetize, temperature=temperature, ) self._remove_cover(container, "mag_mix") return self._add_mag(mag, head, new_tip, new_instruction, "mix")
[docs] def image_plate(self, ref: Union[str, Container], mode: str, dataref: str): """ Capture an image of the specified container. Example Usage: .. code-block:: python p = Protocol() agar_plate = p.ref("agar_plate", None, "1-flat", discard=True) bact = p.ref("bacteria", None, "micro-1.5", discard=True) p.spread(bact.well(0), agar_plate.well(0), "55:microliter") p.incubate(agar_plate, "warm_37", "18:hour") p.image_plate(agar_plate, mode="top", dataref="my_plate_image_1") Autoprotocol Output: .. code-block:: json { "refs": { "bacteria": { "new": "micro-1.5", "discard": true }, "agar_plate": { "new": "1-flat", "discard": true } }, "instructions": [ { "volume": "55.0:microliter", "to": "agar_plate/0", "from": "bacteria/0", "op": "spread" }, { "where": "warm_37", "object": "agar_plate", "co2_percent": 0, "duration": "18:hour", "shaking": false, "op": "incubate" }, { "dataref": "my_plate_image_1", "object": "agar_plate", "mode": "top", "op": "image_plate" } ] } Parameters ---------- ref : str or Container Container to take image of mode : str Imaging mode (currently supported: "top") dataref : str Name of data reference of resulting image Returns ------- ImagePlate Returns the :py:class:`autoprotocol.instruction.ImagePlate` instruction created from the specified parameters """ return self._append_and_return(ImagePlate(ref, mode, dataref))
[docs] def provision( self, resource_id: str, dests: WellParam, amounts: Optional[ Union[AMOUNT_CONCENTRATION, List[AMOUNT_CONCENTRATION]] ] = None, volumes: Optional[Union[VOLUME, List[VOLUME]]] = None, informatics: Optional[List[Informatics]] = None, ): """ Provision a commercial resource from a catalog into the specified destination well(s). A new tip is used for each destination well specified to avoid contamination. Parameters ---------- resource_id : str Resource ID from catalog. dests : Well or WellGroup or list(Well) Destination(s) for specified resource. amounts : str or Unit or list(str) or list(Unit) Volume(s) or Mass(es) to transfer of the resource to each destination well. If one volume or mass is specified, each destination well receive that volume or mass of the resource. If destinations should receive different volume or mass, each one should be specified explicitly in a list matching the order of the specified destinations. Note: Volumes and amounts arguments are mutually exclusive. Only one is required volumes : str or Unit or list(str) or list(Unit) Volume to transfer of the resource to each destination well. If one volume is specified, each destination well receive that volume of the resource. If destinations should receive different volumes, each one should be specified explicitly in a list matching the order of the specified destinations. Note: Volumes and amounts arguments are mutually exclusive. Only one is required informatics: list(Informatics) List of Informatics detailing aliquot effects intended from this instruction. Raises ------ TypeError If resource_id is not a string. TypeError If the unit provided is not supported TypeError If volume or mass is not specified as a string or Unit (or a list of either) RuntimeError If length of the list of volumes or masses specified does not match the number of destination wells specified. ValueError If the resource measurement mode is volume and the provision exceeds max capacity of well. ValueError If the provisioning of volumes or amounts are not supported. Returns ------- list(Provision) :py:class:`autoprotocol.instruction.Provision` instruction object(s) to be appended and returned """ # Check valid well inputs if not is_valid_well(dests): raise TypeError( "Destinations (dests) must be of type Well, " "list of Wells, or WellGroup." ) dests = WellGroup(dests) if not isinstance(resource_id, str): raise TypeError("Resource ID must be a string.") if (volumes is None and amounts is None) or (volumes and amounts): raise ValueError( "Either volumes or amounts should have value(s), but not both." ) if volumes: provision_amounts = volumes else: provision_amounts = amounts if not isinstance(provision_amounts, list): provision_amounts = [Unit(provision_amounts)] * len(dests) else: if len(provision_amounts) != len(dests): raise RuntimeError( "To provision a resource into multiple " "destinations with multiple volumes or masses, the " "list of volumes or masses must correspond with the " "destinations in length and in order." ) provision_amounts = [ parse_unit(v, ["liter", "gram"]) for v in provision_amounts ] measurement_mode = self._identify_provision_mode(provision_amounts) provision_instructions_to_return: Provision = [] for d, amount in zip(dests, provision_amounts): if d.container.is_covered() or d.container.is_sealed(): self._remove_cover(d.container, "provision") xfer = {"well": d, measurement_mode: amount} if measurement_mode == "volume": d_max_vol = d.container.container_type.true_max_vol_ul if amount > d_max_vol: raise ValueError( f"The volume you are trying to provision ({amount}) exceeds the " f"maximum capacity of this well ({d_max_vol})." ) if amount > Unit(900, "microliter"): diff = amount - Unit(900, "microliter") provision_instructions_to_return.append( self.provision( resource_id, d, Unit(900, "microliter"), informatics=informatics, ) ) while diff > Unit(0.0, "microliter"): provision_instructions_to_return.append( self.provision(resource_id, d, diff) ) diff -= diff continue if d.volume: d.volume += amount else: d.set_volume(amount) dest_group = [xfer] if ( self.instructions and self.instructions[-1].op == "provision" and self.instructions[-1].resource_id == resource_id and self.instructions[-1].to[-1]["well"].container == d.container ): if informatics is not None: self.instructions[-1].informatics.extend(informatics) self.instructions[-1].to.append(xfer) else: provision_instructions_to_return.append( self._append_and_return( Provision( resource_id, dest_group, measurement_mode, informatics ) ) ) return provision_instructions_to_return
def _identify_provision_mode(self, provision_amounts: List[Unit]): unique_measure_modes = set() for amount in provision_amounts: if not isinstance(amount, Unit): raise TypeError(f"Provided amount {amount} is not supported.") if amount.dimensionality == Unit(1, "liter").dimensionality: unique_measure_modes.add("volume") elif amount.dimensionality == Unit(1, "gram").dimensionality: unique_measure_modes.add("mass") else: raise ValueError( f"Provisioning of resources with measurement unit of {amount.unit} is not supported." ) if len(unique_measure_modes) != 1: raise ValueError("Received amounts with more than one measurement mode") measurement_mode = unique_measure_modes.pop() return measurement_mode
[docs] def flash_freeze(self, container: Union[str, Container], duration: TIME): """ Flash freeze the contents of the specified container by submerging it in liquid nitrogen for the specified amount of time. Example Usage: .. code-block:: python p = Protocol() sample = p.ref("liquid_sample", None, "micro-1.5", discard=True) p.flash_freeze(sample, "25:second") Autoprotocol Output: .. code-block:: json { "refs": { "liquid_sample": { "new": "micro-1.5", "discard": true } }, "instructions": [ { "duration": "25:second", "object": "liquid_sample", "op": "flash_freeze" } ] } Parameters ---------- container : Container or str Container to be flash frozen. duration : str or Unit Duration to submerge specified container in liquid nitrogen. Returns ------- FlashFreeze Returns the :py:class:`autoprotocol.instruction.FlashFreeze` instruction created from the specified parameters """ return self._append_and_return(FlashFreeze(container, duration))
[docs] def sonicate( self, wells: WellParam, duration: TIME, mode: SonicateMode, mode_params: Union[SonicateModeParamsBath, SonicateModeParamsHorn], frequency: Optional[FREQUENCY] = None, temperature: Optional[TEMPERATURE] = None, ): """ Sonicate wells using high intensity ultrasonic vibrations. Example Usage: .. code-block:: python p = Protocol() sample_wells = p.ref("sample_plate", None, "96-pcr", storage="warm_37").wells_from(0,2) p.sonicate(sample_wells, duration="1:minute", mode="bath", mode_params={"sample_holder": "suspender"}) Autoprotocol Output: .. code-block:: json { "op": "sonicate", "wells": ["sample_plate/0", "sample_plate/1"], "mode": "bath", "duration": "1:minute", "temperature": "ambient", "frequency": "40:kilohertz" "mode_params": { "sample_holder": "suspender" } } Parameters ---------- wells : Well or WellGroup or list(Well) Wells to be sonicated duration : Unit or str Duration for which to sonicate wells mode: Enum({"bath", "horn"}) Sonicating method to be used, must be "horn" or "bath". Sonicate mode "horn" uses metal probe to create a localized shear force directly in the sample media; "bath" mode applies ultrasound to wells held inside a bath. temperature: Unit or str, optional Temperature at which the sample is kept during sonication. Optional, defaults to ambient frequency: Unit or str, optional Frequency of the ultrasonic wave, usually indicated in kHz. Optional; defaults to the most commonly used frequency for each mode: 20 kHz for `horn`, and 40 kHz for `bath` mode mode_params: Dict Dictionary containing mode parameters for the specified mode. .. code-block:: none { "mode": "bath", "mode_params": { "sample_holder": Enum({"suspender", "perforated_container", "solid_container"}) "power": Unit or str, optional } } - or - { "mode": "horn", "mode_params": { "duty_cycle": Float, 0 < value <=1 "amplitude": Unit or str } } Returns ------- Sonicate Returns the :py:class:`autoprotocol.instruction.Sonicate` instruction created from the specified parameters Raises ------ RuntimeError If valid mode is not specified. TypeError If wells not of type WellGroup or List of Wells. ValueError If invalid `mode_params` for specified mode. TypeError If invalid `mode_params` type for specified mode. """ sonic_modes = ["bath", "horn"] if mode not in sonic_modes: raise RuntimeError(f"{mode} is not a valid sonication mode") if not isinstance(mode_params, dict): raise TypeError(f"Invalid mode_params {mode_params}, must be a dict") parsed_mode_params = {} if mode == "bath": valid_sample_holders = [ "suspender", "perforated_container", "solid_container", ] valid_mode_params = ["sample_holder", "power"] sample_holder = mode_params.get("sample_holder") if sample_holder not in valid_sample_holders: raise ValueError( f"'sample_holder' must be specified in 'mode_params' mode: " f"{mode} and must be one of {valid_sample_holders}." ) parsed_mode_params["sample_holder"] = sample_holder power = mode_params.get("power") if power: parsed_power = parse_unit(power, "power-watt") parsed_mode_params["power"] = parsed_power frequency = frequency or "40:kilohertz" if mode == "horn": valid_mode_params = ["duty_cycle", "amplitude"] if not all(k in mode_params for k in valid_mode_params): raise ValueError( f"Incorrect mode_params. All of " f"{valid_mode_params} must be included in " f"mode: {mode}" ) duty_cycle = mode_params["duty_cycle"] amplitude = mode_params["amplitude"] if not isinstance(duty_cycle, (int, float)): raise TypeError(f"Invalid duty_cycle {duty_cycle}, must be a decimal") duty_cycle = float(duty_cycle) if not 0 <= duty_cycle <= 1: raise ValueError( f"Invalid duty_cycle {duty_cycle}, must be between 0 and " f"1 (inclusive)." ) parsed_mode_params["duty_cycle"] = duty_cycle parsed_amplitude = parse_unit(amplitude, "micrometer") parsed_mode_params["amplitude"] = parsed_amplitude frequency = frequency or "20:kilohertz" if not is_valid_well(wells): raise TypeError("Wells must be of type Well, list of Wells, or WellGroup.") wells = WellGroup(wells) parsed_duration = parse_unit(duration, "seconds") parsed_frequency = parse_unit(frequency, "hertz") if temperature: parsed_temperature = parse_unit(temperature, "celsius") else: parsed_temperature = None return self._append_and_return( Sonicate( wells, parsed_duration, mode, parsed_mode_params, parsed_frequency, parsed_temperature, ) )
[docs] def spe( self, well: Well, cartridge: str, pressure_mode: str, load_sample: SpeLoadSample, elute: List[SpeElute], condition: Optional[List[SpeParams]] = None, equilibrate: Optional[List[SpeParams]] = None, rinse: Optional[List[SpeParams]] = None, ): """ Apply a solid phase extraction (spe) technique to a sample. Example Usage: .. code-block:: python p = Protocol() elute_params = [ SPE.builders.mobile_phase_params( is_elute=True, volume="2:microliter", loading_flowrate="100:ul/second", settle_time="2:minute", processing_time="3:minute", flow_pressure="2:bar", resource_id="solvent_a", destination_well=p.ref("Elute %s" % i, None, "micro-1.5", discard=True).well(0)) for i in range(3) ] sample_loading_params = SPE.builders.mobile_phase_params( volume="10:microliter", loading_flowrate="1:ul/second", settle_time="2:minute", processing_time="3:minute", flow_pressure="2:bar", is_sample=True) cartridge = "spe_cartridge" sample = p.ref("Sample", None, "micro-1.5", discard=True).well(0) p.spe(sample, cartridge, "positive", load_sample=sample_loading_params, elute=elute_params) Autoprotocol Output: .. code-block:: none "instructions": [ { "op": "spe", "elute": [ { "loading_flowrate": "100:microliter/second", "resource_id": "solvent_a", "settle_time": "2:minute", "volume": "2:microliter", "flow_pressure": "2:bar", "destination_well": "Elute 0/0", "processing_time": "3:minute" }, { "loading_flowrate": "100:microliter/second", "resource_id": "solvent_a", "settle_time": "2:minute", "volume": "2:microliter", "flow_pressure": "2:bar", "destination_well": "Elute 1/0", "processing_time": "3:minute" }, { "loading_flowrate": "100:microliter/second", "resource_id": "solvent_a", "settle_time": "2:minute", "volume": "2:microliter", "flow_pressure": "2:bar", "destination_well": "Elute 2/0", "processing_time": "3:minute" } ], "cartridge": "spe_cartridge", "well": "Sample/0", "load_sample": { "flow_pressure": "2:bar", "loading_flowrate": "1:microliter/second", "settle_time": "2:minute", "processing_time": "3:minute", "volume": "10:microliter" }, "pressure_mode": "positive" } ] Parameters ---------- well : Well Well to solid phase extract. cartridge : str Cartridge to use for solid phase extraction. pressure_mode : str The direction of pressure applied to the cartridge to force liquid flow. One of "positive", "negative". load_sample: dict Parameters for applying the sample to the cartridge. Single 'mobile_phase_param'. elute: list(dict) Parameters for applying a mobile phase to the cartridge with one or more solvents. List of 'mobile_phase_params'. Requires `destination_well`. condition: list(dict), optional Parameters for applying a mobile phase to the cartridge with one or more solvents. List of 'mobile_phase_params'. equilibrate: list(dict), optional Parameters for applying a mobile phase to the cartridge with one or more solvents. List of 'mobile_phase_params'. rinse: list(dict), optional Parameters for applying a mobile phase to the cartridge with one or more solvents. List of 'mobile_phase_params'. mobile_phase_params: resource_id: str Resource ID of desired solvent. volume: volume Volume added to the cartridge. loading_flowrate: Unit Speed at which volume is added to cartridge. settle_time: Unit Duration for which the solvent remains on the cartridge before a pressure mode is applied. processing_time: Unit Duration for which pressure is applied to the cartridge after `settle_time` has elapsed. flow_pressure: Unit Pressure applied to the column. destination_well: Well Destination well for eluate. Required parameter for each `elute` mobile phase parameter Returns ------- SPE Returns the :py:class:`autoprotocol.instruction.SPE` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. well given is not of type Well ValueError Wells specified are not from the same container ValueError Invalid pressure_mode ValueError settle_time, processing_time, flow_pressure not greater than 0 ValueError If not exactly one elution parameter for each elution container UnitError Improperly formatted units for mobile phase parameters """ if not is_valid_well(well): raise TypeError("Well must be of type Well.") if not isinstance(cartridge, str): raise TypeError("Cartrige must be of type string.") valid_pressure_modes = ["positive", "negative"] if pressure_mode not in valid_pressure_modes: raise ValueError( f"'pressure_mode': {pressure_mode} has to be one " f"of {valid_pressure_modes}" ) load_sample = SPE.builders.mobile_phase_params(is_sample=True, **load_sample) SPE.builders.spe_params(elute, is_elute=True) if condition: SPE.builders.spe_params(condition) if equilibrate: SPE.builders.spe_params(equilibrate) if rinse: SPE.builders.spe_params(rinse) return self._append_and_return( SPE( well, cartridge, pressure_mode, load_sample, elute, condition, equilibrate, rinse, ) )
[docs] def image( self, ref: Container, mode: Union[str, ImageMode], num_images: int, dataref: str, backlighting: Optional[bool] = None, magnification: float = 1.0, exposure: Optional[ImageExposure] = None, ): """ Capture an image of the specified container. Example Usage: .. code-block:: python p = Protocol() sample = p.ref("Sample", None, "micro-1.5", discard=True) p.image(sample, "top", "image_1", num_images=3, backlighting=False, exposure={"iso": 4}, magnification=1.0) Autoprotocol Output: .. code-block:: json { "refs": { "Sample": { "new": "micro-1.5", "discard": true } }, "instructions": [ { "magnification": 1.0, "backlighting": false, "mode": "top", "dataref": "image_1", "object": "Sample", "num_images": 3, "op": "image", "exposure": { "iso": 4 } } ] } Parameters ---------- ref : Container Container of which to take image. mode : Enum("top", "bottom", "side") Angle of image. num_images : int Number of images taken of the container. Defaults to 1. dataref : str Name of data reference of resulting image backlighting : Bool, optional Whether back-lighting is desired. magnification : float Ratio of sizes of the image projected on the camera sensor compared to the actual size of the object captured. Defaults to 1.0. exposure : dict, optional Parameters to control exposure: "aperture", "iso", and "shutter_speed". shutter_speed: Unit, optional Duration that the imaging sensor is exposed. iso : Float, optional Light sensitivity of the imaging sensor. aperture: Float, optional Diameter of the lens opening. Returns ------- Image Returns the :py:class:`autoprotocol.instruction.Image` instruction created from the specified parameters Raises ------ TypeError Invalid input types, e.g. num_images is not a positive integer ValueError Invalid exposure parameter supplied """ 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: raise TypeError("num_images must be a positive integer.") if magnification <= 0: raise TypeError("magnification must be a number.") return self._append_and_return( Image(ref, mode, dataref, num_images, backlighting, exposure, magnification) )
def _ref_for_well(self, well: Well): return f"{self._ref_for_container(well.container)}/{well.index}" def _ref_for_container(self, container: Container): for k in self.refs: v = self.refs[k] if v.container is container: return k def _remove_cover(self, container: Container, action: str): if not container.container_type.is_tube: if not (container.is_covered() or container.is_sealed()): return elif container.cover in COVER_TYPES: self.uncover(container) elif container.cover in SEAL_TYPES: self.unseal(container) else: raise RuntimeError( f"The operation {action} requires an uncovered container, " f"however, {container.cover} is not a recognized cover or " f"seal type." ) def _add_cover(self, container: Container, action: str): if not container.container_type.is_tube: if container.is_covered() or container.is_sealed(): return elif "cover" in container.container_type.capabilities: self.cover(container, container.container_type.cover_types[0]) elif "seal" in container.container_type.capabilities: self.seal(container, container.container_type.seal_types[0]) else: raise RuntimeError( f"The operation {action} requires a covered container, " f"however, {container.container_type.name} does not have " f"a recognized cover or seal type." ) def _add_seal(self, container: Container, action: str): if not container.container_type.is_tube: if container.is_sealed(): return elif container.is_covered(): raise RuntimeError( f"The operation {action} requires a sealed container, " f"however, {container.name} currently hasa lid which needs " f"to be first removed." ) if "seal" in container.container_type.capabilities: self.seal(container, container.container_type.seal_types[0]) else: raise RuntimeError( f"The operation {action} requires a sealed container, " f"however, {container.container_type.name} does not have " f"a recognized seal type." ) def _add_mag(self, sub_op, head, new_tip, new_instruction, sub_op_name): """ Append given magnetic_transfer groups to protocol """ last_instruction = self.instructions[-1] if self.instructions else None maybe_same_instruction = ( new_instruction is False and last_instruction and isinstance(last_instruction, MagneticTransfer) and last_instruction.data.get("magnetic_head") == head ) # Overwriting __dict__ since that's edited on __init__ and we use it # for downstream checks if maybe_same_instruction and new_tip is True: new_groups = last_instruction.data.get("groups") new_groups.append([{sub_op_name: sub_op}]) last_instruction.__dict__ = MagneticTransfer( groups=new_groups, magnetic_head=head ).__dict__ return last_instruction elif maybe_same_instruction and new_tip is False: new_groups = last_instruction.data.get("groups") new_groups[-1].append({sub_op_name: sub_op}) last_instruction.__dict__ = MagneticTransfer( groups=new_groups, magnetic_head=head ).__dict__ return last_instruction else: return self._append_and_return( MagneticTransfer(groups=[[{sub_op_name: sub_op}]], magnetic_head=head) ) # pylint: disable=protected-access def _refify( self, op_data: Union[ Dict[str, Any], List[Any], Well, WellGroup, Container, Unit, Instruction, Ref, Compound, Informatics, ], ): """ Unpacks protocol objects into Autoprotocol compliant ones Used by as_dict(). Parameters ---------- op_data: any protocol object Returns ------- dict or str or list or any Autoprotocol compliant objects """ if type(op_data) is dict: return {k: self._refify(v) for k, v in op_data.items()} elif type(op_data) is list: return [self._refify(i) for i in op_data] elif isinstance(op_data, Well): return self._ref_for_well(op_data) elif isinstance(op_data, WellGroup): return [self._ref_for_well(w) for w in op_data.wells] elif isinstance(op_data, Container): return self._ref_for_container(op_data) elif isinstance(op_data, Unit): return str(op_data) elif isinstance(op_data, Instruction): return self._refify(op_data._as_AST()) elif isinstance(op_data, Ref): return op_data.opts.as_dict() elif isinstance(op_data, Compound): return op_data.as_dict() elif isinstance(op_data, Informatics): return self._refify(op_data.as_dict()) else: return op_data def _ref_containers_and_wells(self, params: Dict[Any, Any]): """ Used by harness.run() to process JSON container and well references .. code-block:: python parameters = { "sample": { "id": null, "type": "micro-1.5", "storage": "cold_4", "discard": null }, "mastermix_loc": "sample_plate/A1", "samples": [ "sample_plate/B1", "sample_plate/B2", "sample_plate/B3", "sample_plate/B4" ] } protocol.make_well_references(parameters) returns: .. code-block:: python { "refs":{ "sample": Container(None, "micro-1.5") }, "mastermix_loc": protocol.refs["sample_plate"].well("A1"), "samples": WellGroup([ protocol.refs["sample_plate"].well("B1"), protocol.refs["sample_plate"].well("B2"), protocol.refs["sample_plate"].well("B3"), protocol.refs["sample_plate"].well("B4") ]) } Parameters ---------- params : dict A dictionary of parameters to be passed to a protocol. Returns ------- dict Dictionary of containers and wells Raises ------ RuntimeError Invalid parameters """ parameters = {} containers = {} # ref containers for k, v in params.items(): if isinstance(v, dict): parameters[str(k)] = self._ref_containers_and_wells(v) if isinstance(v, list) and isinstance(v[0], dict): for cont in v: self._ref_containers_and_wells(cont.encode("utf-8")) elif isinstance(v, dict) and "type" in v: if "discard" in v: discard = v["discard"] if discard and v.get("storage"): raise RuntimeError( "You must either specify a storage " "condition or set discard to true, " "not both." ) else: discard = False containers[str(k)] = self.ref( k, v["id"], v["type"], storage=v.get("storage"), discard=discard ) else: parameters[str(k)] = v parameters["refs"] = containers # ref wells (must be done after reffing containers) for k, v in params.items(): if isinstance(v, list) and "/" in str(v[0]): group = WellGroup([]) for w in v: cont, well = w.rsplit("/", 1) group.append(self.refs[cont].container.well(well)) parameters[str(k)] = group elif "/" in str(v): ref_name = v.rsplit("/")[0] if ref_name not in self.refs: raise RuntimeError( f"Parameters contain well references to a container that isn't referenced in this protocol: '{ref_name}'." ) if v.rsplit("/")[1] == "all_wells": parameters[str(k)] = self.refs[ref_name].container.all_wells() else: parameters[str(k)] = self.refs[ref_name].container.well( v.rsplit("/")[1] ) else: parameters[str(k)] = v return parameters
[docs] def measure_concentration( self, wells: WellParam, dataref: str, measurement: str, volume: str = "2:microliter", ): """ Measure the concentration of DNA, ssDNA, RNA or protein in the specified volume of the source aliquots. Example Usage: .. code-block:: python p = Protocol() test_plate = p.ref("test_plate", id=None, cont_type="96-flat", storage=None, discard=True) p.measure_concentration(test_plate.wells_from(0, 3), "mc_test", "DNA") p.measure_concentration(test_plate.wells_from(3, 3), dataref="mc_test2", measurement="protein", volume="4:microliter") Autoprotocol Output: .. code-block:: none { "refs": { "test_plate": { "new": "96-flat", "discard": true } }, "instructions": [ { "volume": "2.0:microliter", "dataref": "mc_test", "object": [ "test_plate/0", "test_plate/1", "test_plate/2" ], "op": "measure_concentration", "measurement": "DNA" }, ... ] } Parameters ---------- wells : list(Well) or WellGroup or Well WellGroup of wells to be measured volume : str or Unit Volume of sample required for analysis dataref : str Name of this specific dataset of measurements measurement : str Class of material to be measured. One of ["DNA", "ssDNA", "RNA", "protein"]. Returns ------- MeasureConcentration Returns the :py:class:`autoprotocol.instruction.MeasureConcentration` instruction created from the specified parameters Raises ------ TypeError `wells` specified is not of a valid input type """ if not is_valid_well(wells): raise TypeError("Wells must be of type Well, list of Wells, or WellGroup.") wells = WellGroup(wells) return self._append_and_return( MeasureConcentration(wells, volume, dataref, measurement) )
[docs] def measure_mass(self, container: Container, dataref: str): """ Measure the mass of a container. Example Usage: .. code-block:: python p = Protocol() test_plate = p.ref("test_plate", id=None, cont_type="96-flat", storage=None, discard=True) p.measure_mass(test_plate, "test_data") Autoprotocol Output: .. code-block:: json { "refs": { "test_plate": { "new": "96-flat", "discard": true } }, "instructions": [ { "dataref": "test_data", "object": [ "test_plate" ], "op": "measure_mass" } ] } Parameters ---------- container : Container container to be measured dataref : str Name of this specific dataset of measurements Returns ------- MeasureMass Returns the :py:class:`autoprotocol.instruction.MeasureMass` instruction created from the specified parameters Raises ------ TypeError Input given is not of type Container """ if not isinstance(container, Container): raise TypeError(f"{container} has to be of type Container") return self._append_and_return(MeasureMass(container, dataref))
[docs] def measure_volume(self, wells: WellParam, dataref: str): """ Measure the volume of each well in wells. Example Usage: .. code-block:: python p = Protocol() test_plate = p.ref("test_plate", id=None, cont_type="96-flat", storage=None, discard=True) p.measure_volume(test_plate.from_wells(0,2), "test_data") Autoprotocol Output: .. code-block:: json { "refs": { "test_plate": { "new": "96-flat", "discard": true } }, "instructions": [ { "dataref": "test_data", "object": [ "test_plate/0", "test_plate/1" ], "op": "measure_volume" } ] } Parameters ---------- wells : list(Well) or WellGroup or Well list of wells to be measured dataref : str Name of this specific dataset of measurements Returns ------- MeasureVolume Returns the :py:class:`autoprotocol.instruction.MeasureVolume` instruction created from the specified parameters Raises ------ TypeError `wells` specified is not of a valid input type """ if not is_valid_well(wells): raise TypeError("Wells must be of type Well, list of Wells, or WellGroup.") wells = WellGroup(wells) return self._append_and_return(MeasureVolume(wells, dataref))
[docs] def count_cells( self, wells: WellParam, volume: VOLUME, dataref: str, labels: Optional[List[str]] = None, ): """ Count the number of cells in a sample that are positive/negative for a given set of labels. Example Usage: .. code-block:: python p = Protocol() cell_suspension = p.ref( "cells_with_trypan_blue", id=None, cont_type="micro-1.5", discard=True ) p.count_cells( cell_suspension.well(0), "10:microliter", "my_cell_count", ["trypan_blue"] ) Autoprotocol Output: .. code-block:: json { "refs": { "cells_with_trypan_blue": { "new": "micro-1.5", "discard": true } }, "instructions": [ { "dataref": "my_cell_count", "volume": "10:microliter", "wells": [ "cells_with_trypan_blue/0" ], "labels": [ "trypan_blue" ], "op": "count_cells" } ] } Parameters ---------- wells: Well or list(Well) or WellGroup List of wells that will be used for cell counting. volume: Unit Volume that should be consumed from each well for the purpose of cell counting. dataref: str Name of dataset that will be returned. labels: list(string), optional Cells will be scored for presence or absence of each label in this list. If staining is required to visualize these labels, they must be added before execution of this instruction. Returns ------- CountCells Returns the :py:class:`autoprotocol.instruction.CountCells` instruction created from the specified parameters Raises ------ TypeError `wells` specified is not of a valid input type """ # Check valid well inputs if not is_valid_well(wells): raise TypeError("Wells must be of type Well, list of Wells, or WellGroup.") wells = WellGroup(wells) # Parse volume parsed_volume = parse_unit(volume, "microliter") # Eliminate duplicates from labels parsed_labels = list(set(labels)) return self._append_and_return( CountCells(wells, parsed_volume, dataref, parsed_labels) )
[docs] def spectrophotometry( self, dataref: str, obj: Union[Container, str], groups: List, interval: Optional[TIME] = None, num_intervals: Optional[int] = None, temperature: Optional[TEMPERATURE] = None, shake_before: Optional[SpectrophotometryShakeBefore] = None, ): """ Generates an instruction with one or more plate reading steps executed on a single plate with the same device. This could be executed once, or at a defined interval, across some total duration. Example Usage: .. code-block:: python p = Protocol() read_plate = p.ref("read plate", cont_type="96-flat", discard=True) groups = Spectrophotometry.builders.groups( [ Spectrophotometry.builders.group( "absorbance", Spectrophotometry.builders.absorbance_mode_params( wells=read_plate.wells(0, 1), wavelength=["100:nanometer", "200:nanometer"], num_flashes=15, settle_time="1:second" ) ), Spectrophotometry.builders.group( "fluorescence", Spectrophotometry.builders.fluorescence_mode_params( wells=read_plate.wells(0, 1), excitation=[ Spectrophotometry.builders.wavelength_selection( ideal="650:nanometer" ) ], emission=[ Spectrophotometry.builders.wavelength_selection( shortpass="600:nanometer", longpass="700:nanometer" ) ], num_flashes=15, settle_time="1:second", lag_time="9:second", integration_time="2:second", gain=0.3, read_position="top" ) ), Spectrophotometry.builders.group( "luminescence", Spectrophotometry.builders.luminescence_mode_params( wells=read_plate.wells(0, 1), num_flashes=15, settle_time="1:second", integration_time="2:second", gain=0.3 ) ), Spectrophotometry.builders.group( "shake", Spectrophotometry.builders.shake_mode_params( duration="1:second", frequency="9:hertz", path="ccw_orbital", amplitude="1:mm" ) ), ] ) shake_before = Spectrophotometry.builders.shake_before( duration="10:minute", frequency="5:hertz", path="ccw_orbital", amplitude="1:mm" ) p.spectrophotometry( dataref="test data", obj=read_plate, groups=groups, interval="10:minute", num_intervals=2, temperature="37:celsius", shake_before=shake_before ) Autoprotocol Output: .. code-block:: json { "op": "spectrophotometry", "dataref": "test data", "object": "read plate", "groups": [ { "mode": "absorbance", "mode_params": { "wells": [ "read plate/0", "read plate/1" ], "wavelength": [ "100:nanometer", "200:nanometer" ], "num_flashes": 15, "settle_time": "1:second" } }, { "mode": "fluorescence", "mode_params": { "wells": [ "read plate/0", "read plate/1" ], "excitation": [ { "ideal": "650:nanometer" } ], "emission": [ { "shortpass": "600:nanometer", "longpass": "700:nanometer" } ], "num_flashes": 15, "settle_time": "1:second", "lag_time": "9:second", "integration_time": "2:second", "gain": 0.3, "read_position": "top" } }, { "mode": "luminescence", "mode_params": { "wells": [ "read plate/0", "read plate/1" ], "num_flashes": 15, "settle_time": "1:second", "integration_time": "2:second", "gain": 0.3 } }, { "mode": "shake", "mode_params": { "duration": "1:second", "frequency": "9:hertz", "path": "ccw_orbital", "amplitude": "1:millimeter" } } ], "interval": "10:minute", "num_intervals": 2, "temperature": "37:celsius", "shake_before": { "duration": "10:minute", "frequency": "5:hertz", "path": "ccw_orbital", "amplitude": "1:millimeter" } } Parameters ---------- dataref : str Name of the resultant dataset to be returned. obj : Container or str Container to be read. groups : list A list of groups generated by SpectrophotometryBuilders groups builders, any of absorbance_mode_params, fluorescence_mode_params, luminescence_mode_params, or shake_mode_params. interval : Unit or str, optional The time between each of the read intervals. num_intervals : int, optional The number of times that the groups should be executed. temperature : Unit or str, optional The temperature that the entire instruction should be executed at. shake_before : dict, optional A dict of params generated by SpectrophotometryBuilders.shake_before that dictates how the obj should be incubated with shaking before any of the groups are executed. Returns ------- Spectrophotometry Returns the :py:class:`autoprotocol.instruction.Spectrophotometry` instruction created from the specified parameters Raises ------ TypeError Invalid num_intervals specified, must be an int ValueError No interval specified but shake groups specified with no duration """ groups = Spectrophotometry.builders.groups(groups) if interval is not None: interval = parse_unit(interval, "seconds") if num_intervals and not isinstance(num_intervals, int): raise TypeError(f"Invalid num_intervals {num_intervals}, must be an int.") if temperature is not None: temperature = parse_unit(temperature, "celsius") if shake_before is not None: shake_before = Spectrophotometry.builders.shake_before( duration=shake_before.duration, frequency=shake_before.frequency, path=shake_before.path, amplitude=shake_before.amplitude, ) shake_groups = [_ for _ in groups if _["mode"] == "shake"] any_shake_duration_undefined = any( _["mode_params"].get("duration") is None for _ in shake_groups ) if any_shake_duration_undefined and interval is None: raise ValueError( "If no interval is specified, then every shake group must " "include a defined duration." ) return self._append_and_return( Spectrophotometry( dataref=dataref, object=obj, groups=groups, interval=interval, num_intervals=num_intervals, temperature=temperature, shake_before=shake_before, ) )
# pylint: disable=protected-access
[docs] def transfer( self, source: WellParam, destination: WellParam, volume: Union[VOLUME, List[VOLUME]], rows: int = 1, columns: int = 1, source_liquid: LiquidClass = LiquidClass, destination_liquid: LiquidClass = LiquidClass, method: Transfer = Transfer, one_tip: bool = False, density: Optional[DENSITY] = None, mode: Optional[str] = None, informatics: Optional[List[Informatics]] = None, ): """Generates LiquidHandle instructions between wells Transfer liquid between specified pairs of source & destination wells. Parameters ---------- source : Well or WellGroup or list(Well) Well(s) to transfer liquid from. destination : Well or WellGroup or list(Well) Well(s) to transfer liquid to. volume : str or Unit or list(str) or list(Unit) Volume(s) of liquid to be transferred from source wells to destination wells. The number of volumes specified must correspond to the number of destination wells. rows : int, optional Number of rows to be concurrently transferred columns : int, optional Number of columns to be concurrently transferred source_liquid : LiquidClass or list(LiquidClass), optional Type(s) of liquid contained in the source Well. This affects the aspirate and dispense behavior including the flowrates, liquid level detection thresholds, and physical movements. destination_liquid : LiquidClass or list(LiquidClass), optional Type(s) of liquid contained in the destination Well. This affects liquid level detection thresholds. method : Transfer or list(Transfer), optional Integrates with the specified source_liquid and destination_liquid to define a set of physical movements. one_tip : bool, optional If True then a single tip will be used for all operations density : Unit or str, optional Density of the liquid to be aspirated/dispensed mode : str, optional The liquid handling mode informatics : list(Informatics), optional List of Informatics describing the intended aliquot effects upon completion of this instruction. Returns ------- list(LiquidHandle) Returns a list of :py:class:`autoprotocol.instruction.LiquidHandle` instructions created from the specified parameters Raises ------ ValueError if the specified parameters can't be interpreted as lists of equal length ValueError if one_tip is true, but not all transfer methods have a tip_type Examples -------- Transfer between two single wells .. code-block:: python from autoprotocol import Protocol, Unit p = Protocol() source = p.ref("source", cont_type="384-flat", discard=True) destination = p.ref( "destination", cont_type="394-pcr", discard=True ) p.transfer(source.well(0), destination.well(1), "5:ul") Sequential transfers between two groups of wells .. code-block:: python sources = source.wells_from(0, 8, columnwise=True) dests = destination.wells_from(1, 8, columnwise=True) volumes = [Unit(x, "ul") for x in range(1, 9)] p.transfer(sources, dests, volumes) Concurrent transfers between two groups of wells .. code-block:: python # single-column concurrent transfer p.transfer( source.well(0), destination.well(1), "5:ul", rows=8 ) # 96-well concurrent transfer from the A1 to B2 quadrants p.transfer( source.well(0), destination.well(13), "5:ul", rows=8, columns=12 ) # 384-well concurrent transfer p.transfer( source.well(0), destination.well(0), "5:ul", rows=16, columns=24 ) Transfer with extra parameters .. code-block:: python from autoprotocol.liquid_handle import Transfer from autoprotocol.instruction import LiquidHandle p.transfer( source.well(0), destination.well(0), "5:ul", method=Transfer( mix_before=True, dispense_z=LiquidHandle.builders.position_z( reference="well_top" ) ) ) Transfer using other built in Transfer methods .. code-block:: python from autoprotocol.liquid_handle import DryWellTransfer p.transfer( source.well(0), destination.well(1), "5:ul", method=DryWellTransfer ) For examples of other more complicated behavior, see the documentation for LiquidHandleMethod. See Also -------- Transfer : base LiquidHandleMethod for transfer operations """ def location_helper(source, destination, volume, method, density): """Generates LiquidHandle transfer locations Parameters ---------- source : Well Well to transfer liquid from. destination : Well Well to transfer liquid to. volume : Unit The volume of liquid to be transferred from source well to destination well. method : Transfer Integrates the two input liquid classes and defines a set of transfers based on their attributes and methods. density : Unit The density of liquid to be aspirated/dispensed. Returns ------- list(dict) LiquidHandle locations """ self._remove_cover(source.container, "liquid_handle from") self._remove_cover(destination.container, "liquid_handle into") self._transfer_volume(source, destination, volume, method._shape) return [ LiquidHandle.builders.location( location=source, # pylint: disable=protected-access transports=method._aspirate_transports(volume, density), ), LiquidHandle.builders.location( location=destination, # pylint: disable=protected-access transports=method._dispense_transports(volume, density), ), ] def informatics_helper(informatics, dest, multi_src): """ Checks Informatics against the Instruction param values, and split Informatics per destination well as needed. Parameters ---------- informatics: list(Informatics) list of Informatics provided by the user input dest: WellGroup destination wells that could be affected by the instruction multi_src: boolean True if there are multiple sources in the transfer Returns ------- list(Informatics) list of Informatics per destination well Examples -------- .. code-block:: python from autoprotocol.container import Container, Well, WellGroup from autoprotocol.informatics import AttachCompounds from autoprotocol.protocol import Protocol from autoprotocol.unit import Unit p = Protocol() resource = p.ref("resource", None, "96-flat", discard=True) container = p.ref("container", cont_type="96-flat", discard=True) dest_wells = WellGroup( [ container.well[0], container.well[1], container.well[2] ] ) compd1 = Compound("CCCC") compd2 = Compound("C1=CC=CC=C1") # Single Informatics a destination well in a transfer from a source well to a destination well p.transfer( resource.well(0).set_volume("40:microliter"), dest_wells[0], "5:microliter", informatics=[AttachCompounds(dest_wells[0], [compd1])], ) # Single Informatics for a destination well in a transfer from multiple sources to single destination p.transfer( [resource.well(0).set_volume("40:microliter"),resource.well(1).set_volume("40:microliter")], test_wells[0], "5:microliter", informatics=[AttachCompounds(dest_wells[0], [compd1])] ) # Single Informatics for multiple wells in a transfer from single source to multiple destinations p.transfer( resource.well(0).set_volume("40:microliter"), dest_wells, "5:microliter", informatics=[AttachCompounds(dest_wells, [compd1])], ) # Multiple Informatics for a well in a transfer from a source well to a destination well self.p.transfer( resource.well(0).set_volume("40:microliter"), dest_wells[0], "10:microliter", informatics=[ AttachCompounds(dest_wells[0], [compd1]), AttachCompounds(dest_wells[0], [compd2]) ] ) # Multiple Informatics for multiple wells in transfer from a source to many destination wells p.transfer( resource.well(0).set_volume("40:microliter"), dest_wells, "5:microliter", informatics=[ AttachCompounds([dest_wells[0], dest_wells[1]], [compd1]), AttachCompounds(dest_wells, [compd2]) ] ) .. code-block:: json # Only showing details on the informatics attribute for the purpose of demonstrating how informatics # param is being serialized per instruction. [ # a transfer from a source well to a destination well with one Informatics { "op": "liquid_handle" "locations": [ {"location": ["resource/0"], "transports": ...} ] ..., "informatics": [ { "type: "attach_compounds", "data": {"wells": "container/0", "compounds": ["CCCC"]} } ] }, # a transfer from multiple sources to a destination with one Informatics { "op": "liquid_handle" "locations": [ {"location": ["resource/0"], "transports": ...}, {"location": ["contianer/0"], "transports": ...} ], ..., }, { "op": "liquid_handle" "locations": [ {"location": ["resource/1"], "transports": ...}, {"location": ["container/0"], "transports": ...} ], ..., "informatics": [ { "type": "attach_compounds", "data": {"wells": "container/0", "compounds": ["CCCC"]} } ] }, # Single Informatics for multiple wells in a transfer from single source to multiple destinations { "op": "liquid_handle" "locations": [ {"location": ["resource/0"], "transports": ...}, {"location": ["container/0"], "transports": ...} ], ..., "informatics": [ { "type": "attach_compounds", "data": {"wells": "container/0", "compounds": ["CCCC"]} } ] }, { "op": "liquid_handle" "locations": [ {"location": ["resource/0"], "transports": ...}, {"location": ["container/1"], "transports": ...} ], ..., "informatics": [ { "type": "attach_compounds", "data": {"wells": "container/2", "compounds": ["CCCC"]} } ] }, { "op": "liquid_handle" "locations": [ {"location": ["resource/0"], "transports": ...}, {"location": ["container/2"], "transports": ...} ], ..., "informatics": [ { "type": "attach_compounds", "data": {"wells": "container/2", "compounds": ["CCCC"]} } ] }, # Multiple Informatics for a well in a transfer from a source well to a destination well { "op": "liquid_handle" "locations": [ {"location": "resource/0", ...}, {"location": "container/0", ...} ], ..., "informatics": [ { "type": "attach_compounds" "data":{ "wells": "container/0", "compounds": ["CCCC", "C1=CC=CC=C1"] } } ] }, # Multiple Informatics for multiple wells in transfer from a source to many destination wells { "op": "liquid_handle" "locations": [ {"location": "resource/0", ...}, {"location": "container/0", ...} ], ..., "informatics": [ { "type": "attach_compounds" "data":{ "wells": "container/0", "compounds": ["CCCC", "C1=CC=CC=C1"] } } ] }, { "op": "liquid_handle" "locations": [ {"location": "resource/0", ...}, {"location": "container/1", ...} ], ..., "informatics": [ { "type": "attach_compounds" "data":{ "wells": "container/1", "compounds": ["CCCC", "C1=CC=CC=C1"] } } ] }, { "op": "liquid_handle" "locations": [ {"location": "resource/0", ...}, {"location": "container/2", ...} ], ..., "informatics": [ { "type": "attach_compounds" "data":{ "wells": "container/2", "compounds": ["C1=CC=CC=C1"] } } ] } ] Raises ------ TypeError Informatics provided is not valid or not supported ValueError Informatics wells must match wells in Instruction ValueError Multiple instances of Informatics on a Well ValueError Parsed list of Informatics length should match the length of destination """ if multi_src: dest_count = 1 else: dest_count = len(dest) # When one informatics is provided for a sequence of liquid_handle Instruction, # Informatics is instantiated for each Instruction. Currently, this does not accept # users to attach Informatics to part of the wells or sequence. if len(informatics) == 1: # In the future, if we are to add more Informatics subclasses, we may create # subclass for InformaticsWithWells and InformaicsWithoutWells instead of # specifying each type here. if isinstance(informatics[0], AttachCompounds): compounds = informatics[0].compounds wells = WellGroup(informatics[0].wells) # Informatics must include all destination wells if len(wells) == dest_count and set(wells) == set(destination): info_compd = compounds * dest_count informatics_list = [] for well, compd in zip(destination, info_compd): if isinstance(informatics[0], AttachCompounds): if not isinstance(compd, list): compd = [compd] informatics_list.append(AttachCompounds(well, compd)) else: raise ValueError( f"Informatics wells: {wells} do not match wells used in Instruction." ) else: raise TypeError( f"Informatics:{informatics} is not available in this protocol." ) # If liquid_handle has multiple sources and one destination, Informatics should be added # to the last transfer. if multi_src: informatics_list = [None] * (len(dest) - 1) + informatics else: wells_compounds_dict = {} # when multiple Informatics are provided, `wells` in all Informatics must # sum up to include all destination wells. for info in informatics: if isinstance(info, AttachCompounds): wells_count = len(WellGroup(info.wells)) compounds = info.compounds * wells_count for w, c in zip(WellGroup(info.wells).wells, compounds): if not isinstance(c, list): c = [c] # if there are multiple Informatics for a well, all the compounds # will be added to a single instance of Informatics for that well. if w not in wells_compounds_dict.keys(): wells_compounds_dict[w] = c else: all_compounds = wells_compounds_dict[w] if not isinstance(all_compounds, list): all_compounds = [all_compounds] all_compounds.extend(c) compounds_set = set(all_compounds) wells_compounds_dict[w] = list(compounds_set) else: raise TypeError( f"Informatics:{informatics} is not available in this protocol." ) if len(wells_compounds_dict.keys()) == dest_count: informatics_list = [] # sort informatics_list by the destination order wells_compounds_dict = sorted( wells_compounds_dict.items(), key=lambda pair: destination.wells.index(pair[0]), ) for k, v in wells_compounds_dict: informatics_list.append(AttachCompounds(k, v)) else: raise ValueError( f"the length of provided informatics: {len(wells_compounds_dict.keys())} does " f"not match the number of available wells." ) return informatics_list # validate parameter types source = WellGroup(source) destination = WellGroup(destination) count = max((len(source), len(destination))) multiple_source = len(source) > len(destination) if len(source) == 1: source = WellGroup([source[0]] * count) if len(destination) == 1: destination = WellGroup([destination[0]] * count) if not isinstance(volume, list): volume = [volume] * count volume = [parse_unit(_, "uL") for _ in volume] if density: if not isinstance(density, list): density = [density] * count elif len(density) != count: raise ValueError( f"the length of provided density: {len(density)} does not match the number of transports." ) density = [parse_unit(d, "mg/ml") for d in density] for d in density: if d.magnitude <= 0: raise ValueError(f"Density: {d} must be a value larger than 0.") else: # if density is None, it should still be a list of None density = [density] * count if not isinstance(source_liquid, list): source_liquid = [source_liquid] * count source_liquid = [_validate_as_instance(_, LiquidClass) for _ in source_liquid] if not isinstance(destination_liquid, list): destination_liquid = [destination_liquid] * count destination_liquid = [ _validate_as_instance(_, LiquidClass) for _ in destination_liquid ] if not isinstance(method, list): method = [method] * count method = [_validate_as_instance(_, Transfer) for _ in method] # if informatics is provided for multiple wells, split Informatics for each destination well # with the specified compounds. if informatics is not None and len(informatics) > 0: informatics_list = informatics_helper( informatics, destination, multiple_source ) else: informatics_list = [informatics] * count # validate parameter counts countable_parameters = ( source, destination, volume, density, source_liquid, destination_liquid, method, informatics_list, ) correct_parameter_counts = all(len(_) == count for _ in countable_parameters) if not correct_parameter_counts: raise ValueError( f"Specified parameters {countable_parameters} could not all be interpreted as the same length {correct_parameter_counts}." ) # format shape shape = LiquidHandle.builders.shape(rows, columns, None) # validate all containers against the shape for aliquot in sum(source, destination): container_type = aliquot.container.container_type _check_container_type_with_shape(container_type, shape) # apply liquid classes to transfer methods for src, des, met in zip(source_liquid, destination_liquid, method): met._shape = LiquidHandle.builders.shape(**shape) met._source_liquid = src met._destination_liquid = des # apply tip types to transfer methods for vol, met in zip(volume, method): if met._has_calibration() and not met.tip_type: try: # met.tip_type = met._rec_tip_type(vol) met._rec_tip_type(vol) except RuntimeError: met.tip_type = met._get_sorted_tip_types()[-1].name # if one tip is true then all methods need to have the same tip_type if one_tip is True: tip_types = [_.tip_type for _ in method] if not all(_ == tip_types[0] for _ in tip_types): raise ValueError( f"If one_tip is true and any tip_type is set, then all tip types must be the same but {tip_types} was specified." ) # generate either a LiquidHandle location or instruction list locations, instructions = [], [] for src, des, vol, met, dens, informatics in zip( source, destination, volume, method, density, informatics_list ): max_tip_capacity = met._tip_capacity() remaining_vol = vol while remaining_vol > Unit(0, "ul"): transfer_vol = min(remaining_vol, max_tip_capacity) if one_tip is True: locations += location_helper(src, des, transfer_vol, met, dens) else: location_transports = location_helper( src, des, transfer_vol, met, dens ) source_transports = location_transports[0]["transports"] instruction_mode = mode if not instruction_mode: instruction_mode = LiquidHandle.builders.desired_mode( source_transports, mode ) if not isinstance(informatics, list): informatics = [informatics] instructions.append( LiquidHandle( location_transports, shape=met._shape, mode=instruction_mode, mode_params=( LiquidHandle.builders.instruction_mode_params( tip_type=met.tip_type ) ), informatics=informatics, ) ) remaining_vol -= transfer_vol # if one tip is true then there's a locations list if locations: source_transports = locations[0]["transports"] # if not mode: mode = LiquidHandle.builders.desired_mode(source_transports, mode) instructions.append( LiquidHandle( locations, shape=shape, mode=mode, mode_params=LiquidHandle.builders.instruction_mode_params( tip_type=method[0].tip_type ), ) ) return self._append_and_return(instructions)
# pylint: disable=protected-access
[docs] def mix( self, well: WellParam, volume: Union[VOLUME, List[VOLUME]], rows: int = 1, columns: int = 1, liquid: LiquidClass = LiquidClass, method: Mix = Mix, one_tip: bool = False, mode: Optional[str] = None, ): """Generates LiquidHandle instructions within wells Mix liquid in specified wells. Parameters ---------- well : Well or WellGroup or list(Well) Well(s) to be mixed. volume : str or Unit or list(str) or list(Unit) Volume(s) of liquid to be mixed within the specified well(s). The number of volume(s) specified must correspond with the number of well(s). rows : int, optional Number of rows to be concurrently mixed columns : int, optional Number of columns to be concurrently mixed liquid : LiquidClass or list(LiquidClass), optional Type(s) of liquid contained in the Well(s). This affects the aspirate and dispense behavior including the flowrates, liquid level detection thresholds, and physical movements. method : Mix or list(Mix), optional Method(s) with which Integrates with the specified liquid to define a set of physical movements. one_tip : bool, optional If True then a single tip will be used for all operations mode : str, optional The liquid handling mode Returns ------- list(LiquidHandle) Returns a list of :py:class:`autoprotocol.instruction.LiquidHandle` instructions created from the specified parameters Raises ------ ValueError if the specified parameters can't be interpreted as lists of equal length ValueError if one_tip is true, but not all mix methods have a tip_type ValueError if the specified volume is larger than the maximum tip capacity of the available liquid_handling devices for a given mix Examples -------- Mix within a single well .. code-block:: python from autoprotocol import Protocol, Unit p = Protocol() plate = p.ref("example_plate", cont_type="384-flat", discard=True) p.mix(plate.well(0), "5:ul") Sequential mixes within multiple wells .. code-block:: python wells = plate.wells_from(0, 8, columnwise=True) volumes = [Unit(x, "ul") for x in range(1, 9)] p.mix(wells, volumes) Concurrent mixes within multiple wells .. code-block:: python # single-column concurrent mix p.mix(plate.well(0), "5:ul", rows=8) # 96-well concurrent mix in the A1 quadrant p.mix(plate.well(0), "5:ul", rows=8, columns=12) # 96-well concurrent mix in the A2 quadrant p.mix(plate.well(1), "5:ul", rows=8, columns=12) # 384-well concurrent mix p.mix(plate.well(0), "5:ul", rows=16, columns=24) Mix with extra parameters .. code-block:: python from autoprotocol.liquid_handle import Mix from autoprotocol.instruction import LiquidHandle p.mix( plate.well(0), "5:ul", rows=8, method=Mix( mix_params=LiquidHandle.builders.mix( ) ) ) See Also -------- Mix : base LiquidHandleMethod for mix operations """ def location_helper(aliquot: Well, volume: Unit, method: Mix): """Generates LiquidHandle mix locations Parameters ---------- aliquot : Well Wells to transfer mix liquid in. volume : Unit The volume of liquid to be transferred within the aliquot. method : Mix Integrates the input liquid class and defines a set of transfers based on its attributes and methods. Returns ------- list(dict) LiquidHandle locations """ self._remove_cover(aliquot.container, "liquid_handle in") return [ LiquidHandle.builders.location( location=aliquot, transports=method._mix_transports(volume) ) ] # validate parameter types well = WellGroup(well) count = len(well) if not isinstance(volume, list): volume = [volume] * count volume = [parse_unit(_, "uL") for _ in volume] if not isinstance(liquid, list): liquid = [liquid] * count liquid = [_validate_as_instance(_, LiquidClass) for _ in liquid] if not isinstance(method, list): method = [method] * count method = [_validate_as_instance(_, Mix) for _ in method] # validate parameter counts countable_parameters = (well, volume, liquid, method) correct_parameter_counts = all(len(_) == count for _ in countable_parameters) if not correct_parameter_counts: raise ValueError( f"Specified parameters {countable_parameters} could not all be interpreted as the same length {correct_parameter_counts}." ) # format shape shape = LiquidHandle.builders.shape(rows, columns, None) # validate all containers against the shape for aliquot in well: container_type = aliquot.container.container_type _check_container_type_with_shape(container_type, shape) # apply liquid classes to mix methods for liq, met in zip(liquid, method): met._shape = LiquidHandle.builders.shape(**shape) met._liquid = liq # apply tip types to mix methods for vol, met in zip(volume, method): if met._has_calibration() and not met.tip_type: try: met.tip_type = met._rec_tip_type(vol) except RuntimeError: met.tip_type = met._get_sorted_tip_types()[-1].name # if one tip is true then all methods need to have the same tip_type if one_tip is True: tip_types = [_.tip_type for _ in method] if not all(_ == tip_types[0] for _ in tip_types): raise ValueError( f"If one_tip is true and any tip_type is set, then all tip types must be the same but {tip_types} was specified." ) # generate either a LiquidHandle location or instruction list locations, instructions = [], [] for wel, vol, met in zip(well, volume, method): max_tip_capacity = met._tip_capacity() if vol > max_tip_capacity: raise ValueError( f"Attempted mix volume {vol} is larger than the maximum capacity of {max_tip_capacity} for transfer shape {vol.shape}." ) self._remove_cover(wel.container, "liquid_handle mix") if one_tip is True: locations += location_helper(wel, vol, met) else: location_transports = location_helper(wel, vol, met) source_transports = location_transports[0]["transports"] instruction_mode = mode if not instruction_mode: instruction_mode = LiquidHandle.builders.desired_mode( source_transports, mode ) instructions.append( LiquidHandle( location_transports, shape=met._shape, mode=instruction_mode, mode_params=( LiquidHandle.builders.instruction_mode_params( tip_type=met.tip_type ) ), ) ) # if one tip is true then there's a locations list if locations: source_transports = locations[0]["transports"] if not mode: mode = LiquidHandle.builders.desired_mode(source_transports, mode) instructions.append( LiquidHandle( locations, shape=shape, mode=mode, mode_params=LiquidHandle.builders.instruction_mode_params( tip_type=method[0].tip_type ), ) ) return self._append_and_return(instructions)
[docs] def spread( self, source: Well, dest: Well, volume: VOLUME = "50:microliter", dispense_speed: ACCELERATION = "20:microliter/second", ): """ Spread the specified volume of the source aliquot across the surface of the agar contained in the object container. Uses a spiral pattern generated by a set of liquid_handle instructions. Example Usage: .. code-block:: python p = Protocol() agar_plate = p.ref("agar_plate", None, "1-flat", discard=True) bact = p.ref("bacteria", None, "micro-1.5", discard=True) p.spread(bact.well(0), agar_plate.well(0), "55:microliter") Parameters ---------- source : Well Source of material to spread on agar dest : Well Reference to destination location (plate containing agar) volume : str or Unit, optional Volume of source material to spread on agar dispense_speed : str or Unit, optional Speed at which to dispense source aliquot across agar surface Returns ------- LiquidHandle Returns a :py:class:`autoprotocol.instruction.LiquidHandle` instruction created from the specified parameters Raises ------ TypeError If specified source is not of type Well TypeError If specified destination is not of type Well """ def euclidean_distance(point_a, point_b): """ Calculate the euclidean distance between a pair of xy coordinates Parameters ---------- point_a: Iterable First point point_b: Iterable Second point Returns ------- float The distance between the two points """ from math import sqrt x_distance = abs(point_a[0] - point_b[0]) y_distance = abs(point_a[1] - point_b[1]) return sqrt(x_distance**2 + y_distance**2) # Check validity of Well inputs if not isinstance(source, Well): raise TypeError("Source must be of type Well.") if not isinstance(dest, Well): raise TypeError("Destination, (dest), must be of type Well.") self._remove_cover(source.container, "spread") self._remove_cover(dest.container, "spread") volume = Unit(volume) if dest.volume: dest.volume += volume else: dest.volume = volume if source.volume: source.volume -= volume aspirate_transport_list = [ LiquidHandle.builders.transport( mode_params=LiquidHandle.builders.mode_params( position_z=LiquidHandle.builders.position_z( reference="liquid_surface", offset=Unit("-1:mm"), detection_method="capacitance", detection_threshold=AGAR_CLLD_THRESHOLD, ) ) ), LiquidHandle.builders.transport( volume=-volume, mode_params=LiquidHandle.builders.mode_params( position_z=LiquidHandle.builders.position_z( reference="liquid_surface", detection_method="tracked", offset=Unit("-1.0:mm"), ) ), ), ] dispense_transport_list = [ LiquidHandle.builders.transport( mode_params=LiquidHandle.builders.mode_params( position_z=LiquidHandle.builders.position_z( reference="liquid_surface", detection_method="capacitance", detection_threshold=AGAR_CLLD_THRESHOLD, ) ) ) ] distances = [ euclidean_distance(first, second) for first, second in zip(SPREAD_PATH, SPREAD_PATH[1:]) ] distance_total = sum(distances) distance_ratios = [dist / distance_total for dist in distances] for ratio, position in zip(distance_ratios, SPREAD_PATH[1:]): dispense_transport_list += [ LiquidHandle.builders.transport( volume=volume * ratio, flowrate=LiquidHandle.builders.flowrate(dispense_speed), mode_params=LiquidHandle.builders.mode_params( position_x=LiquidHandle.builders.position_xy(position[0]), position_y=LiquidHandle.builders.position_xy(position[1]), position_z=LiquidHandle.builders.position_z( reference="liquid_surface", detection_method="tracked", offset=Unit("0.5:mm"), ), ), ) ] location = [ LiquidHandle.builders.location( location=source, transports=aspirate_transport_list ), LiquidHandle.builders.location( location=dest, transports=dispense_transport_list ), ] return self._append_and_return(LiquidHandle(location))
def _transfer_volume( self, source: Well, destination: Well, volume: Unit, shape: DispenseShape ): """ Transfers volume and properties between aliquots. Parameters ---------- source : Well The shape origin to be transferred from destination : Well The shape origin to be transferred to volume : Unit The volume to be transferred shape : dict See Also Instruction.builders.shape Raises ------ RuntimeError If the inferred sources and destinations aren't the same length """ source_wells = source.container.wells_from_shape(source.index, shape) dest_wells = destination.container.wells_from_shape(destination.index, shape) if not len(source_wells) == len(dest_wells): raise RuntimeError( f"Transfer source: {source_wells} and destination: " f"{dest_wells} WellGroups didn't have the same number of wells." ) for source_well, dest_well in zip(source_wells, dest_wells): if self.propagate_properties: dest_well.add_properties(source_well.properties) if source_well.volume is not None: source_well.volume -= volume if dest_well.volume is not None: dest_well.volume += volume else: dest_well.volume = volume
[docs] def evaporate( self, ref: Container, mode: EvaporateMode, duration: TIME, evaporator_temperature: TEMPERATURE, mode_params: Optional[EvaporateModeParams] = None, ): """ Removes liquid or moisture from a container using the mode specified. Example Usage: .. code-block:: python p = Protocol() c = p.ref("container", id=None, cont_type="micro-1.5", storage="cold_20") blowdown_params = Evaporate.builders.get_mode_params( mode="blowdown", mode_params={ "gas":"nitrogen", "vortex_speed":Unit("200:rpm"), "blow_rate": "200:uL/sec" } ) p.evaporate( c, mode="blowdown", duration="10:minute", evaporator_temperature="22:degC", mode_params = blowdown_params ) .. code-block:: json { "op": "evaporate", "ref": "container", "mode": "blowdown", "duration": "10:minute", "evaporator_temperature": "22:degC", "mode_params": { "gas": "ntirogen", "vortex_speed": "200:rpm", "blow_rate": "200:uL/sec" } } Parameters ---------- ref : Container Sample container mode : Str The mode of evaporation method duration : Unit or Str The length of time the sample is evaporated for evaporator_temperature : Unit or str The incubation temperature of the sample being evaporated mode_params : Dict Dictionary of parameters for evaporation mode Returns ------- Evaporate Returns a :py:class:`autoprotocol.instruction.Evaporate` instruction created from the specified parameters Raises ------ TypeError If the provided object is not a Container type. ValueError If the duration is less than 0 minute TypeError If evaporator_temperature is not provided in Unit or str ValueError If the evaporation_temperature is lower than or equal to condenser_temperature """ duration = parse_unit(duration, "minute") evaporator_temperature = parse_unit(evaporator_temperature, "celsius") mode_params = Evaporate.builders.get_mode_params(mode, mode_params) if not isinstance(ref, Container): raise TypeError("Param `ref` must be a container object.") if duration <= Unit("0:minute"): raise ValueError( f"Param `duration`: {duration} should be longer than 0 minute." ) if mode_params: condenser_temp = mode_params.get("condenser_temperature") if "condenser_temperature" in mode_params.keys(): if condenser_temp >= evaporator_temperature: raise ValueError( f"Param `condenser_temperature`: {condenser_temp} " "cannot be higher than the evaporator_temperature:" f" {evaporator_temperature}" ) # Autoprotocol Evaporate assumes full evaporation - set volume to 0uL for well in ref.all_wells(): well.set_volume(Unit(0, "uL")) return self._append_and_return( Evaporate( ref=ref, duration=duration, evaporator_temperature=evaporator_temperature, mode=mode, mode_params=mode_params, ) )