Pipette transfer

.gitignore

@@ -80,9 +80,9 @@ sample_protocol.py
 
 # Local Calibration Data
 calibrations/
-*/calibrations/
-smoothie-config.ini
-*/smoothie-config.ini
+smoothie/
+*/smoothie
+*/calibrations
 
 # SDK logs
 *.log

opentrons/containers/placeable.py

@@ -86,7 +86,9 @@ def __getitem__(self, name):
  Returns placeable by name or index
  If slice is given, returns a list
  """
- if isinstance(name, int) or isinstance(name, slice):
+ if isinstance(name, slice):
+ return self.get_children_from_slice(name)
+ elif isinstance(name, int):
  return self.get_children_list()[name]
  elif isinstance(name, str):
  return self.get_child_by_name(name)
@@ -106,10 +108,10 @@ def __str__(self):
  self.__class__.__name__, self.get_name())
 
  def __iter__(self):
- return iter(self.children_by_reference.keys())
+ return iter(self.get_children_list())
 
  def __len__(self):
- return len(self.children_by_name)
+ return len(self.get_children_list())
 
  def __bool__(self):
  return True
@@ -236,6 +238,25 @@ def get_child_by_name(self, name):
  """
  return self.children_by_name.get(name)
 
+ def get_index_from_name(self, name):
+ """
+ Retrieves child's name by index
+ """
+ return self.get_children_list().index(
+ self.get_child_by_name(name))
+
+ def get_children_from_slice(self, s):
+ """
+ Retrieves list of children within slice
+ """
+ if isinstance(s.start, str):
+ s = slice(
+ self.get_index_from_name(s.start), s.stop, s.step)
+ if isinstance(s.stop, str):
+ s = slice(
+ s.start, self.get_index_from_name(s.stop), s.step)
+ return WellSeries(self.get_children_list()[s])
+
  def has_children(self):
  """
  Returns *True* if :Placeable: has children
@@ -490,7 +511,7 @@ def transpose(self, rows):
 
  def get_wellseries(self, matrix):
  """
- Returns the grid as a series of WellSeries
+ Returns the grid as a WellSeries of WellSeries
  """
  res = OrderedDict()
  for row, cells in matrix.items():
@@ -500,19 +521,19 @@ def get_wellseries(self, matrix):
  res[row][col] = self.children_by_name[
  ''.join(reversed(cell))
  ]
- res[row] = WellSeries(res[row])
+ res[row] = WellSeries(res[row], name=row)
  return WellSeries(res)
 
  @property
  def rows(self):
  """
  Rows can be accessed as:
- >>> plate.row[0]
- >>> plate.row['1']
+ >>> plate.rows[0]
+ >>> plate.rows['1']
 
  Wells can be accessed as:
- >>> plate.row[0][0]
- >>> plate.row['1']['A']
+ >>> plate.rows[0][0]
+ >>> plate.rows['1']['A']
  """
  self.calculate_grid()
  return self.grid
@@ -544,20 +565,27 @@ def cols(self):
  """
  return self.columns
 
- def well(self, name):
+ def well(self, name=None):
  """
  Returns well by :name:
  """
- return self.get_child_by_name(name)
+ return self.__getitem__(name)
 
- def wells(self):
+ def wells(self, *args):
  """
- Returns all wells
+ Returns child Well or list of child Wells
  """
- return self.get_children()
+ new_wells = None
+ if len(args) > 0:
+ new_wells = WellSeries([self.well(n) for n in args])
+ else:
+ new_wells = WellSeries(self.get_children_list())
+ if len(new_wells) is 1:
+ return new_wells[0]
+ return new_wells
 
 
-class WellSeries(Placeable):
+class WellSeries(Container):
  """
  :WellSeries: represents a series of wells to make
  accessing rows and columns easier. You can access
@@ -568,35 +596,46 @@ class WellSeries(Placeable):
 
  Default well index can be overriden using :set_offset:
  """
