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:
- # Var specifies
+ 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]))
- m = re.match("^\$\((.*)\)$", c[key])
- if m and m.group(1) in p:
- return (m.group(1), 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:
- # backwards compatible, foreach specifies bare parameter name to use
- return (c[key], get_items(p, p[c[key]]))
+ 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
- if "list" in c and "index" in c and "command" in c:
+ 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"])
- if "regex" in c:
+ elif "regex" in c:
pattern = re.compile(c["regex"])
if "filter" in c:
- var, items = var_items(p, c, "filter")
+ # 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:
- var, items = var_items(p, c, "group")
+ # 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)
add_to_group(groups, match)
return [groups[k] for k in groups]
elif "extract" in c:
- var, items = var_items(p, c, "extract")
+ # 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
- if "batch" in c and "size" in c:
- var, items = var_items(p, c, "batch")
+ 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):
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))
- return []
-
+# 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):
with open(value) as f:
items = [line.rstrip("\r\n") for line in f]
return items
- raise Exception("get_items did not yield a list")
+ 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:]
logger.error(pprint.pformat(taskp))
sys.exit(1)
+# rcode holds the return codes produced by each subprocess
+rcode = {}
try:
subprocesses = []
close_streams = []
active = 1
pids = set([s.pid for s in subprocesses])
- rcode = {}
while len(pids) > 0:
(pid, status) = os.wait()
pids.discard(pid)
- rcode[pid] = (status >> 8)
+ if not taskp.get("task.ignore_rcode"):
+ rcode[pid] = (status >> 8)
+ else:
+ rcode[pid] = 0
if sig.sig is not None:
logger.critical("terminating on signal %s" % sig.sig)
else:
outcollection = robust_put.upload(outdir, logger)
-success = reduce(lambda x, y: x & (y == 0), [True]+rcode.values())
+# Success if no non-zero return codes
+success = any(rcode) and not any([status != 0 for status in rcode.values()])
api.job_tasks().update(uuid=arvados.current_task()['uuid'],
body={
projects / arvados.git / blobdiff