Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
}
if "mix_after" in mix_kwargs:
xfer["mix_after"] = {
"volume": mix_kwargs.get("mix_vol_a") or mix_kwargs.get("mix_vol") or v/2,
"repetitions": mix_kwargs.get("repetitions_a") or mix_kwargs.get("repetitions") or 10,
"speed": mix_kwargs.get("flowrate_a") or mix_kwargs.get("flowrate") or "100:microliter/second"
}
# Append transfer options
opt_list = ["aspirate_speed", "dispense_speed"]
for option in opt_list:
assign(xfer, option, eval(option))
x_opt_list = ["x_aspirate_source", "x_dispense_target",
"x_pre_buffer", "x_disposal_vol", "x_transit_vol",
"x_blowout_buffer"]
for x_option in x_opt_list:
assign(xfer, x_option, eval(x_option[2:]))
if v.value > 0:
opts.append(xfer)
diff -= Unit(750, "microliter")
v = diff
# Organize transfer options into dictionary (for json parsing)
xfer = {
"from": s,
"to": d,
"volume": v
}
# Volume accounting
if d.volume:
d.volume += v
if isinstance(sources, (Well, basestring)):
sources = [sources]
if isinstance(volumes, list):
if len(volumes) != len(sources):
raise ValueError("If supplying consolidate "
"volumes as a list, its length "
"must match the number of "
"source wells specified.")
volumes = [Unit.fromstring(v) for v in volumes]
else:
volumes = [Unit.fromstring(volumes)] * len(sources)
# Initialize instructions
cons = {"consolidate": {}}
cons_instr = cons["consolidate"]
assign(cons_instr, "to", dest)
from_wells = []
# Generate instructions for each transfer from source wells
for s, v in zip(sources, volumes):
source_opts = {}
source_opts["well"] = s
source_opts["volume"] = v
assign(source_opts, "aspirate_speed", aspirate_speed)
assign(source_opts, "x_aspirate_source", aspirate_source)
from_wells.append(source_opts)
if dest.volume:
dest.volume += v
else:
dest.volume = v
if s.volume:
s.volume -= v
assign(cons_instr, "from", from_wells)
volumes = [Unit.fromstring(v) for v in volumes]
else:
volumes = [Unit.fromstring(volumes)] * len(sources)
# Initialize instructions
cons = {"consolidate": {}}
cons_instr = cons["consolidate"]
assign(cons_instr, "to", dest)
from_wells = []
# Generate instructions for each transfer from source wells
for s, v in zip(sources, volumes):
source_opts = {}
source_opts["well"] = s
source_opts["volume"] = v
assign(source_opts, "aspirate_speed", aspirate_speed)
assign(source_opts, "x_aspirate_source", aspirate_source)
from_wells.append(source_opts)
if dest.volume:
dest.volume += v
else:
dest.volume = v
if s.volume:
s.volume -= v
assign(cons_instr, "from", from_wells)
# Append mix options
if mix_after:
cons_instr["mix_after"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
# Append transfer options
if "mix_before" in mix_kwargs:
xfer["mix_before"] = {
"volume": mix_kwargs.get("mix_vol_b") or mix_kwargs.get("mix_vol") or v/2,
"repetitions": mix_kwargs.get("repetitions_b") or mix_kwargs.get("repetitions") or 10,
"speed": mix_kwargs.get("flowrate_b") or mix_kwargs.get("flowrate") or "100:microliter/second"
}
if "mix_after" in mix_kwargs:
xfer["mix_after"] = {
"volume": mix_kwargs.get("mix_vol_a") or mix_kwargs.get("mix_vol") or v/2,
"repetitions": mix_kwargs.get("repetitions_a") or mix_kwargs.get("repetitions") or 10,
"speed": mix_kwargs.get("flowrate_a") or mix_kwargs.get("flowrate") or "100:microliter/second"
}
# Append transfer options
opt_list = ["aspirate_speed", "dispense_speed"]
for option in opt_list:
assign(xfer, option, eval(option))
x_opt_list = ["x_aspirate_source", "x_dispense_target",
"x_pre_buffer", "x_disposal_vol", "x_transit_vol",
"x_blowout_buffer"]
for x_option in x_opt_list:
assign(xfer, x_option, eval(x_option[2:]))
if v.value > 0:
opts.append(xfer)
trans = {}
assign(trans, "x_tip_type", tip_type)
if one_tip:
trans["transfer"] = opts
if new_group:
self.append(Pipette([trans]))
else:
self._pipette([trans])
assign(cons_instr, "from", from_wells)
# Append mix options
if mix_after:
cons_instr["mix_after"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
# Append transfer options
opt_list = ["allow_carryover", "dispense_speed"]
for option in opt_list:
assign(cons_instr, option, eval(option))
x_opt_list = ["x_dispense_target", "x_pre_buffer",
"x_transit_vol", "x_blowout_buffer", "x_tip_type"]
for x_option in x_opt_list:
assign(cons_instr, x_option, eval(x_option[2:]))
# Create new pipette instruction group if necessary
if new_group:
self.append(Pipette([cons]))
else:
self._pipette([cons])
else:
dest.volume = v
if s.volume:
s.volume -= v
assign(cons_instr, "from", from_wells)
# Append mix options
if mix_after:
cons_instr["mix_after"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
# Append transfer options
opt_list = ["allow_carryover", "dispense_speed"]
for option in opt_list:
assign(cons_instr, option, eval(option))
x_opt_list = ["x_dispense_target", "x_pre_buffer",
"x_transit_vol", "x_blowout_buffer", "x_tip_type"]
for x_option in x_opt_list:
assign(cons_instr, x_option, eval(x_option[2:]))
# Create new pipette instruction group if necessary
if new_group:
self.append(Pipette([cons]))
else:
self._pipette([cons])
# Initializing transfer dictionary
xfer = {}
xfer["to"] = dest_origin
xfer["from"] = source_origin
xfer["volume"] = Unit.fromstring(volume)
# Adding liquid transfer options
opt_list = ["aspirate_speed", "dispense_speed"]
for option in opt_list:
assign(xfer, option, eval(option))
x_opt_list = ["x_aspirate_source", "x_dispense_target",
"x_pre_buffer", "x_disposal_vol", "x_transit_vol",
"x_blowout_buffer"]
for x_option in x_opt_list:
assign(xfer, x_option, eval(x_option[2:]))
if not mix_vol and (mix_before or mix_after):
mix_vol = volume * .5
if mix_before:
xfer["mix_before"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
if mix_after:
xfer["mix_after"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
# Volume checking
well.volume = Unit.fromstring(volume)
# Set maximum parameters which are defined due to TCLE limitations
maxContainers = 3
if stamp_type == "full":
maxTransfers = 4
elif stamp_type == "col":
maxTransfers = 12
else:
maxTransfers = 8
# Set volume at which tip volume type changes defined by TCLE - hardcoded for the two current tip volume types
volumeSwitch = Unit.fromstring("31:microliter")
trans = {}
assign(trans, "shape", shape)
assign(trans, "tip_layout", 96)
trans["transfer"] = xfer
# Chunk transfers if there is a previous stamp instruction and if its
# valid to append to an existing instruction
if (len(self.instructions) > 0 and
self.instructions[-1].op == "stamp" and
check_stamp_append(trans, self.instructions[-1].groups.transfer,
maxTransfers, maxContainers, volumeSwitch)):
# Append to existing instruction
self.instructions[-1].groups.transfer.append(trans)
else:
# Initialize new stamp list/instruction
self.instructions.append(Stamp([trans]))
raise RuntimeError("No mix volume specified for "
"mix_before.")
opts["mix_before"] = {
"volume": mix_vol,
"repetitions": repetitions,
"speed": flowrate
}
if allow_carryover:
opts["allow_carryover"] = allow_carryover
opts["from"] = d["from"]
opts["to"] = d["to"]
# Append transfer options
opt_list = ["aspirate_speed", "allow_carryover"]
for option in opt_list:
assign(opts, option, eval(option))
x_opt_list = ["x_aspirate_source", "x_pre_buffer",
"x_disposal_vol", "x_transit_vol",
"x_blowout_buffer", "x_tip_type"]
for x_option in x_opt_list:
assign(opts, x_option, eval(x_option[2:]))
groups.append({"distribute": opts})
if new_group:
self.append(Pipette(groups))
else:
self._pipette(groups)
assign(xfer, x_option, eval(x_option[2:]))
if v.value > 0:
opts.append(xfer)
trans = {}
assign(trans, "x_tip_type", tip_type)
if one_tip:
trans["transfer"] = opts
if new_group:
self.append(Pipette([trans]))
else:
self._pipette([trans])
else:
for x in opts:
trans = {}
assign(trans, "x_tip_type", tip_type)
trans["transfer"] = [x]
if new_group:
self.append(Pipette([trans]))
else:
self._pipette([trans])