- def __init__(self, items):
- self.items = items
- if isinstance(items, dict):
- items = list(self.items.values())
- self.values = items
+ def __init__(self, wells, name=None):
+ if isinstance(wells, dict):
+ self.items = wells
+ self.values = list(wells.values())
+ else:
+ self.items = {w.get_name(): w for w in wells}
+ self.values = wells
  self.offset = 0
+ self.name = name
 
  def set_offset(self, offset):
  """
  Set index of a well that will be used to mimic :Placeable:
  """
  self.offset = offset
 
- def __iter__(self):
- return iter(self.values)
-
  def __str__(self):
  return '<Series: {}>'.format(
  ' '.join([str(well) for well in self.values]))
 
- def __getitem__(self, index):
- if isinstance(index, str):
- return self.items[index]
- else:
- return list(self.values)[index]
-
  def __getattr__(self, name):
  # getstate/setstate are used by pickle and are not implemented by
  # downstream objects (Wells) therefore raise attribute error
  if name in ('__getstate__', '__setstate__'):
  raise AttributeError()
  return getattr(self.values[self.offset], name)
+
+ def get_name(self):
+ if self.name is None:
+ return str(self)
+ return str(self.name)
+
+ def get_name_by_instance(self, well):
+ for name, value in self.items.items():
+ if value is well:
+ return name
+ return None
+
+ def get_children_list(self):
+ return list(self.values)
+
+ def get_child_by_name(self, name):
+ return self.items.get(name)

opentrons/helpers/helpers.py

@@ -1,6 +1,7 @@
 import json
 
 from opentrons.util.vector import Vector
+from opentrons.containers.placeable import Placeable
 
 
 def unpack_coordinates(coordinates):
@@ -98,3 +99,184 @@ def import_calibration_file(file_name, robot):
  with open(file_name) as f:
  json_string = '\n'.join(f)
  import_calibration_json(json_string, robot)
