Add ingest and split subworkflow

compress_pipeline.cwl

@@ -5,45 +5,44 @@ label: compress_pipeline.cwl
 inputs:
   - id: flag_autocorrelation
     type: boolean?
-  - id: surls
-    type: 'string[]'
+  - id: msin
+    type: Directory
 outputs:
-  - id: output
+  - id: inspect
     outputSource:
       - inspect_flagging_dataloss/output
-    type: 'File[]'
+    type: 'File'
   - id: logfile
     outputSource:
       - dppp/logfile
-    type:
-      type: array
-      items:
-        items: File
-        type: array
+    type: File[]
   - id: compressed
     outputSource:
       - compress/compressed
-    type: File[]
+    type: File
+  - id: ingest
+    outputSource:
+      - format_ingest/ingest
+    type: Any
 steps:
-  - id: fetch_data
+  - id: extract_sip_meta
     in:
-      - id: surl_link
-        source: surls
+      - id: msin
+        source: dppp/msout
+      - id: compressed_file
+        source: compress/compressed
     out:
-      - id: uncompressed
-    run: steps/fetch_data.cwl
-    label: fetch_data
-    scatter:
-      - surl_link
+      - id: ingest
+    run: steps/extract_sip_meta.cwl
   - id: dppp
     in:
       - id: parset
         source: define_parset/output
       - id: msin
-        source: fetch_data/uncompressed
+        source: msin
       - id: msout_name
-        source: fetch_data/uncompressed
-        valueFrom: '$("COMPRESSED_" + self[0].basename)'
+        source: msin
+        valueFrom: '$("COMPRESSED_" + self.basename)'
       - id: writefullresflag
         default: true
       - id: storagemanager
@@ -54,8 +53,15 @@ steps:
       - id: msout
       - id: logfile
     run: steps/DPPP.cwl
-    scatter:
-      - msin
+  - id: format_ingest
+    in:
+      - id: metadata
+        source: extract_sip_meta/ingest
+      - id: output_name
+        default: compressed
+    out:
+      - id: ingest
+    run: steps/format_ingest.cwl
   - id: define_parset
     in:
       - id: flag_autocorrelation
@@ -67,13 +73,11 @@ steps:
   - id: inspect_flagging_dataloss
     in:
       - id: input
-        source: fetch_data/uncompressed
+        source: msin
     out:
       - id: output
     run: steps/inspect_flagging_dataloss.cwl
     label: inspect_flagging_dataloss
-    scatter:
-      - input
   - id: compress
     in:
       - id: directory
@@ -82,9 +86,6 @@ steps:
       - id: compressed
     run: steps/compress.cwl
     label: compress
-    scatter:
-      - directory
 requirements:
-  - class: ScatterFeatureRequirement
   - class: StepInputExpressionRequirement
   - class: InlineJavascriptRequirement

download_and_compress_pipeline.cwl

+class: Workflow
+cwlVersion: v1.0
+id: compress_pipeline_cwl
+label: compress_pipeline.cwl
+inputs:
+  - id: flag_autocorrelation
+    type: boolean?
+  - id: surls
+    type: 'string[]'
+outputs:
+  - id: compressed
+    outputSource:
+      - compress/compressed
+    type: File[]
+  - id: inspect
+    outputSource:
+      - compress/inspect
+    type: File[]
+  - id: logfile
+    outputSource:
+      - compress/logfile
+    type:
+      type: array
+      items:
+        type: array
+        items: File
+  - id: ingest
+    outputSource:
+      - compress/inspect
+    type: Any[]
+steps:
+  - id: fetch_data
+    in:
+      - id: surl_link
+        source: surls
+    out:
+      - id: uncompressed
+    run: steps/fetch_data.cwl
+    label: fetch_data
+    scatter:
+      - surl_link
+  - id: compress
+    in:
+      - id: msin
+        source: fetch_data/uncompressed
+      - id: flag_autocorrelation
+        source: flag_autocorrelation
+    scatter:
+      - msin
+    out:
+      - id: compressed
+      - id: inspect
+      - id: logfile
+      - id: ingest
+    run: ./compress_pipeline.cwl
+requirements:
+  - class: ScatterFeatureRequirement
+  - class: SubworkflowFeatureRequirement
+  - class: StepInputExpressionRequirement
+  - class: InlineJavascriptRequirement

steps/DPPP.cwl

@@ -84,7 +84,7 @@ outputs:
       glob: 'DPPP*.log'
 hints:
   - class: DockerRequirement
-    dockerPull: lofareosc/lofar-pipeline:latest
+    dockerPull: astronsdc/lofar-ms-software:latest
 stdout: DPPP.log
 stderr: DPPP_err.log
 requirements:

steps/extract_sip_meta.cwl

+class: CommandLineTool
+cwlVersion: v1.0
+id: inspect_flagging_dataloss
+baseCommand:
+  - bash
+  - script.sh
+inputs:
+  - id: msin
+    type: Directory
+    inputBinding:
+      position: 0
+  - id: compressed_file
+    type: File
+    inputBinding:
+      position: 1
+outputs:
+  - id: ingest
+    type: Any
+    outputBinding:
+      glob: metadata.json
+      loadContents: True
+      outputEval: |
+        ${
+          return JSON.parse(self[0].contents)
+         }
+label: inspect_flagging_dataloss
+hints:
+  - class: DockerRequirement
+    dockerPull: 'astronsdc/lofar-ms-software:latest'
+
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+     - entryname: pipeline_info.json
+       entry: |
+        {
+            "pipelineRun": {
+            "_type": "AveragingPipeline",
+            "demixing": "false",
+            "pipelineName": "compress_pipeline",
+            "strategyName": "",
+            "pipelineVersion": "v01.23",
+            "strategyDescription": "",
+            "timeIntegrationStep": "1",
+            "flagAutoCorrelations": "true",
+            "frequencyIntegrationStep": "1",
+            "numberOfCorrelatedDataProducts": "1"
+          }
+         }
+     - entryname: dataproduct.json
+       entry: |
+         {
+            "dataProduct":
+              {
+                "fileName": $(inputs.compressed_file.basename),
+                "size": $(inputs.compressed_file.size)
+              }
+         }
+     - entryname: script.sh
+       entry: |
+         md5=`md5sum $2 | awk '{ print $1 }'`
+         cat > md5_sum.json << EOF
+         {
+             "dataProduct":
+               {
+                 "fileName": $(inputs.compressed_file.basename),
+                 "size": $(inputs.compressed_file.size),
+                 "md5checksum": $md5
+               }
+         }
+         EOF
+         lofar_sip_from_ms.py --json metadata.json $1
+         lofar_sip_merge.py metadata.json pipeline_info.json --json metadata.json
+         lofar_sip_merge.py metadata.json dataproduct.json --json metadata.json
+         lofar_sip_merge.py metadata.json md5_sum.json --json metadata.json

steps/format_ingest.cwl

+cwlVersion: v1.2
+class: ExpressionTool
+inputs:
+  - id: metadata
+    type: Any
+
+  - id: output_name
+    type: string
+
+outputs:
+  - id: ingest
+    type: Any
+requirements:
+  - class: InlineJavascriptRequirement
+expression: |
+  ${
+    return { "ingest": {
+        "path": inputs.output_name,
+        "metadata": inputs.metadata
+      }
+    }
+  }

steps/inspect_flagging_dataloss.cwl

@@ -8,7 +8,7 @@ baseCommand:
   - script.py
 inputs:
   - id: input
-    type: Directory?
+    type: Directory
     inputBinding:
       position: 0
       shellQuote: false