707 lab 680 update gateway to read bacalhau job state from appropriat…

…e state value (#708)

README.md

@@ -15,9 +15,6 @@
     <a href="https://img.shields.io/github/downloads/labdao/plex/total">
         <img src="https://img.shields.io/github/downloads/labdao/plex/total" alt="total download">
     </a>
-    <a href="https://github.com/labdao/plex/actions/workflows/go.yml" alt="Tests">
-        <img src="https://github.com/labdao/plex/actions/workflows/go.yml/badge.svg" />
-    </a>
     <a href="https://labdao.xyz/">
         <img alt="LabDAO website" src="https://img.shields.io/badge/website-labdao.xyz-red">
     </a>

gateway/handlers/flows.go

@@ -255,6 +255,8 @@ func UpdateFlowHandler(db *gorm.DB) http.HandlerFunc {
 				flow.Jobs[index].State = "processing"
 			} else if updatedJob.State.State == model.JobStateCompleted {
 				flow.Jobs[index].State = "completed"
+			} else if len(updatedJob.State.Executions) > 0 && updatedJob.State.Executions[0].State == model.ExecutionStateFailed {
+				flow.Jobs[index].State = "failed"
 			}
 
 			log.Println("Updated job")

gateway/handlers/jobs.go

@@ -88,6 +88,8 @@ func UpdateJobHandler(db *gorm.DB) http.HandlerFunc {
 			job.State = "processing"
 		} else if updatedJob.State.State == model.JobStateCompleted {
 			job.State = "completed"
+		} else if len(updatedJob.State.Executions) > 0 && updatedJob.State.Executions[0].State == model.ExecutionStateFailed {
+			job.State = "failed"
 		}
 		log.Println("Updated job")
 

internal/bacalhau/bacalhau.go

@@ -30,37 +30,27 @@ func GetBacalhauApiHost() string {
 	}
 }
 