+
+
+def _get_list(n):
+ if not hasattr(n, '__len__') or len(n) == 0 or isinstance(n, tuple):
+ n = [n]
+ if isinstance(n, Placeable) and len(n) == 1:
+ n = [n[0]]
+ return n
+
+
+def _create_source_target_lists(s, t, **kwargs):
+ s = _get_list(s)
+ t = _get_list(t)
+ mode = kwargs.get('mode', 'transfer')
+ if mode == 'transfer':
+ if len(s) != len(t):
+ raise RuntimeError(
+ 'Transfer sources/targets must be same length')
+ elif mode == 'distribute':
+ if not (len(t) >= len(s) == 1):
+ raise RuntimeError(
+ 'Distribute must have 1 source and multiple targets')
+ s *= len(t)
+ elif mode == 'consolidate':
+ if not (len(s) >= len(t) == 1):
+ raise RuntimeError(
+ 'Consolidate must have multiple sources and 1 target')
+ t *= len(s)
+ return (s, t)
+
+
+def _create_volume_list(v, total, **kwargs):
+
+ gradient = kwargs.get('gradient', None)
+
+ if isinstance(v, tuple):
+ return _create_volume_gradient(
+ v[0], v[-1], total, gradient=gradient)
+
+ v = _get_list(v)
+ t_vol = len(v)
+ if (t_vol < total and t_vol != 1) or t_vol > total:
+ raise RuntimeError(
+ '{0} volumes do not match with {1} transfers'.format(
+ t_vol, total))
+ if t_vol < total:
+ v = [v[0]] * total
+ return v
+
+
+def _create_volume_gradient(min_v, max_v, total, gradient=None):
+
+ diff_vol = max_v - min_v
+
+ def _map_volume(i):
+ nonlocal diff_vol, total
+ rel_x = i / (total - 1)
+ rel_y = gradient(rel_x) if gradient else rel_x
+ return (rel_y * diff_vol) + min_v
+
+ return [_map_volume(i) for i in range(total)]
+
+
+def _compress_for_repeater(min_vol, max_vol, plan, **kwargs):
+ min_vol = float(min_vol)
+ max_vol = float(max_vol)
+ mode = kwargs.get('mode', 'transfer')
+ if mode == 'distribute': # combine target volumes into single aspirate
+ return _compress_for_distribute(min_vol, max_vol, plan, **kwargs)
+ if mode == 'consolidate': # combine target volumes into multiple aspirates
+ return _compress_for_consolidate(min_vol, max_vol, plan, **kwargs)
+ else:
+ return plan
+
+
+def _compress_for_distribute(min_vol, max_vol, plan, **kwargs):
+ source = plan[0]['aspirate']['location']
+ a_vol = 0
+ temp_dispenses = []
+ new_transfer_plan = []
+
+ def _add():
+ nonlocal a_vol, temp_dispenses, new_transfer_plan, source
+
+ if not temp_dispenses:
+ return
+
+ # distribute commands get an extra volume added
+ if len(temp_dispenses) > 1:
+ a_vol += min_vol
+
+ new_transfer_plan.append({
+ 'aspirate': {
+ 'location': source, 'volume': a_vol
+ }
+ })
+ for d in temp_dispenses:
+ new_transfer_plan.append({
+ 'dispense': {
+ 'location': d['location'], 'volume': d['volume']
+ }
+ })
+
+ if min_vol and len(temp_dispenses) > 1:
+ new_transfer_plan.append({
+ 'blow_out': {'blow_out': True}
+ })
+
+ for p in plan:
+ this_vol = p['aspirate']['volume']
+ if this_vol + a_vol > max_vol - min_vol:
+ _add()
+ a_vol = 0
+ temp_dispenses = []
+ a_vol += this_vol
+ temp_dispenses.append(p['dispense'])
+ _add()
+ return new_transfer_plan
+
+
+def _compress_for_consolidate(min_vol, max_vol, plan, **kwargs):
+ target = plan[0]['dispense']['location']
+ d_vol = 0
+ temp_aspirates = []
+ new_transfer_plan = []
+
+ def _add():
+ nonlocal d_vol, temp_aspirates, new_transfer_plan, target
+ for a in temp_aspirates:
+ new_transfer_plan.append({
+ 'aspirate': {
+ 'location': a['location'], 'volume': a['volume']
+ }
+ })
+ new_transfer_plan.append({
+ 'dispense': {
+ 'location': target, 'volume': d_vol
+ }
+ })
+ d_vol = 0
+ temp_aspirates = []
+
+ for i, p in enumerate(plan):
+ this_vol = p['aspirate']['volume']
+ if this_vol + d_vol > max_vol:
+ _add()
+ d_vol += this_vol
+ temp_aspirates.append(p['aspirate'])
+ _add()
+ return new_transfer_plan
+
+
+def _expand_for_carryover(min_vol, max_vol, plan, **kwargs):
+ max_vol = float(max_vol)
+ carryover = kwargs.get('carryover', True)
+ if not carryover:
+ return plan
+ new_transfer_plan = []
+ for p in plan:
+ source = p['aspirate']['location']
+ target = p['dispense']['location']
+ volume = float(p['aspirate']['volume'])
+ while volume > max_vol * 2:
+ new_transfer_plan.append({
+ 'aspirate': {'location': source, 'volume': max_vol},
+ 'dispense': {'location': target, 'volume': max_vol}
+ })
+ volume -= max_vol
+
+ # try and make sure no volumes are less than min_vol
+ if volume > max_vol:
+ volume /= 2
+ new_transfer_plan.append({
+ 'aspirate': {'location': source, 'volume': float(volume)},
+ 'dispense': {'location': target, 'volume': float(volume)}
+ })
+ new_transfer_plan.append({
+ 'aspirate': {'location': source, 'volume': float(volume)},
+ 'dispense': {'location': target, 'volume': float(volume)}
+ })
+ return new_transfer_plan