<s> from . container import Container , Well , WellGroup <EOL> from . container_type import ContainerType , _CONTAINER_TYPES <EOL> from . unit import Unit <EOL> from . instruction import * <EOL> from . pipette_tools import assign <EOL> from . util import check_valid_origin <EOL> from . util import check_stamp_append <EOL> from . util import check_valid_mag <EOL> from . util import check_valid_mag_params <EOL> from . util import check_valid_incubate_params <EOL> from . util import check_valid_gel_purify_extract <EOL> import sys <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> xrange = range <EOL> basestring = str <EOL> '''<STR_LIT>''' <EOL> class Ref ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , opts , container ) : <EOL> self . name = name <EOL> self . opts = opts <EOL> self . container = container <EOL> class Protocol ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , refs = None , instructions = None ) : <EOL> super ( Protocol , self ) . __init__ ( ) <EOL> self . refs = refs or { } <EOL> self . instructions = instructions or [ ] <EOL> def container_type ( self , shortname ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( shortname , ContainerType ) : <EOL> return shortname <EOL> elif shortname in _CONTAINER_TYPES : <EOL> return _CONTAINER_TYPES [ shortname ] <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> ( shortname , str ( _CONTAINER_TYPES . keys ( ) ) ) ) <EOL> def ref ( self , name , id = None , cont_type = None , storage = None , discard = None ) : <EOL> """<STR_LIT>""" <EOL> if name in self . refs . keys ( ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> opts = { } <EOL> try : <EOL> cont_type = self . container_type ( cont_type ) <EOL> if id and cont_type : <EOL> opts [ "<STR_LIT:id>" ] = id <EOL> elif cont_type : <EOL> opts [ "<STR_LIT>" ] = cont_type . shortname <EOL> except ValueError : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if storage : <EOL> opts [ "<STR_LIT:store>" ] = { "<STR_LIT>" : storage } <EOL> elif discard and not storage : <EOL> opts [ "<STR_LIT>" ] = discard <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> container = Container ( <EOL> id , cont_type , name = name , storage = storage if storage else None ) <EOL> self . refs [ name ] = Ref ( name , opts , container ) <EOL> return container <EOL> def append ( self , instructions ) : <EOL> """<STR_LIT>""" <EOL> if type ( instructions ) is list : <EOL> self . instructions . extend ( instructions ) <EOL> else : <EOL> self . instructions . append ( instructions ) <EOL> def as_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> outs = { } <EOL> for n , ref in self . refs . items ( ) : <EOL> for well in ref . container . _wells : <EOL> if well . name or len ( well . properties ) > <NUM_LIT:0> : <EOL> if n not in outs . keys ( ) : <EOL> outs [ n ] = { } <EOL> outs [ n ] [ str ( well . index ) ] = { } <EOL> if well . name : <EOL> outs [ n ] [ str ( well . index ) ] [ "<STR_LIT:name>" ] = well . name <EOL> if len ( well . properties ) > <NUM_LIT:0> : <EOL> outs [ n ] [ str ( well . index ) ] [ <EOL> "<STR_LIT>" ] = well . properties <EOL> if "<STR_LIT:store>" in ref . opts : <EOL> ref . opts [ "<STR_LIT:store>" ] = { "<STR_LIT>" : ref . container . storage } <EOL> if ref . container . storage is None and "<STR_LIT>" not in ref . opts : <EOL> ref . opts [ "<STR_LIT>" ] = True <EOL> del ref . opts [ "<STR_LIT:store>" ] <EOL> elif ref . container . storage is not None and "<STR_LIT>" in ref . opts : <EOL> ref . opts [ "<STR_LIT:store>" ] = { "<STR_LIT>" : ref . container . storage } <EOL> del ref . opts [ "<STR_LIT>" ] <EOL> if outs : <EOL> return { <EOL> "<STR_LIT>" : dict ( <EOL> ( key , value . opts ) <EOL> for key , value in self . refs . items ( ) <EOL> ) , <EOL> "<STR_LIT>" : list ( map ( lambda x : self . _refify ( x . data ) , <EOL> self . instructions ) ) , <EOL> "<STR_LIT>" : outs <EOL> } <EOL> else : <EOL> return { <EOL> "<STR_LIT>" : dict ( <EOL> ( key , value . opts ) <EOL> for key , value in self . refs . items ( ) <EOL> ) , <EOL> "<STR_LIT>" : list ( map ( lambda x : self . _refify ( x . data ) , <EOL> self . instructions ) ) <EOL> } <EOL> def store ( self , container , condition ) : <EOL> """<STR_LIT>""" <EOL> if not condition or condition == "<STR_LIT>" : <EOL> condition = None <EOL> if not isinstance ( container , Container ) : <EOL> raise TypeError ( <EOL> "<STR_LIT>" ) <EOL> container . storage = condition <EOL> r = self . refs . get ( container . name ) <EOL> if not r : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" ) <EOL> if "<STR_LIT>" in r . opts : <EOL> r . opts . pop ( "<STR_LIT>" ) <EOL> if condition : <EOL> r . opts [ "<STR_LIT:store>" ] = { "<STR_LIT>" : str ( condition ) } <EOL> else : <EOL> r . opts . pop ( "<STR_LIT:store>" ) <EOL> r . opts [ "<STR_LIT>" ] = True <EOL> def distribute ( self , source , dest , volume , allow_carryover = False , <EOL> mix_before = False , mix_vol = None , repetitions = <NUM_LIT:10> , <EOL> flowrate = "<STR_LIT>" , aspirate_speed = None , <EOL> aspirate_source = None , distribute_target = None , <EOL> pre_buffer = None , disposal_vol = None , transit_vol = None , <EOL> blowout_buffer = None , tip_type = None , new_group = False ) : <EOL> """<STR_LIT>""" <EOL> opts = { } <EOL> try : <EOL> dists = self . fill_wells ( dest , source , volume , distribute_target ) <EOL> except ValueError : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> groups = [ ] <EOL> for d in dists : <EOL> opts = { } <EOL> if mix_before : <EOL> if not mix_vol : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> opts [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if allow_carryover : <EOL> opts [ "<STR_LIT>" ] = allow_carryover <EOL> opts [ "<STR_LIT>" ] = d [ "<STR_LIT>" ] <EOL> opts [ "<STR_LIT:to>" ] = d [ "<STR_LIT:to>" ] <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( opts , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( opts , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> groups . append ( { "<STR_LIT>" : opts } ) <EOL> if new_group : <EOL> self . append ( Pipette ( groups ) ) <EOL> else : <EOL> self . _pipette ( groups ) <EOL> def transfer ( self , source , dest , volume , one_source = False , one_tip = False , <EOL> aspirate_speed = None , dispense_speed = None , <EOL> aspirate_source = None , dispense_target = None , pre_buffer = None , <EOL> disposal_vol = None , transit_vol = None , blowout_buffer = None , <EOL> tip_type = None , new_group = False , ** mix_kwargs ) : <EOL> """<STR_LIT>""" <EOL> opts = [ ] <EOL> source = WellGroup ( source ) <EOL> dest = WellGroup ( dest ) <EOL> len_source = len ( source . wells ) <EOL> len_dest = len ( dest . wells ) <EOL> if not one_source : <EOL> if len_dest > <NUM_LIT:1> and len_source == <NUM_LIT:1> : <EOL> source = WellGroup ( source . wells * len_dest ) <EOL> len_source = len ( source . wells ) <EOL> if len_dest == <NUM_LIT:1> and len_source > <NUM_LIT:1> : <EOL> dest = WellGroup ( dest . wells * len_source ) <EOL> len_dest = len ( dest . wells ) <EOL> if len_source != len_dest : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( volume , basestring ) or isinstance ( volume , Unit ) : <EOL> if len_dest == <NUM_LIT:1> and not one_source : <EOL> volume = [ Unit . fromstring ( volume ) . to ( "<STR_LIT>" ) ] * len_source <EOL> else : <EOL> volume = [ Unit . fromstring ( volume ) . to ( "<STR_LIT>" ) ] * len_dest <EOL> elif isinstance ( volume , list ) and len ( volume ) == len_dest : <EOL> volume = list ( <EOL> map ( lambda x : Unit . fromstring ( x ) . to ( "<STR_LIT>" ) , volume ) ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if one_source : <EOL> try : <EOL> source_vol = [ s . volume for s in source . wells ] <EOL> if sum ( [ a for a in volume ] ) > sum ( [ a for a in source_vol ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if len_source >= len_dest and all ( i > j for i , j in zip ( source_vol , volume ) ) : <EOL> sources = source . wells [ : len_dest ] <EOL> destinations = dest . wells <EOL> volumes = volume <EOL> else : <EOL> sources = [ ] <EOL> source_counter = <NUM_LIT:0> <EOL> destinations = [ ] <EOL> volumes = [ ] <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> max_decimal_places = <NUM_LIT:12> <EOL> for idx , d in enumerate ( dest . wells ) : <EOL> vol_d = volume [ idx ] <EOL> while vol_d > Unit . fromstring ( "<STR_LIT>" ) : <EOL> if vol > vol_d : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol_d ) <EOL> vol -= vol_d <EOL> vol . _magnitude = round ( <EOL> vol . _magnitude , max_decimal_places ) <EOL> vol_d -= vol_d <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> else : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol ) <EOL> vol_d -= vol <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> source_counter += <NUM_LIT:1> <EOL> if source_counter < len_source : <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> source = WellGroup ( sources ) <EOL> dest = WellGroup ( destinations ) <EOL> volume = volumes <EOL> except ( ValueError , AttributeError , TypeError ) as e : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> for s , d , v in list ( zip ( source . wells , dest . wells , volume ) ) : <EOL> if v > Unit ( <NUM_LIT> , "<STR_LIT>" ) : <EOL> diff = Unit . fromstring ( v ) <EOL> while diff > Unit ( <NUM_LIT> , "<STR_LIT>" ) : <EOL> v = Unit ( <NUM_LIT> , "<STR_LIT>" ) <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> if d . volume : <EOL> d . volume += v <EOL> else : <EOL> d . volume = v <EOL> if s . volume : <EOL> s . volume -= v <EOL> if mix_kwargs and ( <EOL> "<STR_LIT>" not in mix_kwargs and ( <EOL> "<STR_LIT>" not in mix_kwargs ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> if mix_kwargs . get ( "<STR_LIT>" ) : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or v / <NUM_LIT:2> , <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or <NUM_LIT:10> , <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or "<STR_LIT>" <EOL> } <EOL> if mix_kwargs . get ( "<STR_LIT>" ) : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or v / <NUM_LIT:2> , <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or <NUM_LIT:10> , <EOL> "<STR_LIT>" : mix_kwargs . get ( <EOL> "<STR_LIT>" ) or mix_kwargs . get ( <EOL> "<STR_LIT>" ) or "<STR_LIT>" <EOL> } <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( xfer , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( xfer , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if v > Unit ( <NUM_LIT:0> , "<STR_LIT>" ) : <EOL> opts . append ( xfer ) <EOL> diff -= Unit ( <NUM_LIT> , "<STR_LIT>" ) <EOL> v = diff <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> if d . volume : <EOL> d . volume += v <EOL> else : <EOL> d . volume = v <EOL> if s . volume : <EOL> s . volume -= v <EOL> if mix_kwargs and ( "<STR_LIT>" not in mix_kwargs and "<STR_LIT>" not in mix_kwargs ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if mix_kwargs . get ( "<STR_LIT>" ) : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or v / <NUM_LIT:2> , <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or <NUM_LIT:10> , <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or "<STR_LIT>" <EOL> } <EOL> if mix_kwargs . get ( "<STR_LIT>" ) : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or v / <NUM_LIT:2> , <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or <NUM_LIT:10> , <EOL> "<STR_LIT>" : mix_kwargs . get ( "<STR_LIT>" ) or mix_kwargs . get ( "<STR_LIT>" ) or "<STR_LIT>" <EOL> } <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( xfer , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( xfer , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if v > Unit ( <NUM_LIT:0> , "<STR_LIT>" ) : <EOL> opts . append ( xfer ) <EOL> trans = { } <EOL> assign ( trans , "<STR_LIT>" , tip_type ) <EOL> if one_tip : <EOL> trans [ "<STR_LIT>" ] = opts <EOL> if new_group : <EOL> self . append ( Pipette ( [ trans ] ) ) <EOL> else : <EOL> self . _pipette ( [ trans ] ) <EOL> else : <EOL> for x in opts : <EOL> trans = { } <EOL> assign ( trans , "<STR_LIT>" , tip_type ) <EOL> trans [ "<STR_LIT>" ] = [ x ] <EOL> if new_group : <EOL> self . append ( Pipette ( [ trans ] ) ) <EOL> else : <EOL> self . _pipette ( [ trans ] ) <EOL> def consolidate ( self , sources , dest , volumes , allow_carryover = False , <EOL> mix_after = False , mix_vol = None , <EOL> flowrate = "<STR_LIT>" , repetitions = <NUM_LIT:10> , <EOL> aspirate_speed = None , dispense_speed = None , aspirate_source = None , <EOL> dispense_target = None , pre_buffer = None , transit_vol = None , <EOL> blowout_buffer = None , tip_type = None , new_group = False ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( dest , ( Well , str ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( sources , ( Well , basestring ) ) : <EOL> sources = [ sources ] <EOL> if isinstance ( volumes , list ) : <EOL> if len ( volumes ) != len ( sources ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> volumes = [ Unit . fromstring ( v ) for v in volumes ] <EOL> else : <EOL> volumes = [ Unit . fromstring ( volumes ) ] * len ( sources ) <EOL> cons = { "<STR_LIT>" : { } } <EOL> cons_instr = cons [ "<STR_LIT>" ] <EOL> assign ( cons_instr , "<STR_LIT:to>" , dest ) <EOL> from_wells = [ ] <EOL> for s , v in zip ( sources , volumes ) : <EOL> source_opts = { } <EOL> source_opts [ "<STR_LIT>" ] = s <EOL> source_opts [ "<STR_LIT>" ] = v <EOL> assign ( source_opts , "<STR_LIT>" , aspirate_speed ) <EOL> assign ( source_opts , "<STR_LIT>" , aspirate_source ) <EOL> from_wells . append ( source_opts ) <EOL> if dest . volume : <EOL> dest . volume += v <EOL> else : <EOL> dest . volume = v <EOL> if s . volume : <EOL> s . volume -= v <EOL> assign ( cons_instr , "<STR_LIT>" , from_wells ) <EOL> if mix_after : <EOL> cons_instr [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( cons_instr , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( cons_instr , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if new_group : <EOL> self . append ( Pipette ( [ cons ] ) ) <EOL> else : <EOL> self . _pipette ( [ cons ] ) <EOL> def acoustic_transfer ( self , source , dest , volume , one_source = False , <EOL> droplet_size = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> transfers = [ ] <EOL> source = WellGroup ( source ) <EOL> dest = WellGroup ( dest ) <EOL> len_source = len ( source . wells ) <EOL> len_dest = len ( dest . wells ) <EOL> if not one_source : <EOL> if len_dest > <NUM_LIT:1> and len_source == <NUM_LIT:1> : <EOL> source = WellGroup ( source . wells * len_dest ) <EOL> len_source = len ( source . wells ) <EOL> if len_dest == <NUM_LIT:1> and len_source > <NUM_LIT:1> : <EOL> dest = WellGroup ( dest . wells * len_source ) <EOL> len_dest = len ( dest . wells ) <EOL> if len_source != len_dest : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( volume , basestring ) or isinstance ( volume , Unit ) : <EOL> if len_dest == <NUM_LIT:1> and not one_source : <EOL> volume = [ Unit . fromstring ( volume ) . to ( "<STR_LIT>" ) ] * len_source <EOL> else : <EOL> volume = [ Unit . fromstring ( volume ) . to ( "<STR_LIT>" ) ] * len_dest <EOL> elif isinstance ( volume , list ) and len ( volume ) == len_dest : <EOL> volume = list ( <EOL> map ( lambda x : Unit . fromstring ( x ) . to ( "<STR_LIT>" ) , volume ) ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if one_source : <EOL> try : <EOL> source_vol = [ s . volume for s in source . wells ] <EOL> if sum ( [ a for a in volume ] ) > sum ( [ a for a in source_vol ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if len_source >= len_dest and all ( i > j for i , j in zip ( source_vol , volume ) ) : <EOL> sources = source . wells [ : len_dest ] <EOL> destinations = dest . wells <EOL> volumes = volume <EOL> else : <EOL> sources = [ ] <EOL> source_counter = <NUM_LIT:0> <EOL> destinations = [ ] <EOL> volumes = [ ] <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> max_decimal_places = <NUM_LIT:12> <EOL> for idx , d in enumerate ( dest . wells ) : <EOL> vol_d = volume [ idx ] <EOL> while vol_d > Unit . fromstring ( "<STR_LIT>" ) : <EOL> if vol > vol_d : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol_d ) <EOL> vol -= vol_d <EOL> vol . _magnitude = round ( <EOL> vol . _magnitude , max_decimal_places ) <EOL> vol_d -= vol_d <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> else : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol ) <EOL> vol_d -= vol <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> source_counter += <NUM_LIT:1> <EOL> if source_counter < len_source : <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> source = WellGroup ( sources ) <EOL> dest = WellGroup ( destinations ) <EOL> volume = volumes <EOL> except ( ValueError , AttributeError , TypeError ) as e : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> for s , d , v in list ( zip ( source . wells , dest . wells , volume ) ) : <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> if d . volume : <EOL> d . volume += v <EOL> else : <EOL> d . volume = v <EOL> if s . volume : <EOL> s . volume -= v <EOL> if v > Unit ( <NUM_LIT:0> , '<STR_LIT>' ) : <EOL> transfers . append ( xfer ) <EOL> for x in transfers : <EOL> x [ "<STR_LIT>" ] = x [ "<STR_LIT>" ] . to ( "<STR_LIT>" ) <EOL> if self . instructions and self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" : <EOL> prev_inst = self . instructions [ - <NUM_LIT:1> ] . data [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] [ - <NUM_LIT:1> ] <EOL> if ( prev_inst [ "<STR_LIT>" ] . container == transfers [ <NUM_LIT:0> ] [ "<STR_LIT>" ] . container and <EOL> prev_inst [ "<STR_LIT:to>" ] . container == transfers [ <NUM_LIT:0> ] [ "<STR_LIT:to>" ] . container and <EOL> droplet_size == self . instructions [ - <NUM_LIT:1> ] . data [ "<STR_LIT>" ] ) : <EOL> self . instructions [ - <EOL> <NUM_LIT:1> ] . data [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] . extend ( transfers ) <EOL> return <EOL> self . append ( AcousticTransfer ( transfers , droplet_size ) ) <EOL> def stamp ( self , source_origin , dest_origin , volume , shape = dict ( rows = <NUM_LIT:8> , <EOL> columns = <NUM_LIT:12> ) , mix_before = False , mix_after = False , mix_vol = None , <EOL> repetitions = <NUM_LIT:10> , flowrate = "<STR_LIT>" , <EOL> aspirate_speed = None , dispense_speed = None , aspirate_source = None , <EOL> dispense_target = None , pre_buffer = None , disposal_vol = None , <EOL> transit_vol = None , blowout_buffer = None , one_source = False , <EOL> one_tip = False , new_group = False ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( source_origin , Container ) : <EOL> source_plate = source_origin <EOL> source_plate_type = source_plate . container_type <EOL> if source_plate_type . well_count == <NUM_LIT> : <EOL> source_origin = source_plate . well ( <NUM_LIT:0> ) <EOL> elif source_plate_type . well_count == <NUM_LIT> : <EOL> source_origin = source_plate . wells ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( dest_origin , Container ) : <EOL> dest_plate = dest_origin <EOL> dest_plate_type = dest_plate . container_type <EOL> if dest_plate_type . well_count == <NUM_LIT> : <EOL> dest_origin = dest_plate . well ( <NUM_LIT:0> ) <EOL> elif dest_plate_type . well_count == <NUM_LIT> : <EOL> dest_origin = dest_plate . wells ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if not ( isinstance ( source_origin , Well ) or isinstance ( source_origin , WellGroup ) ) or not ( isinstance ( dest_origin , Well ) or isinstance ( dest_origin , WellGroup ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> source = WellGroup ( source_origin ) <EOL> dest = WellGroup ( dest_origin ) <EOL> opts = [ ] <EOL> oshp = [ ] <EOL> osta = [ ] <EOL> len_source = len ( source . wells ) <EOL> len_dest = len ( dest . wells ) <EOL> if not one_source : <EOL> if len_dest > <NUM_LIT:1> and len_source == <NUM_LIT:1> : <EOL> source = WellGroup ( source . wells * len_dest ) <EOL> len_source = len ( source . wells ) <EOL> if len_dest == <NUM_LIT:1> and len_source > <NUM_LIT:1> : <EOL> dest = WellGroup ( dest . wells * len_source ) <EOL> len_dest = len ( dest . wells ) <EOL> if len_source != len_dest : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( volume , basestring ) or isinstance ( volume , Unit ) : <EOL> if len_dest == <NUM_LIT:1> and not one_source : <EOL> volume = [ Unit . fromstring ( volume ) ] * len_source <EOL> else : <EOL> volume = [ Unit . fromstring ( volume ) ] * len_dest <EOL> elif isinstance ( volume , list ) and len ( volume ) == len_dest : <EOL> volume = list ( map ( lambda x : Unit . fromstring ( x ) , volume ) ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( shape , dict ) : <EOL> if len_dest == <NUM_LIT:1> and not one_source : <EOL> shape = [ shape ] * len_source <EOL> else : <EOL> shape = [ shape ] * len_dest <EOL> elif isinstance ( shape , list ) and len ( shape ) == len_dest : <EOL> shape = shape <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> stamp_type = [ ] <EOL> rows = [ ] <EOL> columns = [ ] <EOL> for s in shape : <EOL> if "<STR_LIT>" not in s or "<STR_LIT>" not in s : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> r = s [ "<STR_LIT>" ] <EOL> c = s [ "<STR_LIT>" ] <EOL> rows . append ( r ) <EOL> columns . append ( c ) <EOL> if c == <NUM_LIT:12> and r == <NUM_LIT:8> : <EOL> stamp_type . append ( "<STR_LIT>" ) <EOL> elif c == <NUM_LIT:12> : <EOL> stamp_type . append ( "<STR_LIT>" ) <EOL> elif r == <NUM_LIT:8> : <EOL> stamp_type . append ( "<STR_LIT>" ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> for s , d , c , r , st in list ( zip ( source . wells , dest . wells , columns , rows , stamp_type ) ) : <EOL> src_col_count = s . container . container_type . col_count <EOL> dest_col_count = d . container . container_type . col_count <EOL> if c < <NUM_LIT:0> or c > src_col_count or c > dest_col_count : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> src_row_count = s . container . container_type . well_count // src_col_count <EOL> dest_row_count = d . container . container_type . well_count // dest_col_count <EOL> if r < <NUM_LIT:0> or r > src_row_count or r > dest_row_count : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> check_valid_origin ( s , st , c , r ) <EOL> check_valid_origin ( d , st , c , r ) <EOL> if one_tip or one_source : <EOL> if not all ( [ s == shape [ <NUM_LIT:0> ] for s in shape ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if one_source : <EOL> try : <EOL> for w , c , r , st in list ( zip ( source . wells , columns , rows , stamp_type ) ) : <EOL> columnWise = False <EOL> if st == "<STR_LIT>" : <EOL> columnWise = True <EOL> if w . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> source_wells = [ w . container . wells_from ( <EOL> w , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = [ w . container . wells_from ( <EOL> w , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = w . container . wells_from ( <EOL> w , c * r , columnWise ) <EOL> if not all ( [ s . volume >= w . volume for s in source_wells ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> source_vol = [ s . volume for s in source . wells ] <EOL> if sum ( [ a for a in volume ] ) > sum ( [ a for a in source_vol ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if len_source >= len_dest and all ( i > j for i , j in zip ( source_vol , volume ) ) : <EOL> sources = source . wells [ : len_dest ] <EOL> destinations = dest . wells <EOL> volumes = volume <EOL> else : <EOL> sources = [ ] <EOL> source_counter = <NUM_LIT:0> <EOL> destinations = [ ] <EOL> volumes = [ ] <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> max_decimal_places = <NUM_LIT:12> <EOL> for idx , d in enumerate ( dest . wells ) : <EOL> vol_d = volume [ idx ] <EOL> while vol_d > Unit . fromstring ( "<STR_LIT>" ) : <EOL> if vol > vol_d : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol_d ) <EOL> vol -= vol_d <EOL> vol . _magnitude = round ( <EOL> vol . _magnitude , max_decimal_places ) <EOL> vol_d -= vol_d <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> else : <EOL> sources . append ( s ) <EOL> destinations . append ( d ) <EOL> volumes . append ( vol ) <EOL> vol_d -= vol <EOL> vol_d . _magnitude = round ( <EOL> vol_d . _magnitude , max_decimal_places ) <EOL> source_counter += <NUM_LIT:1> <EOL> if source_counter < len_source : <EOL> s = source . wells [ source_counter ] <EOL> vol = s . volume <EOL> source = WellGroup ( sources ) <EOL> dest = WellGroup ( destinations ) <EOL> volume = volumes <EOL> shape = [ shape [ <NUM_LIT:0> ] ] * len ( volume ) <EOL> rows = [ rows [ <NUM_LIT:0> ] ] * len ( volume ) <EOL> columns = [ columns [ <NUM_LIT:0> ] ] * len ( volume ) <EOL> stamp_type = [ stamp_type [ <NUM_LIT:0> ] ] * len ( volume ) <EOL> except ( ValueError , AttributeError , TypeError ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> volumeSwitch = Unit . fromstring ( "<STR_LIT>" ) <EOL> if one_tip : <EOL> if not mix_vol : <EOL> temp_vol = Unit . fromstring ( "<STR_LIT>" ) <EOL> else : <EOL> temp_vol = Unit . fromstring ( mix_vol ) <EOL> if not ( all ( [ v > volumeSwitch for v in volume ] ) or all ( [ v <= volumeSwitch for v in volume ] ) or ( temp_vol > volumeSwitch ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> st = stamp_type [ <NUM_LIT:0> ] <EOL> if st == "<STR_LIT>" : <EOL> maxContainers = <NUM_LIT:3> <EOL> else : <EOL> maxContainers = <NUM_LIT:2> <EOL> all_wells = source + dest <EOL> if len ( set ( map ( lambda x : x . container , all_wells . wells ) ) ) > maxContainers : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> tip_capacity = Unit . fromstring ( "<STR_LIT>" ) <EOL> primer_resid = Unit . fromstring ( "<STR_LIT>" ) <EOL> transit_resid = Unit . fromstring ( "<STR_LIT>" ) <EOL> pre_buffer_resid = Unit . fromstring ( "<STR_LIT>" ) <EOL> if aspirate_source : <EOL> if '<STR_LIT>' in aspirate_source . keys ( ) : <EOL> primer_resid = Unit . fromstring ( aspirate_source [ '<STR_LIT>' ] ) <EOL> if pre_buffer : <EOL> pre_buffer_resid = Unit . fromstring ( pre_buffer ) <EOL> if transit_vol : <EOL> transit_resid = Unit . fromstring ( transit_vol ) <EOL> if primer_resid > Unit . fromstring ( transit_resid ) : <EOL> primer_or_transit = primer_resid <EOL> else : <EOL> primer_or_transit = transit_resid <EOL> max_tip_vol = tip_capacity - pre_buffer_resid - primer_or_transit <EOL> for s , d , v , c , r , st , sh in list ( zip ( source . wells , dest . wells , volume , columns , rows , stamp_type , shape ) ) : <EOL> if v > max_tip_vol : <EOL> diff = v <EOL> while diff > max_tip_vol : <EOL> if diff < max_tip_vol * <NUM_LIT:2> : <EOL> diff = diff / <NUM_LIT:2> <EOL> v = diff <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> columnWise = False <EOL> if st == "<STR_LIT>" : <EOL> columnWise = True <EOL> if d . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = d . container . wells_from ( <EOL> d , c * r , columnWise ) <EOL> if s . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = s . container . wells_from ( <EOL> s , c * r , columnWise ) <EOL> for well in source_wells : <EOL> if well . volume : <EOL> well . volume -= v <EOL> for well in dest_wells : <EOL> if well . volume : <EOL> well . volume += v <EOL> else : <EOL> well . volume = v <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( xfer , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( xfer , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if not mix_vol and ( mix_before or mix_after ) : <EOL> mix_vol = v * <NUM_LIT> <EOL> if mix_before : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if mix_after : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if v > Unit ( <NUM_LIT:0> , '<STR_LIT>' ) : <EOL> opts . append ( xfer ) <EOL> oshp . append ( sh ) <EOL> osta . append ( st ) <EOL> else : <EOL> diff -= max_tip_vol <EOL> v = max_tip_vol <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> columnWise = False <EOL> if st == "<STR_LIT>" : <EOL> columnWise = True <EOL> if d . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = d . container . wells_from ( <EOL> d , c * r , columnWise ) <EOL> if s . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = s . container . wells_from ( <EOL> s , c * r , columnWise ) <EOL> for well in source_wells : <EOL> if well . volume : <EOL> well . volume -= v <EOL> for well in dest_wells : <EOL> if well . volume : <EOL> well . volume += v <EOL> else : <EOL> well . volume = v <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( xfer , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( xfer , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if not mix_vol and ( mix_before or mix_after ) : <EOL> mix_vol = v * <NUM_LIT> <EOL> if mix_before : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if mix_after : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if v > Unit ( <NUM_LIT:0> , '<STR_LIT>' ) : <EOL> opts . append ( xfer ) <EOL> oshp . append ( sh ) <EOL> osta . append ( st ) <EOL> v = diff <EOL> xfer = { <EOL> "<STR_LIT>" : s , <EOL> "<STR_LIT:to>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> columnWise = False <EOL> if st == "<STR_LIT>" : <EOL> columnWise = True <EOL> if d . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = [ d . container . wells_from ( <EOL> d , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> dest_wells = d . container . wells_from ( d , c * r , columnWise ) <EOL> if s . container . container_type . col_count == <NUM_LIT> : <EOL> if columnWise : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT:16> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = [ s . container . wells_from ( <EOL> s , c * r * <NUM_LIT:4> , columnWise ) [ x ] for x in range ( c * r * <NUM_LIT:4> ) if ( x % <NUM_LIT:2> ) == ( x // <NUM_LIT> ) % <NUM_LIT:2> == <NUM_LIT:0> ] <EOL> else : <EOL> source_wells = s . container . wells_from ( s , c * r , columnWise ) <EOL> for well in source_wells : <EOL> if well . volume : <EOL> well . volume -= v <EOL> for well in dest_wells : <EOL> if well . volume : <EOL> well . volume += v <EOL> else : <EOL> well . volume = v <EOL> opt_list = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> for option in opt_list : <EOL> assign ( xfer , option , eval ( option ) ) <EOL> x_opt_list = [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> for x_option in x_opt_list : <EOL> assign ( xfer , x_option , eval ( x_option [ <NUM_LIT:2> : ] ) ) <EOL> if not mix_vol and ( mix_before or mix_after ) : <EOL> mix_vol = v * <NUM_LIT> <EOL> if mix_before : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if mix_after : <EOL> xfer [ "<STR_LIT>" ] = { <EOL> "<STR_LIT>" : mix_vol , <EOL> "<STR_LIT>" : repetitions , <EOL> "<STR_LIT>" : flowrate <EOL> } <EOL> if v > Unit ( <NUM_LIT:0> , '<STR_LIT>' ) : <EOL> opts . append ( xfer ) <EOL> oshp . append ( sh ) <EOL> osta . append ( st ) <EOL> trans = { } <EOL> if one_tip : <EOL> trans [ "<STR_LIT>" ] = opts <EOL> assign ( trans , "<STR_LIT>" , oshp [ <NUM_LIT:0> ] ) <EOL> assign ( trans , "<STR_LIT>" , <NUM_LIT> ) <EOL> stamp_type = osta [ <NUM_LIT:0> ] <EOL> if stamp_type == "<STR_LIT>" : <EOL> maxTransfers = <NUM_LIT:4> <EOL> maxContainers = <NUM_LIT:3> <EOL> elif stamp_type == "<STR_LIT>" : <EOL> maxTransfers = <NUM_LIT:12> <EOL> maxContainers = <NUM_LIT:2> <EOL> else : <EOL> maxTransfers = <NUM_LIT:8> <EOL> maxContainers = <NUM_LIT:3> <EOL> if new_group : <EOL> self . instructions . append ( Stamp ( [ trans ] ) ) <EOL> elif ( len ( self . instructions ) > <NUM_LIT:0> and self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" and check_stamp_append ( trans , self . instructions [ - <NUM_LIT:1> ] . groups , maxTransfers , maxContainers , volumeSwitch ) ) : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups . append ( trans ) <EOL> else : <EOL> self . instructions . append ( Stamp ( [ trans ] ) ) <EOL> else : <EOL> for x , y , z in list ( zip ( opts , oshp , osta ) ) : <EOL> trans = { } <EOL> trans [ "<STR_LIT>" ] = [ x ] <EOL> assign ( trans , "<STR_LIT>" , y ) <EOL> assign ( trans , "<STR_LIT>" , <NUM_LIT> ) <EOL> stamp_type = z <EOL> if stamp_type == "<STR_LIT>" : <EOL> maxTransfers = <NUM_LIT:4> <EOL> maxContainers = <NUM_LIT:3> <EOL> elif stamp_type == "<STR_LIT>" : <EOL> maxTransfers = <NUM_LIT:12> <EOL> maxContainers = <NUM_LIT:2> <EOL> else : <EOL> maxTransfers = <NUM_LIT:8> <EOL> maxContainers = <NUM_LIT:3> <EOL> if new_group : <EOL> self . instructions . append ( Stamp ( [ trans ] ) ) <EOL> elif ( len ( self . instructions ) > <NUM_LIT:0> and self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" and check_stamp_append ( trans , self . instructions [ - <NUM_LIT:1> ] . groups , maxTransfers , maxContainers , volumeSwitch ) ) : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups . append ( trans ) <EOL> else : <EOL> self . instructions . append ( Stamp ( [ trans ] ) ) <EOL> def sangerseq ( self , cont , wells , dataref , type = "<STR_LIT>" , primer = None ) : <EOL> """<STR_LIT>""" <EOL> type = type . lower ( ) <EOL> if type == "<STR_LIT>" and not primer : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( wells , WellGroup ) : <EOL> container = set ( [ w . container for w in wells ] ) <EOL> if len ( container ) > <NUM_LIT:1> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> wells = [ str ( w . index ) for w in wells ] <EOL> if not isinstance ( wells , list ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . instructions . append ( SangerSeq ( cont , wells , dataref , type , primer ) ) <EOL> def mix ( self , well , volume = "<STR_LIT>" , speed = "<STR_LIT>" , <EOL> repetitions = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( well , ( Well , basestring , WellGroup , list ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( well , ( Well , basestring ) ) : <EOL> well = WellGroup ( [ well ] ) <EOL> if isinstance ( well , list ) : <EOL> well = WellGroup ( well ) <EOL> for w in well . wells : <EOL> opts = { <EOL> "<STR_LIT>" : w , <EOL> "<STR_LIT>" : volume , <EOL> "<STR_LIT>" : speed , <EOL> "<STR_LIT>" : repetitions <EOL> } <EOL> self . _pipette ( [ { "<STR_LIT>" : [ opts ] } ] ) <EOL> def dispense ( self , ref , reagent , columns , speed_percentage = None ) : <EOL> """<STR_LIT>""" <EOL> if ( speed_percentage != None and <EOL> ( speed_percentage > <NUM_LIT:100> or speed_percentage < <NUM_LIT:1> ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if not isinstance ( columns , list ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> for c in columns : <EOL> wells = WellGroup ( ref . wells_from ( c [ "<STR_LIT>" ] , ref . container_type . row_count ( ) , <EOL> columnwise = True ) ) <EOL> for w in wells : <EOL> if w . volume : <EOL> w . volume += Unit . fromstring ( c [ "<STR_LIT>" ] ) <EOL> else : <EOL> w . set_volume ( c [ "<STR_LIT>" ] ) <EOL> self . instructions . append ( <EOL> Dispense ( ref , reagent , columns , speed_percentage ) ) <EOL> def dispense_full_plate ( self , ref , reagent , volume , speed_percentage = None ) : <EOL> """<STR_LIT>""" <EOL> if ( speed_percentage != None and <EOL> ( speed_percentage > <NUM_LIT:100> or speed_percentage < <NUM_LIT:1> ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> columns = [ ] <EOL> for col in range ( <NUM_LIT:0> , ref . container_type . col_count ) : <EOL> columns . append ( { "<STR_LIT>" : col , "<STR_LIT>" : volume } ) <EOL> self . dispense ( ref , reagent , columns , speed_percentage ) <EOL> def spin ( self , ref , acceleration , duration ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Spin ( ref , acceleration , duration ) ) <EOL> def thermocycle ( self , ref , groups , <EOL> volume = "<STR_LIT>" , <EOL> dataref = None , <EOL> dyes = None , <EOL> melting_start = None , <EOL> melting_end = None , <EOL> melting_increment = None , <EOL> melting_rate = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( groups , list ) : <EOL> raise AttributeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . instructions . append ( <EOL> Thermocycle ( ref , groups , volume , dataref , dyes , melting_start , <EOL> melting_end , melting_increment , melting_rate ) ) <EOL> def incubate ( self , ref , where , duration , shaking = False , co2 = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Incubate ( ref , where , duration , shaking , co2 ) ) <EOL> def absorbance ( self , ref , wells , wavelength , dataref , flashes = <NUM_LIT> , <EOL> incubate_before = None , temperature = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( wells , Well ) : <EOL> wells = WellGroup ( wells ) <EOL> if isinstance ( wells , WellGroup ) : <EOL> wells = wells . indices ( ) <EOL> if incubate_before : <EOL> check_valid_incubate_params ( incubate_before ) <EOL> self . instructions . append ( <EOL> Absorbance ( ref , wells , wavelength , dataref , flashes , <EOL> incubate_before , temperature ) ) <EOL> def fluorescence ( self , ref , wells , excitation , emission , dataref , <EOL> flashes = <NUM_LIT> , temperature = None , gain = None , <EOL> incubate_before = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( wells , Well ) : <EOL> wells = WellGroup ( wells ) <EOL> if isinstance ( wells , WellGroup ) : <EOL> wells = wells . indices ( ) <EOL> if gain is not None and not ( <NUM_LIT:0> <= gain <= <NUM_LIT:1> ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % gain ) <EOL> if incubate_before : <EOL> check_valid_incubate_params ( incubate_before ) <EOL> self . instructions . append ( <EOL> Fluorescence ( ref , wells , excitation , emission , dataref , flashes , <EOL> incubate_before , temperature , gain ) ) <EOL> def luminescence ( self , ref , wells , dataref , incubate_before = None , <EOL> temperature = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( wells , Well ) : <EOL> wells = WellGroup ( wells ) <EOL> if isinstance ( wells , WellGroup ) : <EOL> wells = wells . indices ( ) <EOL> if incubate_before : <EOL> check_valid_incubate_params ( incubate_before ) <EOL> self . instructions . append ( Luminescence ( ref , wells , dataref , <EOL> incubate_before , temperature ) ) <EOL> def gel_separate ( self , wells , volume , matrix , ladder , duration , dataref ) : <EOL> """<STR_LIT>""" <EOL> max_well = int ( matrix . split ( "<STR_LIT:(>" , <NUM_LIT:1> ) [ <NUM_LIT:1> ] . split ( "<STR_LIT:U+002C>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] ) <EOL> if len ( wells ) > max_well : <EOL> datarefs = <NUM_LIT:1> <EOL> for x in xrange ( <NUM_LIT:0> , len ( wells ) , max_well ) : <EOL> self . gel_separate ( wells [ x : x + max_well ] , volume , matrix , ladder , <EOL> duration , "<STR_LIT>" % ( dataref , datarefs ) ) <EOL> datarefs += <NUM_LIT:1> <EOL> else : <EOL> self . instructions . append ( GelSeparate ( wells , volume , matrix , ladder , <EOL> duration , dataref ) ) <EOL> def gel_purify ( self , extracts , volume , matrix , ladder , dataref ) : <EOL> """<STR_LIT>""" <EOL> from itertools import groupby <EOL> from operator import itemgetter <EOL> try : <EOL> max_well = int ( matrix . split ( "<STR_LIT:(>" , <NUM_LIT:1> ) [ <NUM_LIT:1> ] . split ( "<STR_LIT:U+002C>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] ) <EOL> except ( AttributeError , IndexError ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> volume = Unit ( volume ) <EOL> if volume <= Unit ( "<STR_LIT>" ) : <EOL> raise ValueError ( "<STR_LIT>" % volume ) <EOL> if not isinstance ( ladder , str ) : <EOL> raise TypeError ( "<STR_LIT>" % ladder ) <EOL> if not isinstance ( dataref , str ) : <EOL> raise TypeError ( "<STR_LIT>" % dataref ) <EOL> if not isinstance ( extracts , list ) : <EOL> extracts = [ extracts ] <EOL> for e in extracts : <EOL> check_valid_gel_purify_extract ( e ) <EOL> gel_set = [ e [ "<STR_LIT>" ] for e in extracts ] <EOL> lane_set = [ e [ "<STR_LIT>" ] for e in extracts ] <EOL> if None in gel_set : <EOL> if any ( gel_set ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> else : <EOL> if None in lane_set : <EOL> if any ( lane_set ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> else : <EOL> for i , e in enumerate ( extracts ) : <EOL> e [ "<STR_LIT>" ] = i // max_well <EOL> e [ "<STR_LIT>" ] = i % max_well <EOL> else : <EOL> for e in extracts : <EOL> e [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> sort_key = itemgetter ( "<STR_LIT>" ) <EOL> parsed_extracts = [ list ( grp ) for key , grp in groupby ( sorted ( extracts , key = sort_key ) , key = sort_key ) ] <EOL> for pe in parsed_extracts : <EOL> lane_set = [ e [ "<STR_LIT>" ] for e in pe ] <EOL> if len ( lane_set ) > max_well : <EOL> raise RuntimeError ( "<STR_LIT>" % ( max_well , len ( lane_set ) ) ) <EOL> if None in lane_set : <EOL> if any ( lane_set ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> else : <EOL> for i , e in enumerate ( pe ) : <EOL> e [ "<STR_LIT>" ] = i % max_well <EOL> lane_set = [ e [ "<STR_LIT>" ] for e in pe ] <EOL> if sorted ( lane_set ) != list ( range ( <NUM_LIT:0> , max ( lane_set ) + <NUM_LIT:1> ) ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if len ( parsed_extracts ) > <NUM_LIT:1> : <EOL> dataref_gel = "<STR_LIT>" % ( dataref , pe [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) <EOL> else : <EOL> dataref_gel = dataref <EOL> pe = sorted ( pe , key = itemgetter ( "<STR_LIT>" ) ) <EOL> samples = [ e [ "<STR_LIT:source>" ] for e in pe ] <EOL> pe_unpacked = [ ] <EOL> for e in pe : <EOL> for b in e [ "<STR_LIT>" ] : <EOL> ext = b <EOL> ext [ "<STR_LIT>" ] = e [ "<STR_LIT>" ] <EOL> pe_unpacked . append ( ext ) <EOL> self . instructions . append ( GelPurify ( samples , volume , matrix , ladder , dataref_gel , pe_unpacked ) ) <EOL> def seal ( self , ref , type = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Seal ( ref , type ) ) <EOL> def unseal ( self , ref ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Unseal ( ref ) ) <EOL> def cover ( self , ref , lid = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Cover ( ref , lid ) ) <EOL> def uncover ( self , ref ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Uncover ( ref ) ) <EOL> def flow_analyze ( self , dataref , FSC , SSC , neg_controls , samples , <EOL> colors = None , pos_controls = None ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( FlowAnalyze ( dataref , FSC , SSC , neg_controls , <EOL> samples , colors , pos_controls ) ) <EOL> def oligosynthesize ( self , oligos ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( Oligosynthesize ( oligos ) ) <EOL> def spread ( self , source , dest , volume ) : <EOL> """<STR_LIT>""" <EOL> volume = Unit . fromstring ( volume ) <EOL> if dest . volume : <EOL> dest . volume += volume <EOL> else : <EOL> dest . volume = volume <EOL> if source . volume : <EOL> source . volume -= volume <EOL> self . instructions . append ( Spread ( source , dest , volume ) ) <EOL> def autopick ( self , sources , dests , min_abort = <NUM_LIT:0> , criteria = { } , <EOL> dataref = "<STR_LIT>" , newpick = False ) : <EOL> """<STR_LIT>""" <EOL> pick = { } <EOL> if isinstance ( sources , Well ) or isinstance ( sources , str ) : <EOL> sources = [ sources ] <EOL> pick [ "<STR_LIT>" ] = sources <EOL> if len ( set ( [ s . container for s in pick [ "<STR_LIT>" ] ] ) ) > <NUM_LIT:1> : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( dests , Well ) or isinstance ( dests , str ) : <EOL> dests = [ dests ] <EOL> pick [ "<STR_LIT:to>" ] = dests <EOL> pick [ "<STR_LIT>" ] = min_abort <EOL> group = [ pick ] <EOL> if ( not newpick and self . instructions and <EOL> self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" and <EOL> self . instructions [ - <NUM_LIT:1> ] . dataref == dataref and <EOL> self . instructions [ - <NUM_LIT:1> ] . criteria == criteria and <EOL> self . instructions [ - <NUM_LIT:1> ] . groups [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . container == <EOL> sources [ <NUM_LIT:0> ] . container ) : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups . extend ( group ) <EOL> else : <EOL> self . instructions . append ( Autopick ( group , criteria , dataref ) ) <EOL> def mag_dry ( self , head , container , duration , new_tip = False , new_instruction = False ) : <EOL> """<STR_LIT>""" <EOL> check_valid_mag ( container , head ) <EOL> mag = { } <EOL> mag [ "<STR_LIT:object>" ] = container <EOL> for kw in [ "<STR_LIT>" ] : <EOL> val = locals ( ) [ kw ] <EOL> if isinstance ( val , str ) and "<STR_LIT::>" in val : <EOL> val = Unit ( val ) <EOL> mag [ kw ] = val <EOL> check_valid_mag_params ( mag ) <EOL> self . _add_mag ( mag , head , new_tip , new_instruction , "<STR_LIT>" ) <EOL> def mag_incubate ( self , head , container , duration , magnetize = False , tip_position = <NUM_LIT> , temperature = None , new_tip = False , new_instruction = False ) : <EOL> """<STR_LIT>""" <EOL> check_valid_mag ( container , head ) <EOL> mag = { } <EOL> mag [ "<STR_LIT:object>" ] = container <EOL> for kw in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> val = locals ( ) [ kw ] <EOL> if isinstance ( val , str ) and "<STR_LIT::>" in val : <EOL> val = Unit ( val ) <EOL> mag [ kw ] = val <EOL> check_valid_mag_params ( mag ) <EOL> self . _add_mag ( mag , head , new_tip , new_instruction , "<STR_LIT>" ) <EOL> def mag_collect ( self , head , container , cycles , pause_duration , bottom_position = <NUM_LIT:0.0> , temperature = None , new_tip = False , new_instruction = False ) : <EOL> """<STR_LIT>""" <EOL> check_valid_mag ( container , head ) <EOL> mag = { } <EOL> mag [ "<STR_LIT:object>" ] = container <EOL> for kw in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> val = locals ( ) [ kw ] <EOL> if isinstance ( val , str ) and "<STR_LIT::>" in val : <EOL> val = Unit ( val ) <EOL> mag [ kw ] = val <EOL> check_valid_mag_params ( mag ) <EOL> self . _add_mag ( mag , head , new_tip , new_instruction , "<STR_LIT>" ) <EOL> def mag_release ( self , head , container , duration , frequency , center = <NUM_LIT:0.5> , amplitude = <NUM_LIT:0.5> , temperature = None , new_tip = False , new_instruction = False ) : <EOL> """<STR_LIT>""" <EOL> check_valid_mag ( container , head ) <EOL> mag = { } <EOL> mag [ "<STR_LIT:object>" ] = container <EOL> for kw in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> val = locals ( ) [ kw ] <EOL> if isinstance ( val , str ) and "<STR_LIT::>" in val : <EOL> val = Unit ( val ) <EOL> mag [ kw ] = val <EOL> check_valid_mag_params ( mag ) <EOL> self . _add_mag ( mag , head , new_tip , new_instruction , "<STR_LIT>" ) <EOL> def mag_mix ( self , head , container , duration , frequency , center = <NUM_LIT:0.5> , amplitude = <NUM_LIT:0.5> , magnetize = False , temperature = None , new_tip = False , new_instruction = False ) : <EOL> """<STR_LIT>""" <EOL> check_valid_mag ( container , head ) <EOL> mag = { } <EOL> mag [ "<STR_LIT:object>" ] = container <EOL> for kw in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> val = locals ( ) [ kw ] <EOL> if isinstance ( val , str ) and "<STR_LIT::>" in val : <EOL> val = Unit ( val ) <EOL> mag [ kw ] = val <EOL> check_valid_mag_params ( mag ) <EOL> self . _add_mag ( mag , head , new_tip , new_instruction , "<STR_LIT>" ) <EOL> def image_plate ( self , ref , mode , dataref ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( ImagePlate ( ref , mode , dataref ) ) <EOL> def provision ( self , resource_id , dests , volumes ) : <EOL> """<STR_LIT>""" <EOL> dests = WellGroup ( dests ) <EOL> if not isinstance ( resource_id , basestring ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if not isinstance ( volumes , list ) : <EOL> volumes = [ Unit . fromstring ( volumes ) ] * len ( dests ) <EOL> else : <EOL> if len ( volumes ) != len ( dests ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> volumes = [ Unit . fromstring ( v ) for v in volumes ] <EOL> for v in volumes : <EOL> if not isinstance ( v , ( basestring , Unit ) ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> for d , v in zip ( dests , volumes ) : <EOL> dest_group = [ ] <EOL> if v > Unit ( <NUM_LIT> , "<STR_LIT>" ) : <EOL> diff = v - Unit ( <NUM_LIT> , "<STR_LIT>" ) <EOL> self . provision ( resource_id , d , Unit ( <NUM_LIT> , "<STR_LIT>" ) ) <EOL> while diff > Unit ( <NUM_LIT:0.0> , "<STR_LIT>" ) : <EOL> self . provision ( resource_id , d , diff ) <EOL> diff -= diff <EOL> continue <EOL> xfer = { } <EOL> xfer [ "<STR_LIT>" ] = d <EOL> xfer [ "<STR_LIT>" ] = v <EOL> if d . volume : <EOL> d . volume += v <EOL> else : <EOL> d . set_volume ( v ) <EOL> dest_group . append ( xfer ) <EOL> if ( self . instructions and self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" and <EOL> self . instructions [ - <NUM_LIT:1> ] . resource_id == resource_id and <EOL> self . instructions [ - <NUM_LIT:1> ] . to [ - <NUM_LIT:1> ] [ "<STR_LIT>" ] . container == d . container ) : <EOL> self . instructions [ - <NUM_LIT:1> ] . to . append ( xfer ) <EOL> else : <EOL> self . instructions . append ( Provision ( resource_id , dest_group ) ) <EOL> def flash_freeze ( self , container , duration ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( FlashFreeze ( container , duration ) ) <EOL> def _ref_for_well ( self , well ) : <EOL> return "<STR_LIT>" % ( self . _ref_for_container ( well . container ) , well . index ) <EOL> def _ref_for_container ( self , container ) : <EOL> for k in self . refs : <EOL> v = self . refs [ k ] <EOL> if v . container is container : <EOL> return k <EOL> @ staticmethod <EOL> def fill_wells ( dst_group , src_group , volume , distribute_target ) : <EOL> """<STR_LIT>""" <EOL> src = None <EOL> distributes = [ ] <EOL> src_group = WellGroup ( src_group ) <EOL> dst_group = WellGroup ( dst_group ) <EOL> if isinstance ( volume , list ) : <EOL> if len ( volume ) != len ( dst_group . wells ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> volume = [ Unit . fromstring ( x ) for x in volume ] <EOL> else : <EOL> volume = [ Unit . fromstring ( volume ) ] * len ( dst_group . wells ) <EOL> for d , v in list ( zip ( dst_group . wells , volume ) ) : <EOL> v = v . to ( "<STR_LIT>" ) <EOL> if len ( distributes ) == <NUM_LIT:0> or src . volume < v : <EOL> src = next ( <EOL> ( w for w in src_group . wells if w . volume >= v ) , None ) <EOL> if src is None : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" % <EOL> ( str ( v ) . rsplit ( "<STR_LIT::>" ) [ <NUM_LIT:0> ] , str ( v ) . rsplit ( "<STR_LIT::>" ) [ <NUM_LIT:1> ] ) ) <EOL> distributes . append ( { <EOL> "<STR_LIT>" : src , <EOL> "<STR_LIT:to>" : [ ] <EOL> } ) <EOL> opts = { <EOL> "<STR_LIT>" : d , <EOL> "<STR_LIT>" : v <EOL> } <EOL> if distribute_target : <EOL> opts [ "<STR_LIT>" ] = distribute_target <EOL> distributes [ - <NUM_LIT:1> ] [ "<STR_LIT:to>" ] . append ( opts ) <EOL> src . volume -= v <EOL> if d . volume : <EOL> d . volume += v <EOL> else : <EOL> d . volume = v <EOL> return distributes <EOL> def _pipette ( self , groups ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . instructions ) > <NUM_LIT:0> and self . instructions [ - <NUM_LIT:1> ] . op == '<STR_LIT>' : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups += groups <EOL> else : <EOL> self . instructions . append ( Pipette ( groups ) ) <EOL> def _add_mag ( self , mag , head , new_tip , new_instruction , name ) : <EOL> """<STR_LIT>""" <EOL> max_containers = <NUM_LIT:8> <EOL> if ( not new_instruction and self . instructions and <EOL> self . instructions [ - <NUM_LIT:1> ] . op == "<STR_LIT>" and <EOL> self . instructions [ - <NUM_LIT:1> ] . magnetic_head == head ) : <EOL> if self . _count_mag_containers ( mag , new_tip ) <= max_containers : <EOL> if not new_tip : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups [ - <NUM_LIT:1> ] . append ( { name : mag } ) <EOL> elif new_tip : <EOL> self . instructions [ - <NUM_LIT:1> ] . groups . append ( [ { name : mag } ] ) <EOL> elif self . _count_mag_containers ( mag , new_tip ) > max_containers : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> self . instructions . append ( MagneticTransfer ( [ [ { name : mag } ] ] , head ) ) <EOL> def _count_mag_containers ( self , mag , new_tip ) : <EOL> """<STR_LIT>""" <EOL> num_groups = len ( self . instructions [ - <NUM_LIT:1> ] . groups ) <EOL> containers = [ list ( g . values ( ) ) [ <NUM_LIT:0> ] [ "<STR_LIT:object>" ] for group in self . instructions [ - <NUM_LIT:1> ] . groups for g in group ] <EOL> containers . append ( mag [ "<STR_LIT:object>" ] ) <EOL> containers = list ( set ( containers ) ) <EOL> num_containers = len ( containers ) <EOL> num_containers += num_groups <EOL> if new_tip : <EOL> num_containers += <NUM_LIT:1> <EOL> return num_containers <EOL> def _refify ( self , op_data ) : <EOL> if type ( op_data ) is dict : <EOL> return { k : self . _refify ( v ) for k , v in op_data . items ( ) } <EOL> elif type ( op_data ) is list : <EOL> return [ self . _refify ( i ) for i in op_data ] <EOL> elif isinstance ( op_data , Well ) : <EOL> return self . _ref_for_well ( op_data ) <EOL> elif isinstance ( op_data , WellGroup ) : <EOL> return [ self . _ref_for_well ( w ) for w in op_data . wells ] <EOL> elif isinstance ( op_data , Container ) : <EOL> return self . _ref_for_container ( op_data ) <EOL> elif isinstance ( op_data , Unit ) : <EOL> return str ( op_data ) <EOL> else : <EOL> return op_data <EOL> def _ref_containers_and_wells ( self , params ) : <EOL> """<STR_LIT>""" <EOL> parameters = { } <EOL> containers = { } <EOL> for k , v in params . items ( ) : <EOL> if isinstance ( v , dict ) : <EOL> parameters [ str ( k ) ] = self . _ref_containers_and_wells ( v ) <EOL> if isinstance ( v , list ) and isinstance ( v [ <NUM_LIT:0> ] , dict ) : <EOL> for cont in v : <EOL> self . _ref_containers_and_wells ( cont . encode ( '<STR_LIT:utf-8>' ) ) <EOL> elif isinstance ( v , dict ) and "<STR_LIT:type>" in v : <EOL> if "<STR_LIT>" in v : <EOL> discard = v [ "<STR_LIT>" ] <EOL> if discard and v . get ( "<STR_LIT>" ) : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> discard = False <EOL> containers [ str ( k ) ] = self . ref ( k , v [ "<STR_LIT:id>" ] , v [ "<STR_LIT:type>" ] , storage = v . get ( "<STR_LIT>" ) , <EOL> discard = discard ) <EOL> else : <EOL> parameters [ str ( k ) ] = v <EOL> parameters [ "<STR_LIT>" ] = containers <EOL> for k , v in params . items ( ) : <EOL> if isinstance ( v , list ) and "<STR_LIT:/>" in str ( v [ <NUM_LIT:0> ] ) : <EOL> group = WellGroup ( [ ] ) <EOL> for w in v : <EOL> cont , well = w . rsplit ( "<STR_LIT:/>" , <NUM_LIT:1> ) <EOL> group . append ( self . refs [ cont ] . container . well ( well ) ) <EOL> parameters [ str ( k ) ] = group <EOL> elif "<STR_LIT:/>" in str ( v ) : <EOL> ref_name = v . rsplit ( "<STR_LIT:/>" ) [ <NUM_LIT:0> ] <EOL> if not ref_name in self . refs : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ref_name ) <EOL> if v . rsplit ( "<STR_LIT:/>" ) [ <NUM_LIT:1> ] == "<STR_LIT>" : <EOL> parameters [ str ( k ) ] = self . refs [ <EOL> ref_name ] . container . all_wells ( ) <EOL> else : <EOL> parameters [ str ( k ) ] = self . refs [ ref_name ] . container . well ( <EOL> v . rsplit ( "<STR_LIT:/>" ) [ <NUM_LIT:1> ] ) <EOL> else : <EOL> parameters [ str ( k ) ] = v <EOL> return parameters <EOL> def measure_concentration ( self , wells , dataref , measurement , <EOL> volume = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> self . instructions . append ( MeasureConcentration ( wells , volume , dataref , <EOL> measurement ) ) <EOL> def measure_mass ( self , containers , dataref ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( containers , Container ) : <EOL> containers = [ containers ] <EOL> elif isinstance ( containers , list ) : <EOL> for container in containers : <EOL> if not isinstance ( container , Container ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . instructions . append ( MeasureMass ( containers , dataref ) ) <EOL> def measure_volume ( self , wells , dataref ) : <EOL> """<STR_LIT>""" <EOL> wells = WellGroup ( wells ) <EOL> self . instructions . append ( MeasureVolume ( wells , dataref ) ) </s>