-func CreateBacalhauJobV2(inputs map[string]string, container, cmd, selector string, maxTime, memory int, gpu, network bool, annotations []string) (job *model.Job, err error) {
+func CreateBacalhauJobV2(inputs map[string]string, container, selector string, cmd []string, maxTime, memory int, gpu, network bool, annotations []string) (job *model.Job, err error) {
 	log.Println("Creating job inside v2 function")
 	job, err = model.NewJobWithSaneProductionDefaults()
 	if err != nil {
 		return nil, err
 	}
-	cmdJoined := strings.Join([]string{"/bin/bash", "-c", cmd}, " ")
+	fmt.Println("container cmd", cmd)
 	job.Spec.EngineSpec = model.NewDockerEngineBuilder(container).
-		WithEntrypoint(cmdJoined).Build()
+		WithEntrypoint(cmd...).Build()
 	job.Spec.PublisherSpec = model.PublisherSpec{
 		Type: model.PublisherIpfs,
 	}
 	job.Spec.Annotations = annotations
 	job.Spec.Timeout = int64(maxTime * 60)
-	job.Spec.Engine = model.EngineDocker
-	job.Spec.Docker.Image = container
-	job.Spec.Docker.Entrypoint = []string{"/bin/bash", "-c", cmd}
 
-	// plexEnv, _ := os.LookupEnv("PLEX_ENV")
-	// if selector == "" && plexEnv == "stage" {
-	// 	selector = "owner=labdaostage"
-	// } else if selector == "" && plexEnv == "prod" {
-	// 	selector = "owner=labdao"
-	// }
-	// nodeSelectorRequirements, err := parse.NodeSelector(selector)
-	// if err != nil {
-	// 	return nil, err
-	// }
-	// job.Spec.NodeSelectors = nodeSelectorRequirements
+	nodeSelectorRequirements, err := parse.NodeSelector(selector)
+	if err != nil {
+		return nil, err
+	}
+	job.Spec.NodeSelectors = nodeSelectorRequirements
 
-	log.Println("Checking memory arg")
 	if memory > 0 {
 		job.Spec.Resources.Memory = fmt.Sprintf("%dgb", memory)
 	}
@@ -71,14 +61,13 @@ func CreateBacalhauJobV2(inputs map[string]string, container, cmd, selector stri
 		job.Spec.Network = model.NetworkConfig{Type: model.NetworkFull}
 	}
 	job.Spec.Inputs = []model.StorageSpec{}
-	log.Println("mapping inputs")
 	for key, cid := range inputs {
 		job.Spec.Inputs = append(job.Spec.Inputs,
 			// ToDo for arrays split by comma and put inside a dir that is key/index
 			model.StorageSpec{
 				StorageSource: model.StorageSourceIPFS,
 				CID:           cid,
-				Path:          "/" + key,
+				Path:          "/inputs/" + key,
 			})
 	}
 	job.Spec.Outputs = []model.StorageSpec{{Name: "outputs", StorageSource: model.StorageSourceIPFS, Path: "/outputs"}}

internal/ipwl/ipwl.go

@@ -111,7 +111,7 @@ func SubmitIoList(ioList []IO, selector string, maxTime int, annotations []strin
 			memory = *toolConfig.MemoryGB
 		}
 		log.Println("creating bacalhau job")
-		bacalhauJob, err := bacalhau.CreateBacalhauJobV2(bacalhauInputs, toolConfig.DockerPull, cmd, selector, maxTime, memory, toolConfig.GpuBool, toolConfig.NetworkBool, annotations)
+		bacalhauJob, err := bacalhau.CreateBacalhauJobV2(bacalhauInputs, toolConfig.DockerPull, selector, cmd, maxTime, memory, toolConfig.GpuBool, toolConfig.NetworkBool, annotations)
 		if err != nil {
 			submittedIOList[i].State = "failed"
 			submittedIOList[i].ErrMsg = fmt.Sprintf("error creating Bacalhau job: %v", err)

internal/ipwl/tool.go

@@ -112,7 +112,7 @@ func ReadToolConfig(toolPath string) (Tool, ToolInfo, error) {
 	return tool, toolInfo, nil
 }
 
-func toolToCmd2(toolConfig Tool, ioEntry IO, ioGraph []IO) (string, error) {
+func toolToCmd2(toolConfig Tool, ioEntry IO, ioGraph []IO) ([]string, error) {
 	arguments := strings.Join(toolConfig.Arguments, " ")
 
 	placeholderRegex := regexp.MustCompile(`\$\((inputs\..+?(\.filepath|\.basename|\.ext|\.default))\)`)
@@ -154,7 +154,8 @@ func toolToCmd2(toolConfig Tool, ioEntry IO, ioGraph []IO) (string, error) {
 		}
 	}
 
-	cmd := fmt.Sprintf("%s \"%s\"", strings.Join(toolConfig.BaseCommand, " "), arguments)
+	cmd := toolConfig.BaseCommand
+	cmd = append(cmd, arguments)
 
 	return cmd, nil
 }

tools/equibind/equibind_gateway.json

@@ -0,0 +1,45 @@
+{
+  "class": "CommandLineTool",
+  "name": "equibind gatway",
+  "description": "Docking of small molecules to a protein",
+  "author": "@misc{stärk2022equibind,\n      title={EquiBind: Geometric Deep Learning for Drug Binding Structure Prediction}, \n      author={Hannes Stärk and Octavian-Eugen Ganea and Lagnajit Pattanaik and Regina Barzilay and Tommi Jaakkola},\n      year={2022},\n      eprint={2202.05146},\n      archivePrefix={arXiv},\n      primaryClass={q-bio.BM}\n}",
+  "baseCommand": ["/bin/bash", "-c"],
+  "arguments": [
+    "mkdir -p /tmp-inputs/tmp;",
+    "mkdir -p /tmp-outputs/tmp;",
+    "cp /inputs/protein/* /tmp-inputs/tmp/;",
+    "cp /inputs/small_molecule/* /tmp-inputs/tmp/;",
+    "ls /tmp-inputs/tmp;",
+    "cd /src && python /src/inference.py --config=/src/configs_clean/bacalhau.yml;",
+    "mv /tmp-outputs/tmp/* /outputs/;",
+    "mv /outputs/lig_equibind_corrected.sdf /outputs/$(inputs.protein.basename)_$(inputs.small_molecule.basename)_docked.$(inputs.small_molecule.ext);",
+    "mv /tmp-inputs/tmp/*.pdb /outputs/;"
+  ],
+  "dockerPull": "ghcr.io/labdao/equibind:main@sha256:21a381d9ab1ff047565685044569c8536a55e489c9531326498b28d6b3cc244f",
+  "gpuBool": false,
+  "networkBool": false,
+  "inputs": {
+    "protein": {
+      "type": "File",
+      "item": "",
+      "glob": ["*.pdb"]
+    },
+    "small_molecule": {
+      "type": "File",
+      "item": "",
+      "glob": ["*.sdf", "*.mol2"]
+    }
+  },
+  "outputs": {
+    "best_docked_small_molecule": {
+      "type": "File",
+      "item": "",
+      "glob": ["*_docked.sdf", "*_docked.mol2"]
+    },
+    "protein": {
+      "type": "File", 
+      "item": "",
+      "glob": ["*.pdb"]
+    }
+  }
+}