Update to 0.1.0 spec (WIP)

Author: Peter Amstutz

cwl_runner_wes.py

@@ -8,26 +8,36 @@
 import uuid
 import os
 import json
+import urllib
 
 class Workflow(object):
-    def __init__(self, workflow_ID):
+    def __init__(self, workflow_id):
         super(Workflow, self).__init__()
-        self.workflow_ID = workflow_ID
-        self.workdir = os.path.abspath(self.workflow_ID)
+        self.workflow_id = workflow_id
+        self.workdir = os.path.join(os.getcwd(), "workflows", self.workflow_id)
 
-    def run(self, path, inputobj):
-        path = os.path.abspath(path)
-        os.mkdir(self.workdir)
+    def run(self, request):
+        os.makedirs(self.workdir)
         outdir = os.path.join(self.workdir, "outdir")
         os.mkdir(outdir)
+
+        with open(os.path.join(self.workdir, "request.json"), "w") as f:
+            json.dump(request, f)
+
         with open(os.path.join(self.workdir, "cwl.input.json"), "w") as inputtemp:
-            json.dump(inputobj, inputtemp)
-        with open(os.path.join(self.workdir, "workflow_url"), "w") as f:
-            f.write(path)
+            inputtemp.write(request["workflow_params"])
+
+        if request.get("workflow_descriptor"):
+            with open(os.path.join(self.workdir, "workflow.cwl"), "w") as f:
+                f.write(workflow_descriptor)
+                workflow_url = urllib.pathname2url(os.path.join(self.workdir, "workflow.cwl"))
+        else:
+            workflow_url = request.get("workflow_url")
+
         output = open(os.path.join(self.workdir, "cwl.output.json"), "w")
         stderr = open(os.path.join(self.workdir, "stderr"), "w")
 
-        proc = subprocess.Popen(["cwl-runner", path, inputtemp.name],
+        proc = subprocess.Popen(["cwl-runner", workflow_url, inputtemp.name],
                                 stdout=output,
                                 stderr=stderr,
                                 close_fds=True,
@@ -60,70 +70,96 @@ def getstate(self):
         if exit_code == 0:
             state = "Complete"
         elif exit_code != -1:
-            state = "Failed"
+            state = "Error"
 
         return (state, exit_code)
 
     def getstatus(self):
         state, exit_code = self.getstate()
 
-        with open(os.path.join(self.workdir, "cwl.input.json"), "r") as inputtemp:
-            inputobj = json.load(inputtemp)
-        with open(os.path.join(self.workdir, "workflow_url"), "r") as f:
-            workflow_url = f.read()
-
-        outputobj = {}
-        if state == "Complete":
-            with open(os.path.join(self.workdir, "cwl.output.json"), "r") as outputtemp:
-                outputobj = json.load(outputtemp)
-
         return {
-            "workflow_ID": self.workflow_ID,
-            "workflow_url": workflow_url,
-            "input": inputobj,
-            "output": outputobj,
+            "workflow_id": self.workflow_id,
             "state": state
         }
 
 
     def getlog(self):
         state, exit_code = self.getstate()
 
+        with open(os.path.join(self.workdir, "request.json"), "r") as f:
+            request = json.load(f)
+
         with open(os.path.join(self.workdir, "stderr"), "r") as f:
             stderr = f.read()
 
+        if state == "Complete":
+            with open(os.path.join(self.workdir, "cwl.output.json"), "r") as outputtemp:
+                outputobj = json.load(outputtemp)
+
         return {
-            "workflow_ID": self.workflow_ID,
-            "log": {
+            "workflow_id": self.workflow_id,
+            "request": request,
+            "state": state,
+            "workflow_log": {
                 "cmd": [""],
                 "startTime": "",
                 "endTime": "",
                 "stdout": "",
                 "stderr": stderr,
                 "exitCode": exit_code
-            }
+            },
+            "task_logs": [],
+            "outputs": []
         }
 
     def cancel(self):
         pass
 
-def GetWorkflowStatus(workflow_ID):
-    job = Workflow(workflow_ID)
-    return job.getstatus()
+def GetServiceInfo():
+    return {
+        "workflow_type_versions": {
+            "CWL": ["v1.0"]
+        },
+        "supported_wes_versions": "0.1.0",
+        "supported_filesystem_protocols": ["file"],
+        "engine_versions": "cwl-runner",
+        "system_state_counts": {},
+        "key_values": {}
+    }
+
+def ListWorkflows(body):
+    # body["page_size"]
+    # body["page_token"]
+    # body["key_value_search"]
+
+    wf = []
+    for l in os.listdir(os.path.join(os.getcwd(), "workflows")):
+        if os.path.isdir(os.path.join(os.getcwd(), "workflows", l)):
+            wf.append(Workflow(l))
+    return {
+        "workflows": [{"workflow_id": w.workflow_id, "state": w.getstate()} for w in wf],
+        "next_page_token": ""
+    }
 
-def GetWorkflowLog(workflow_ID):
-    job = Workflow(workflow_ID)
+def RunWorkflow(body):
+    if body["workflow_type"] != "CWL" or body["workflow_type_version"] != "v1.0":
+        return
+    workflow_id = uuid.uuid4().hex
+    job = Workflow(workflow_id)
+    job.run(body)
+    return {"workflow_id": workflow_id}
+
+def GetWorkflowLog(workflow_id):
+    job = Workflow(workflow_id)
     return job.getlog()
 
-def CancelWorkflow(workflow_ID):
-    job = Workflow(workflow_ID)
+def CancelJob(workflow_id):
+    job = Workflow(workflow_id)
     job.cancel()
-    return job.getstatus()
+    return {"workflow_id": workflow_id}
 
-def RunWorkflow(body):
-    workflow_ID = uuid.uuid4().hex
-    job = Workflow(workflow_ID)
-    job.run(body["workflow_url"], body["input"])
+def GetWorkflowStatus(workflow_id):
+    job = Workflow(workflow_id)
     return job.getstatus()
 
 def main():