+
+# http://rightfootin.blogspot.com/2006/09/more-on-python-flatten.html
+def flatten(l, ltypes=(list, tuple)):
+ ltype = type(l)
+ l = list(l)
+ i = 0
+ while i < len(l):
+ while isinstance(l[i], ltypes):
+ if not l[i]:
+ l.pop(i)
+ i -= 1
+ break
+ else:
+ l[i:i + 1] = l[i]
+ i += 1
+ return ltype(l)
+
+def add_to_group(gr, match):
+ m = match.groups()
+ if m not in gr:
+ gr[m] = []
+ gr[m].append(match.group(0))
+
+class EvaluationError(Exception):
+ pass
+
+# Return the name of variable ('var') that will take on each value in 'items'
+# when performing an inner substitution
+def var_items(p, c, key):
+ if key not in c:
+ raise EvaluationError("'%s' was expected in 'p' but is missing" % key)
+
+ if "var" in c:
+ if not isinstance(c["var"], basestring):
+ raise EvaluationError("Value of 'var' must be a string")
+ # Var specifies the variable name for inner parameter substitution
+ return (c["var"], get_items(p, c[key]))
+ else:
+ # The component function ('key') value is a list, so return the list
+ # directly with no parameter selected.
+ if isinstance(c[key], list):
+ return (None, get_items(p, c[key]))
+ elif isinstance(c[key], basestring):
+ # check if c[key] is a string that looks like a parameter
+ m = re.match("^\$\((.*)\)$", c[key])
+ if m and m.group(1) in p:
+ return (m.group(1), get_items(p, c[key]))
+ else:
+ # backwards compatible, foreach specifies bare parameter name to use
+ return (c[key], get_items(p, p[c[key]]))
+ else:
+ raise EvaluationError("Value of '%s' must be a string or list" % key)
+
+# "p" is the parameter scope, "c" is the item to be expanded.
+# If "c" is a dict, apply function expansion.
+# If "c" is a list, recursively expand each item and return a new list.
+# If "c" is a string, apply parameter substitution
+def expand_item(p, c):
+ if isinstance(c, dict):
+ if "foreach" in c and "command" in c:
+ # Expand a command template for each item in the specified user
+ # parameter
+ var, items = var_items(p, c, "foreach")
+ if var is None:
+ raise EvaluationError("Must specify 'var' in foreach")
+ r = []
+ for i in items:
+ params = copy.copy(p)
+ params[var] = i
+ r.append(expand_item(params, c["command"]))
+ return r
+ elif "list" in c and "index" in c and "command" in c:
+ # extract a single item from a list
+ var, items = var_items(p, c, "list")
+ if var is None:
+ raise EvaluationError("Must specify 'var' in list")
+ params = copy.copy(p)
+ params[var] = items[int(c["index"])]
+ return expand_item(params, c["command"])
+ elif "regex" in c:
+ pattern = re.compile(c["regex"])
+ if "filter" in c:
+ # filter list so that it only includes items that match a
+ # regular expression
+ _, items = var_items(p, c, "filter")
+ return [i for i in items if pattern.match(i)]
+ elif "group" in c:
+ # generate a list of lists, where items are grouped on common
+ # subexpression match
+ _, items = var_items(p, c, "group")
+ groups = {}
+ for i in items:
+ match = pattern.match(i)
+ if match:
+ add_to_group(groups, match)
+ return [groups[k] for k in groups]
+ elif "extract" in c:
+ # generate a list of lists, where items are split by
+ # subexpression match
+ _, items = var_items(p, c, "extract")
+ r = []
+ for i in items:
+ match = pattern.match(i)
+ if match:
+ r.append(list(match.groups()))
+ return r
+ elif "batch" in c and "size" in c:
+ # generate a list of lists, where items are split into a batch size
+ _, items = var_items(p, c, "batch")
+ sz = int(c["size"])
+ r = []
+ for j in xrange(0, len(items), sz):
+ r.append(items[j:j+sz])
+ return r
+ raise EvaluationError("Missing valid list context function")
+ elif isinstance(c, list):
+ return [expand_item(p, arg) for arg in c]
+ elif isinstance(c, basestring):
+ m = re.match("^\$\((.*)\)$", c)
+ if m and m.group(1) in p:
+ return expand_item(p, p[m.group(1)])
+ else:
+ return subst.do_substitution(p, c)
+ else:
+ raise EvaluationError("expand_item() unexpected parameter type %s" % type(c))
+
+# Evaluate in a list context
+# "p" is the parameter scope, "value" will be evaluated
+# if "value" is a list after expansion, return that
+# if "value" is a path to a directory, return a list consisting of each entry in the directory
+# if "value" is a path to a file, return a list consisting of each line of the file
+def get_items(p, value):
+ value = expand_item(p, value)
+ if isinstance(value, list):
+ return value
+ elif isinstance(value, basestring):
+ mode = os.stat(value).st_mode
+ prefix = value[len(os.environ['TASK_KEEPMOUNT'])+1:]
+ if mode is not None:
+ if stat.S_ISDIR(mode):
+ items = [os.path.join(value, l) for l in os.listdir(value)]
+ elif stat.S_ISREG(mode):
+ with open(value) as f:
+ items = [line.rstrip("\r\n") for line in f]
+ return items
+ raise EvaluationError("get_items did not yield a list")
+
+stdoutname = None
+stdoutfile = None
+stdinname = None
+stdinfile = None
+
+# Construct the cross product of all values of each variable listed in fvars
+def recursive_foreach(params, fvars):
+ var = fvars[0]
+ fvars = fvars[1:]
+ items = get_items(params, params[var])
+ logger.info("parallelizing on %s with items %s" % (var, items))
+ if items is not None:
+ for i in items:
+ params = copy.copy(params)
+ params[var] = i
+ if len(fvars) > 0:
+ recursive_foreach(params, fvars)
+ else:
+ if not args.dry_run:
+ arvados.api().job_tasks().create(body={
+ 'job_uuid': arvados.current_job()['uuid'],
+ 'created_by_job_task_uuid': arvados.current_task()['uuid'],
+ 'sequence': 1,
+ 'parameters': params
+ }).execute()
+ else:
+ if isinstance(params["command"][0], list):
+ for c in params["command"]:
+ logger.info(flatten(expand_item(params, c)))
+ else:
+ logger.info(flatten(expand_item(params, params["command"])))
+ else:
+ logger.error("parameter %s with value %s in task.foreach yielded no items" % (var, params[var]))
+ sys.exit(1)