feat: add async support for OSCAR (#72)

* Updated generic OSCAR script for merge * feat: revert to correct version * style * style * fix: fix indent * style * Deleted unused variables * Undo removal of python arg -u * feat: create buckets automatically * update oscar script * fix: fix typos * update script oscar --------- Co-authored-by: catttam <[email protected]> Co-authored-by: SergioLangaritaBenitez <[email protected]>
ai4os · Dec 10, 2024 · 387a21f · 387a21f
1 parent f4cdd49
commit 387a21f
Showing 1 changed file with 52 additions and 10 deletions.
diff --git a/etc/oscar.yaml b/etc/oscar.yaml
@@ -11,11 +11,18 @@ memory: ${MEMORY}Mi
 vo: ${VO}
 allowed_users: ${ALLOWED_USERS}
 environment: ${ENV_VARS}
+input:
+  - storage_provider: minio.default
+    path: ${NAME}/inputs
+output:
+  - storage_provider: minio.default
+    path: ${NAME}/outputs
 script: |-
   echo "[*] Using DEEPaaS version $(deepaas-run --version)"
   echo "[*] Using Python version $(python3 --version)"
-
-  python -u - << EOF
+  FILE_NAME=`basename "$INPUT_FILE_PATH" | cut -f 1 -d '.' `
+  OUTPUT_FILE="$TMP_OUTPUT_DIR/$FILE_NAME"
+  python >service.log 2>&1 << EOF
 
   import base64
   import json
@@ -25,6 +32,18 @@ script: |-
   import subprocess
 
 
+  # Aux function to get a random filename
+  def get_rnd_filename(ex):
+      rnd_str = ''.join(random.choice(string.ascii_lowercase) for i in range(5))
+      if ex == "":
+          TMP_OUTPUT_DIR = os.getenv("TMP_OUTPUT_DIR")
+          filename = ''.join([TMP_OUTPUT_DIR, "/" ,"tmp-file-", rnd_str])
+          print("Output will be saved on:", filename)
+          return filename
+      else:
+          return ''.join(["tmp-file-", rnd_str, ".", ex])
+
+
   # Check the DEEPaaS version
   def compare_versions(version1, version2):
       """
@@ -61,22 +80,30 @@ script: |-
       print(f"Error: DEEPaaS version must be >={required}. Current version is: {current}")
       exit(1)
 
-  # Read input file with params and create the command
-  subprocess.run(["mv", "$INPUT_FILE_PATH", "$INPUT_FILE_PATH.json"])
-  FILE_PATH = os.getenv("INPUT_FILE_PATH") + ".json"
+  # Read input file
+  FILE_PATH = os.getenv("INPUT_FILE_PATH")
+  SAVE_OUT = True
+  if not FILE_PATH.endswith(".json"):
+      SAVE_OUT = False
+      subprocess.run(["mv", "$INPUT_FILE_PATH", "$INPUT_FILE_PATH.json"])
+      FILE_PATH = os.getenv("INPUT_FILE_PATH") + ".json"
+
+  # Process input
   with open(FILE_PATH, "r") as f:
       params = json.loads(f.read())
 
   # Create the DEEPaaS predict command
-  DEEPAAS_CLI_COMMAND = ["deepaas-cli", "predict"]
+  if SAVE_OUT:
+      DEEPAAS_CLI_COMMAND = ["deepaas-cli", "--deepaas_method_output", get_rnd_filename(""), "predict"]
+  else:
+      DEEPAAS_CLI_COMMAND = ["deepaas-cli", "predict"]
 
   for k, v in params.items():
 
       # If param is 'oscar-files' decode the array of files
       if k == "oscar-files":
           for file in v:
-              rnd_str = "".join(random.choice(string.ascii_lowercase) for i in range(5))
-              filename = "".join(["tmp-file-", rnd_str, ".", file["file_format"]])
+              filename = get_rnd_filename(file["file_format"])
               k, v = file["key"], filename
 
               print("[*] Processing file: ", filename)
@@ -94,6 +121,21 @@ script: |-
           DEEPAAS_CLI_COMMAND += [f"--{k}", v]
 
   print(f"[*] Final command: {' '.join(DEEPAAS_CLI_COMMAND)}")
-  subprocess.run(DEEPAAS_CLI_COMMAND)
-
+  try:
+    subprocess.run(DEEPAAS_CLI_COMMAND)
+  except:
+    print("Something went wrong during the execution.")
+    exit(1)
   EOF
+  
+  if grep -q 'New output is' "service.log"
+  then
+    FILE_NAME_2=`cat service.log | grep 'New output is' | cut -f 5 -d ' '`
+    mv $FILE_NAME_2 $OUTPUT_FILE.json
+    mv service.log  $OUTPUT_FILE.log
+  elif echo "$FILE_NAME" | grep event-file
+  then
+    cat service.log
+  else
+    mv service.log  $OUTPUT_FILE.log
+  fi