4 import arvados.commands.ws as ws
14 import arvados.commands._util as arv_cmd
16 logger = logging.getLogger('arvados.arv-run')
17 logger.setLevel(logging.INFO)
19 arvrun_parser = argparse.ArgumentParser(parents=[arv_cmd.retry_opt])
20 arvrun_parser.add_argument('--dry-run', action="store_true", help="Print out the pipeline that would be submitted and exit")
21 arvrun_parser.add_argument('--local', action="store_true", help="Run locally using arv-run-pipeline-instance")
22 arvrun_parser.add_argument('--docker-image', type=str, help="Docker image to use, otherwise use instance default.")
23 arvrun_parser.add_argument('--ignore-rcode', action="store_true", help="Commands that return non-zero return codes should not be considered failed.")
24 arvrun_parser.add_argument('--no-reuse', action="store_true", help="Do not reuse past jobs.")
25 arvrun_parser.add_argument('--no-wait', action="store_true", help="Do not wait and display logs after submitting command, just exit.")
26 arvrun_parser.add_argument('--project-uuid', type=str, help="Parent project of the pipeline")
27 arvrun_parser.add_argument('--git-dir', type=str, default="", help="Git repository passed to arv-crunch-job when using --local")
28 arvrun_parser.add_argument('--repository', type=str, default="arvados", help="repository field of component, default 'arvados'")
29 arvrun_parser.add_argument('--script-version', type=str, default="master", help="script_version field of component, default 'master'")
30 arvrun_parser.add_argument('args', nargs=argparse.REMAINDER)
32 class ArvFile(object):
33 def __init__(self, prefix, fn):
37 class UploadFile(ArvFile):
40 # Determine if a file is in a collection, and return a tuple consisting of the
41 # portable data hash and the path relative to the root of the collection.
42 # Return None if the path isn't with an arv-mount collection or there was is error.
43 def is_in_collection(root, branch):
47 fn = os.path.join(root, ".arvados#collection")
48 if os.path.exists(fn):
49 with file(fn, 'r') as f:
51 return (c["portable_data_hash"], branch)
53 sp = os.path.split(root)
54 return is_in_collection(sp[0], os.path.join(sp[1], branch))
55 except (IOError, OSError):
58 # Determine the project to place the output of this command by searching upward
59 # for arv-mount psuedofile indicating the project. If the cwd isn't within
60 # an arv-mount project or there is an error, return current_user.
61 def determine_project(root, current_user):
65 fn = os.path.join(root, ".arvados#project")
66 if os.path.exists(fn):
67 with file(fn, 'r') as f:
69 if 'writable_by' in c and current_user in c['writable_by']:
74 sp = os.path.split(root)
75 return determine_project(sp[0], current_user)
76 except (IOError, OSError):
79 # Determine if string corresponds to a file, and if that file is part of a
80 # arv-mounted collection or only local to the machine. Returns one of
81 # ArvFile() (file already exists in a collection), UploadFile() (file needs to
82 # be uploaded to a collection), or simply returns prefix+fn (which yields the
83 # original parameter string).
84 def statfile(prefix, fn, fnPattern="$(file %s/%s)", dirPattern="$(dir %s/%s/)"):
85 absfn = os.path.abspath(fn)
86 if os.path.exists(absfn):
88 if stat.S_ISREG(st.st_mode):
89 sp = os.path.split(absfn)
90 (pdh, branch) = is_in_collection(sp[0], sp[1])
92 return ArvFile(prefix, fnPattern % (pdh, branch))
94 # trim leading '/' for path prefix test later
95 return UploadFile(prefix, absfn[1:])
96 if stat.S_ISDIR(st.st_mode):
97 sp = os.path.split(absfn)
98 (pdh, branch) = is_in_collection(sp[0], sp[1])
100 return ArvFile(prefix, dirPattern % (pdh, branch))
104 def uploadfiles(files, api, dry_run=False, num_retries=0, project=None, fnPattern="$(file %s/%s)", name=None):
105 # Find the smallest path prefix that includes all the files that need to be uploaded.
106 # This starts at the root and iteratively removes common parent directory prefixes
107 # until all file pathes no longer have a common parent.
116 # no parent directories left
119 # path step takes next directory
120 pathstep = sp[0] + "/"
122 # check if pathstep is common prefix for all files
123 if not c.fn.startswith(pathstep):
127 # pathstep is common parent directory for all files, so remove the prefix
129 pathprefix += pathstep
131 c.fn = c.fn[len(pathstep):]
136 logger.info("Upload local files: \"%s\"", '" "'.join([c.fn for c in files]))
139 logger.info("$(input) is %s", pathprefix.rstrip('/'))
142 files = sorted(files, key=lambda x: x.fn)
143 collection = arvados.CollectionWriter(api, num_retries=num_retries)
146 sp = os.path.split(f.fn)
149 collection.start_new_stream(stream)
150 collection.write_file(f.fn, sp[1])
152 exists = api.collections().list(filters=[["owner_uuid", "=", project],
153 ["portable_data_hash", "=", collection.portable_data_hash()],
154 ["name", "=", name]]).execute(num_retries=num_retries)
156 item = exists["items"][0]
157 logger.info("Using collection %s", item["uuid"])
159 body = {"owner_uuid": project, "manifest_text": collection.manifest_text()}
162 item = api.collections().create(body=body, ensure_unique_name=True).execute()
163 logger.info("Uploaded to %s", item["uuid"])
165 pdh = item["portable_data_hash"]
168 c.fn = fnPattern % (pdh, c.fn)
173 def main(arguments=None):
174 args = arvrun_parser.parse_args(arguments)
176 if len(args.args) == 0:
177 arvrun_parser.print_help()
180 starting_args = args.args
184 # Parse the command arguments into 'slots'.
185 # All words following '>' are output arguments and are collected into slots[0].
186 # All words following '<' are input arguments and are collected into slots[1].
187 # slots[2..] store the parameters of each command in the pipeline.
189 # e.g. arv-run foo arg1 arg2 '|' bar arg3 arg4 '<' input1 input2 input3 '>' output.txt
190 # will be parsed into:
192 # ['input1', 'input2', 'input3'],
193 # ['foo', 'arg1', 'arg2'],
194 # ['bar', 'arg3', 'arg4']]
197 if c.startswith('>'):
200 slots[reading_into].append(c[1:])
201 elif c.startswith('<'):
204 slots[reading_into].append(c[1:])
206 reading_into = len(slots)
209 slots[reading_into].append(c)
211 if slots[0] and len(slots[0]) > 1:
212 logger.error("Can only specify a single stdout file (run-command substitutions are permitted)")
216 api = arvados.api('v1')
217 if args.project_uuid:
218 project = args.project_uuid
220 project = determine_project(os.getcwd(), api.users().current().execute()["uuid"])
222 # Identify input files. Look at each parameter and test to see if there is
223 # a file by that name. This uses 'patterns' to look for within
224 # command line arguments, such as --foo=file.txt or -lfile.txt
225 patterns = [re.compile("([^=]+=)(.*)"),
226 re.compile("(-[A-Za-z])(.+)")]
227 for j, command in enumerate(slots[1:]):
228 for i, a in enumerate(command):
230 # j == 0 is stdin, j > 0 is commands
231 # always skip program executable (i == 0) in commands
233 elif a.startswith('\\'):
234 # if it starts with a \ then don't do any interpretation
237 # See if it looks like a file
238 command[i] = statfile('', a)
240 # If a file named command[i] was found, it would now be an
241 # ArvFile or UploadFile. If command[i] is a basestring, that
242 # means it doesn't correspond exactly to a file, so do some
244 if isinstance(command[i], basestring):
248 command[i] = statfile(m.group(1), m.group(2))
251 files = [c for command in slots[1:] for c in command if isinstance(c, UploadFile)]
253 uploadfiles(files, api, dry_run=args.dry_run, num_retries=args.retries, project=project)
255 for i in xrange(1, len(slots)):
256 slots[i] = [("%s%s" % (c.prefix, c.fn)) if isinstance(c, ArvFile) else c for c in slots[i]]
259 "script": "run-command",
260 "script_version": args.script_version,
261 "repository": args.repository,
262 "script_parameters": {
264 "runtime_constraints": {}
267 if args.docker_image:
268 component["runtime_constraints"]["docker_image"] = args.docker_image
271 group_parser = argparse.ArgumentParser()
272 group_parser.add_argument('-b', '--batch-size', type=int)
273 group_parser.add_argument('args', nargs=argparse.REMAINDER)
275 for s in xrange(2, len(slots)):
276 for i in xrange(0, len(slots[s])):
277 if slots[s][i] == '--':
278 inp = "input%i" % (s-2)
279 groupargs = group_parser.parse_args(slots[2][i+1:])
280 if groupargs.batch_size:
281 component["script_parameters"][inp] = {"value": {"batch":groupargs.args, "size":groupargs.batch_size}}
282 slots[s] = slots[s][0:i] + [{"foreach": inp, "command": "$(%s)" % inp}]
284 component["script_parameters"][inp] = groupargs.args
285 slots[s] = slots[s][0:i] + ["$(%s)" % inp]
286 task_foreach.append(inp)
288 if slots[s][i] == '\--':
292 component["script_parameters"]["task.stdout"] = slots[0][0]
294 task_foreach.append("stdin")
295 component["script_parameters"]["stdin"] = slots[1]
296 component["script_parameters"]["task.stdin"] = "$(stdin)"
299 component["script_parameters"]["task.foreach"] = task_foreach
301 component["script_parameters"]["command"] = slots[2:]
302 if args.ignore_rcode:
303 component["script_parameters"]["task.ignore_rcode"] = args.ignore_rcode
306 "name": "arv-run " + " | ".join([s[0] for s in slots[2:]]),
307 "description": "@" + " ".join(starting_args) + "@",
311 "state": "RunningOnClient" if args.local else "RunningOnServer"
315 print(json.dumps(pipeline, indent=4))
317 pipeline["owner_uuid"] = project
318 pi = api.pipeline_instances().create(body=pipeline, ensure_unique_name=True).execute()
319 logger.info("Running pipeline %s", pi["uuid"])
322 subprocess.call(["arv-run-pipeline-instance", "--instance", pi["uuid"], "--run-jobs-here"] + (["--no-reuse"] if args.no_reuse else []))
323 elif not args.no_wait:
324 ws.main(["--pipeline", pi["uuid"]])
326 pi = api.pipeline_instances().get(uuid=pi["uuid"]).execute()
327 logger.info("Pipeline is %s", pi["state"])
328 if "output_uuid" in pi["components"]["command"]:
329 logger.info("Output is %s", pi["components"]["command"]["output_uuid"])
331 logger.info("No output")
333 if __name__ == '__main__':