Merge pull request #7 from Sage-Bionetworks-Workflows/bwmac/ORCA-222/…

…scoring_step

[ORCA-222] Adds Rudimentary Scoring Step

bin/score.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import sys
+
+
+def score_submission(predictions_path: str, status: str) -> dict:
+    """Determine the score of a submission. This is a placeholder function.
+
+    Args:
+        predictions_path (str): path to the predictions file
+        status (str): current submission status
+
+    Returns:
+        result (dict): dictionary containing score, status and errors
+    """
+    if status == "INVALID":
+        score_status = "INVALID"
+        score = None
+    else:
+        # placeholder file reading
+        with open(predictions_path, "r") as sub_file:
+            predictions_contents = sub_file.read()
+        try:
+            # placeholder scoring
+            score = 1 + 1
+            score_status = "SCORED"
+            message = ""
+        except Exception as e:
+            message = f"Error {e} occurred while scoring"
+            score = None
+            score_status = "INVALID"
+    result = {
+        "score": {
+            "score_category": "auc",
+            "score": score,
+            "status": score_status,
+            "errors": message,
+        }
+    }
+    return score_status, result
+
+
+def update_json(results_path: str, result: dict) -> None:
+    """Update the results.json file with the current score and status
+
+    Args:
+        results_path (str): path to the results.json file
+        result (dict): dictionary containing score, status and errors
+    """
+    file_size = os.path.getsize(results_path)
+    with open(results_path, "r") as o:
+        data = json.load(o) if file_size else {}
+    data.update(result)
+    with open(results_path, "w") as o:
+        o.write(json.dumps(data))
+
+
+if __name__ == "__main__":
+    predictions_path = sys.argv[1]
+    results_path = sys.argv[2]
+    status = sys.argv[3]
+    score_status, result = score_submission(predictions_path, status)
+    update_json(results_path, result)
+    print(score_status)

bin/validate.py

@@ -17,8 +17,7 @@
             prediction_status = "INVALID"
             invalid_reasons.append("Predicitons file is empty")
     result = {
-        "submission_errors": ";".join(invalid_reasons),
-        "submission_status": prediction_status,
+        "validation": {"errors": ";".join(invalid_reasons), "status": prediction_status}
     }
     with open("results.json", "w") as o:
         o.write(json.dumps(result))

main.nf

@@ -50,7 +50,6 @@ process GET_SUBMISSIONS {
 
 // runs docker containers
 process RUN_DOCKER {
-    debug true
     secret "SYNAPSE_AUTH_TOKEN"
     cpus "${cpus}"
     memory "${memory}"
@@ -62,6 +61,7 @@ process RUN_DOCKER {
     path staged_path
     val cpus
     val memory
+    val ready
 
     output:
     tuple val(submission_id), path('predictions.csv')
@@ -77,7 +77,9 @@ process RUN_DOCKER {
 include { UPDATE_SUBMISSION_STATUS as UPDATE_SUBMISSION_STATUS_BEFORE_RUN } from './modules/update_submission_status.nf'
 include { UPDATE_SUBMISSION_STATUS as UPDATE_SUBMISSION_STATUS_AFTER_RUN } from './modules/update_submission_status.nf'
 include { UPDATE_SUBMISSION_STATUS as UPDATE_SUBMISSION_STATUS_AFTER_VALIDATE } from './modules/update_submission_status.nf'
+include { UPDATE_SUBMISSION_STATUS as UPDATE_SUBMISSION_STATUS_AFTER_SCORE } from './modules/update_submission_status.nf'
 include { VALIDATE } from './modules/validate.nf'
+include { SCORE } from './modules/score.nf'
 
 workflow {
     SYNAPSE_STAGE(params.input_id)
@@ -86,9 +88,11 @@ workflow {
     image_ch = GET_SUBMISSIONS.output 
         .splitCsv(header:true) 
         .map { row -> tuple(row.submission_id, row.image_id) }
-    UPDATE_SUBMISSION_STATUS_BEFORE_RUN(image_ch.map { tuple(it[0], "EVALUATION_IN_PROGRESS") })
-    RUN_DOCKER(image_ch, staged_path, params.cpus, params.memory)
-    UPDATE_SUBMISSION_STATUS_AFTER_RUN(RUN_DOCKER.output.map { tuple(it[0], "ACCEPTED") })
+    UPDATE_SUBMISSION_STATUS_BEFORE_RUN(image_ch.map { tuple(it[0], "EVALUATION_IN_PROGRESS") }, "ready")
+    RUN_DOCKER(image_ch, staged_path, params.cpus, params.memory, UPDATE_SUBMISSION_STATUS_BEFORE_RUN.output)
+    UPDATE_SUBMISSION_STATUS_AFTER_RUN(RUN_DOCKER.output.map { tuple(it[0], "ACCEPTED") }, UPDATE_SUBMISSION_STATUS_BEFORE_RUN.output)
     VALIDATE(RUN_DOCKER.output)
-    UPDATE_SUBMISSION_STATUS_AFTER_VALIDATE(VALIDATE.output.map { tuple(it[0], it[2]) })
+    UPDATE_SUBMISSION_STATUS_AFTER_VALIDATE(VALIDATE.output.map { tuple(it[0], it[2]) }, UPDATE_SUBMISSION_STATUS_AFTER_RUN.output)
+    SCORE(VALIDATE.output)
+    UPDATE_SUBMISSION_STATUS_AFTER_SCORE(SCORE.output.map { tuple(it[0], it[2]) }, UPDATE_SUBMISSION_STATUS_AFTER_VALIDATE.output)
 }

modules/score.nf

@@ -0,0 +1,16 @@
+// validate submission results
+process SCORE {
+    secret "SYNAPSE_AUTH_TOKEN"
+    container "python:3.12.0rc1"
+
+    input:
+    tuple val(submission_id), path(predictions), val(status), path(results)
+
+    output:
+    tuple val(submission_id), path(predictions), stdout, path("results.json")
+
+    script:
+    """
+    score.py '${predictions}' '${results}' '${status}'
+    """
+}

modules/update_submission_status.nf

@@ -5,6 +5,10 @@ process UPDATE_SUBMISSION_STATUS {
 
     input:
     tuple val(submission_id), val(new_status)
+    val ready
+
+    output:
+    val ready
 
     script:
     """

modules/validate.nf

@@ -7,7 +7,7 @@ process VALIDATE {
     tuple val(submission_id), path(predictions)
 
     output:
-    tuple val(submission_id), path(predictions), stdout, path("*.json")
+    tuple val(submission_id), path(predictions), stdout, path("results.json")
 
     script:
     """