11 #import arvados.command.ws as ws
15 logger = logging.getLogger('arvados.arv-run')
17 arvrun_parser = argparse.ArgumentParser()
18 arvrun_parser.add_argument('--dry-run', action="store_true")
19 arvrun_parser.add_argument('--local', action="store_true")
20 arvrun_parser.add_argument('--docker-image', type=str, default="arvados/jobs")
21 arvrun_parser.add_argument('--git-dir', type=str, default="")
22 arvrun_parser.add_argument('args', nargs=argparse.REMAINDER)
24 class UploadFile(object):
25 def __init__(self, prefix, fn):
29 def is_in_collection(root, branch):
32 fn = os.path.join(root, ".arvados#collection")
33 if os.path.exists(fn):
34 with file(fn, 'r') as f:
36 return (c["portable_data_hash"], branch)
38 sp = os.path.split(root)
39 return is_in_collection(sp[0], os.path.join(sp[1], branch))
41 def statfile(prefix, fn):
42 absfn = os.path.abspath(fn)
43 if os.path.exists(absfn):
45 if stat.S_ISREG(st.st_mode):
46 sp = os.path.split(absfn)
47 (pdh, branch) = is_in_collection(sp[0], sp[1])
49 return "%s$(file %s/%s)" % (prefix, pdh, branch)
51 # trim leading '/' for path prefix test later
52 return UploadFile(prefix, absfn[1:])
55 def main(arguments=None):
56 args = arvrun_parser.parse_args(arguments)
67 reading_into = len(slots)
70 slots[reading_into].append(c)
72 if slots[0] and len(slots[0]) > 1:
73 logger.error("Can only specify a single stdout file (run-command substitutions are permitted)")
76 patterns = [re.compile("(--[^=]+=)(.*)"),
77 re.compile("(-[^=]+=)(.*)"),
78 re.compile("(-.)(.+)")]
80 for command in slots[1:]:
81 for i in xrange(0, len(command)):
84 # parameter starts with '-' so it might be a command line
85 # parameter with a file name, do some pattern matching
90 command[i] = statfile(m.group(1), m.group(2))
93 # parameter might be a file, so test it
94 command[i] = statfile('', a)
98 files = [c for command in slots[1:] for c in command if isinstance(c, UploadFile)]
100 # Find the smallest path prefix that includes all the files that need to be uploaded.
101 # This starts at the root and iteratively removes common parent directory prefixes
102 # until all file pathes no longer have a common parent.
109 # no parent directories left
112 # path step takes next directory
113 pathstep = sp[0] + "/"
115 # check if pathstep is common prefix for all files
116 if not c.fn.startswith(pathstep):
120 # pathstep is common parent directory for all files, so remove the prefix
122 pathprefix += pathstep
124 c.fn = c.fn[len(pathstep):]
129 print("cd %s" % pathprefix)
130 print("arv-put \"%s\"" % '" "'.join([c.fn for c in files]))
133 pdh = put.main(["--portable-data-hash"]+[c.fn for c in files])
135 for i in xrange(1, len(slots)):
136 slots[i] = [("%s$(file %s/%s)" % (c.prefix, pdh, c.fn)) if isinstance(c, UploadFile) else c for c in slots[i]]
139 "script": "run-command",
140 "script_version": "3609-arv-run",
141 "repository": "arvados",
142 "script_parameters": {
145 "runtime_constraints": {
146 "docker_image": args.docker_image
153 component["script_parameters"]["task.stdout"] = slots[0][0]
155 task_foreach.append("stdin")
156 component["script_parameters"]["stdin"] = slots[1]
157 component["script_parameters"]["task.stdin"] = "$(stdin)"\
160 component["script_parameters"]["task.foreach"] = task_foreach
167 "state":"RunningOnServer"
171 print(json.dumps(pipeline, indent=4))
173 subprocess.call(["arv-crunch-job", "--job", json.dumps(component), "--git-dir", args.git_dir])
175 api = arvados.api('v1')
176 pi = api.pipeline_instances().create(body=pipeline).execute()
177 #ws.main(["--pipeline", pi["uuid"]])
179 if __name__ == '__main__':