Adding scripts to aggregate data and edc-client references

workflows/iderha-aggregate/aggregate-central.cwl.yml

@@ -28,8 +28,9 @@ inputs:
 
 outputs:
   - id: final
-    type: stdout
+    type: File
     doc: "The aggregated values from all partial data"
+    outputBinding:
+      glob: avg-ages.json
 
-stdout: stdout.txt
-baseCommand: ["python", "aggregate-central.py"]
+baseCommand: ["aggregate_central"]

workflows/iderha-aggregate/aggregate-pulldata.cwl.yml

@@ -7,7 +7,7 @@ doc: "Task to pull data from EDC"
 
 hints:
   - class: DockerRequirement
-    dockerPull: gitlab.lcsb.uni.lu:4567/luca.bolzani/iderha-test-deployment/test-aggregate-remote
+    dockerPull: gitlab.lcsb.uni.lu:4567/luca.bolzani/iderha-test-deployment/edc-client
 
 inputs:
   - id: input
@@ -16,8 +16,9 @@ inputs:
       position: 1
 
 outputs:
-  - id: datalink
-    type: stdout
+  datalink:
+    type: File
+    outputBinding:
+      glob: aggregated-ages.json
 
-stdout: stdout.txt
-baseCommand: ["aggregate_remote"]
+baseCommand: ["/app/edc_client.sh"]

workflows/resources/aggregate-central.Dockerfile

 FROM repomanager.lcsb.uni.lu:9999/python:3.9
 LABEL authors="francois.ancien"
 
+WORKDIR /
 COPY aggregate-central.py /aggregate-central.py
+COPY aggregate-central.sh /usr/local/bin/aggregate_central
+
+RUN chmod +x /usr/local/bin/aggregate_central
 
 ENTRYPOINT ["/usr/bin/bash"]
 

workflows/resources/aggregate-central.py

 #!/usr/bin/env python3.9
 
-def main():
-    print("Hello world! From central node!")
+import argparse
+import json
+import sys
+
+def main(args=None):
+    if args is None:
+        args = sys.argv[1:]
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("infiles", metavar="Input", type=str, nargs="+", help="PATH to an input file")
+
+    args = parser.parse_args(args)
+
+    total_sum = 0.0
+    total_count = 0.0
+    for file_path in args.infiles:
+        with open(file_path, "r") as f:
+            data = json.load(f)
+            total_sum += data["sum"]
+            total_count += data["count"]
+
+    with open("avg-ages.json", "w") as f:
+        json.dump({"avg": total_sum / total_count}, f)
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

workflows/resources/aggregate-central.sh

+#!/bin/bash
+python /aggregate-central.py $1
\ No newline at end of file

workflows/resources/aggregate-remote.Dockerfile

 FROM repomanager.lcsb.uni.lu:9999/python:3.9
 LABEL authors="francois.ancien"
 
-RUN pip install requests --no-cache
+RUN pip install --no-cache requests
 WORKDIR /
 COPY aggregate-remote.py /aggregate-remote.py
 COPY aggregate-remote.sh /usr/local/bin/aggregate_remote

workflows/resources/aggregate-remote.py

 #!/usr/bin/env python
 
 import argparse
-import requests
+import json
+from ftplib import FTP
 import sys
 
 
@@ -9,20 +10,47 @@ def main(args=None):
     if args is None:
         args = sys.argv[1:]
     parser = argparse.ArgumentParser()
-    parser.add_argument("-i", "--input", help="URL to input file")
+    parser.add_argument("-i", "--input", help="PATH to input file")
 
     parsed = parser.parse_args(args)
 
-    res = requests.get(parsed.input)
-    if not res.ok:
-        issue = res.text
-        raise requests.HTTPError(f"Error {res.status_code}: {issue}")
-
-    try:
-        data = res.json()
-        print(data)
-    except requests.JSONDecodeError:
-        print(f"Issue with data in {res.text}. Not a valid json")
+    data = None
+    ftp_path = None
+    with open(parsed.input, "r") as f:
+        try:
+            data = json.load(f)
+            ftp_path = data["FTP_DATA_ADDRESS"]
+        except json.JSONDecodeError as e:
+            raise json.JSONDecodeError(f"Impossible to parse data in {parsed.input}. Not a valid json.") from e
+
+    # Getting data from ftp source
+    ftp_host = "http://ftp-upload-service/api/ftp/upload"
+    ftp = FTP(ftp_host)
+    ftp.login(user="ftp_iderha_user", passwd="ftp_iderha_pass")
+    tmp_path = "tmp_data.json"
+    with open(tmp_path, "wb") as fh:
+        ftp.retrbinary(ftp_path, fh.write)
+
+    # Extracting data from json
+    with open(tmp_path, "rb") as f:
+        try:
+            data = json.load(f)
+        except json.JSONDecodeError as e:
+            raise json.JSONDecodeError(f"Impossible to parse data in {tmp_path}. Not a valid json.") from e
+
+    # Calculating avg username lengths
+    sum_usrname = 0
+    for row in data:
+        try:
+            sum_usrname += len(row["username"])
+        except (AttributeError, KeyError):
+            continue
+
+    count_rows = len(data)
+
+    # Saving
+    with open("aggregated-ages.json", "w") as f:
+        json.dump({"sum": sum_usrname, "count": count_rows}, f)
 
 
 if __name__ == "__main__":

workflows/resources/iderha-edc-input.json

@@ -10,5 +10,6 @@
     "EDC_PROVIDER_PROTOCOL_URL": "http://edc-provider:19194/protocol",
     "EDC_PROVIDER_PUBLIC_URL": "http://edc-provider:19291/public",
     "EDR_ENDPOINT_URL": "http://edc-consumer:29191/api/edr/query",
-    "DATA_DESTINATION_URL": "http://ftp-upload-service/api/ftp/upload"
+    "DATA_DESTINATION_URL": "http://ftp-upload-service/api/ftp/upload",
+    "FTP_DATA_ADDRESS": "/home/ftp_iderha_user/isst-data.json"
 }