Integrate with ATDB execution framework

examples/full_parset.sh

+#!/bin/sh -eu
+
+# Grep for keys, optional (commented) keys, and sections;
+# uncomment and convert ini to json
+egrep '^((#[[:blank:]]*)?[[:alnum:]_]* =|\[[^]]*\])' rapthor.parset \
+  | sed 's,^# *,,' \
+  > rapthor_full.parset

examples/ini2json.sh

+#!/bin/sh -e
+#
+# Convert an INI-file to a JSON-file.
+
+if [ $# -lt 1 -o $# -gt 2 ]
+then
+  echo "Usage: ${0} <ini-file> [<json-file>]"
+  exit 1
+fi
+
+INPUT="${1}"
+OUTPUT="${2:-${INPUT}.json}"
+
+# Check if input file exists. Bail out if not.
+if ! [ -f "${INPUT}" ]
+then
+  echo "${INPUT}: No such file or directory"
+  exit 1
+fi
+
+# Check if `jc`, a tool for converting INI to JSON, is available.
+# If not, install it in a virtual environment.
+if ! $(which jc > /dev/null)
+then
+  # Are we inside a virtual environment? If not, create one and activate it.
+  if [ -z "${VIRTUAL_ENV}" ]
+  then
+    ROOT="$(git rev-parse --show-toplevel)"
+    python3 -m venv "${ROOT}/venv"
+    . "${ROOT}/venv/bin/activate"
+  fi
+  pip install --disable-pip-version-check --quiet --require-virtualenv jc
+fi
+
+# Do the actual conversion from INI to JSON
+cat "${INPUT}" | jc --ini -mp > "${OUTPUT}"

rapthor/pipeline/execution/examples/hba_nl/hba_nl.json

 {
   "surls": [
     "gsiftp://gridftp.grid.sara.nl/pnfs/grid.sara.nl/data/lofar/user/disk/rapthor/testing/L667526_121MHz_uv_pre-cal.ms.tar",
-    "gsiftp://gridftp.grid.sara.nl/pnfs/grid.sara.nl/data/lofar/user/disk/rapthor/testing/L667526_123MHz_uv_pre-cal.ms.tar"
+    "gsiftp://gridftp.grid.sara.nl/pnfs/grid.sara.nl/data/lofar/user/disk/rapthor/testing/L667526_123MHz_uv_pre-cal.ms.tar",
+    "gsiftp://gridftp.grid.sara.nl/pnfs/grid.sara.nl/data/lofar/user/disk/rapthor/testing/L667526_125MHz_uv_pre-cal.ms.tar",
+    "gsiftp://gridftp.grid.sara.nl/pnfs/grid.sara.nl/data/lofar/user/disk/rapthor/testing/L667526_127MHz_uv_pre-cal.ms.tar"
   ],
   "settings": {
-    "class": "File",
-    "path": "L667526.config.json"
+    "global": {
+      "regroup_input_skymodel": "True",
+      "selfcal_data_fraction": "0.01",
+      "strategy": "/project/rapthor/Data/rapthor/L667526/custom_calibration_strategy.py"
+    },
+    "calibration": {
+    },
+    "imaging": {
+        "dde_method": "facets"
+    },
+    "cluster": {
+      "use_container": "True",
+      "container_type": "singularity"
+    }
+  },
+  "virtualenv": {
+    "python": {
+      "version": "3.9"
+    },
+    "rapthor": {
+      "version": "atdb_integration_test",
+      "requirements": []
+    }
   }
 }

rapthor/pipeline/execution/hba_nl.cwl

@@ -6,6 +6,7 @@ doc: |
   This is the top-level workflow for the Rapthor HBA NL pipeline.
 
 requirements:
+- class: MultipleInputFeatureRequirement
 - class: ScatterFeatureRequirement
 - class: SubworkflowFeatureRequirement
 
@@ -14,51 +15,30 @@ inputs:
   doc: List of SURLs to the input data
   type: string[]
 - id: settings
-  doc: File containing the settings for the Rapthor pipeline in JSON format
-  type: File
-- id: skymodel
-  type: File?
-  doc: Optional input sky model
-- id: apparent_skymodel
-  type: File?
-  doc: Optional apparent sky model
-- id: strategy
+  doc: Pipeline settings, used to generate a parset file
   type:
-  - File?
-  - string?
-  doc: |
-    Optional strategy; either a name (e.g., "selfcal"), or a path to a python
-    strategy file (e.g., "/path/to/my_fancy_strategy.py")
+    type: record
+    fields:
+      - name: global
+        type: Any
+      - name: calibration
+        type: Any
+      - name: imaging
+        type: Any
+      - name: cluster
+        type: Any
+- id: virtualenv
+  type: Any
 
 outputs:
-- id: images
-  type: Directory[]
-  outputSource:
-    - run_rapthor/images
-- id: logs
-  type: Directory[]
-  outputSource:
-    - run_rapthor/logs
 - id: parset
   type: File
   outputSource:
-    - run_rapthor/parset
-- id: plots
-  type: Directory[]
-  outputSource:
-    - run_rapthor/plots
-- id: regions
-  type: Directory[]
-  outputSource:
-    - run_rapthor/regions
-- id: skymodels
-  type: Directory[]
-  outputSource:
-    - run_rapthor/skymodels
-- id: solutions
-  type: Directory[]
+    run_rapthor/parset
+- id: tarballs
+  type: File[]
   outputSource:
-    - run_rapthor/solutions
+    tar_directory/tarball
 
 steps:
 # fetch MS files
@@ -94,16 +74,12 @@ steps:
     Run the Rapthor pipeline in a virtual environment that is created on
     the fly
   in:
+    - id: msin
+      source: concat_ms/msout
     - id: settings
       source: settings
-    - id: ms
-      source: concat_ms/msout
-    - id: skymodel
-      source: skymodel
-    - id: apparent_sky
-      source: apparent_skymodel
-    - id: strategy
-      source: strategy
+    - id: virtualenv
+      source: virtualenv
   out:
     - id: images
     - id: logs
@@ -113,3 +89,22 @@ steps:
     - id: skymodels
     - id: solutions
   run: run_rapthor.cwl
+
+- id: tar_directory
+  label: Tar the pipeline results
+  doc: |
+    Create tar-balls for each of the output directories produced by Rapthor.
+  in:
+   - id: directory
+     linkMerge: merge_flattened
+     source:
+      - run_rapthor/images
+      - run_rapthor/logs
+      - run_rapthor/plots
+      - run_rapthor/regions
+      - run_rapthor/skymodels
+      - run_rapthor/solutions
+  out:
+  - id: tarball
+  run: tar_directory.cwl
+  scatter: directory

rapthor/pipeline/execution/run_rapthor.cwl

@@ -12,32 +12,30 @@ doc: |
     are put in the parset file need to be the same as the paths used when
     Rapthor is actually being run.
 
-# Input should be a JSON file with the correct pipeline settings. The contents of
-# this file will be converted to a valid Rapthor parset file.
 inputs:
-- id: settings
-  type: File
-  doc: File containing the settings for the Rapthor pipeline in JSON format
-- id: ms
+- id: msin
   type: Directory
   doc: |
     Input Measurement Set
     In principle, Rapthor can handle multiple input Measurement Sets with data
     from different epochs containing the same frequency bands (e.g., from
     multiple nights of observations). This CWL step does _not_ support this.
-- id: skymodel
-  type: File?
-  doc: Optional input sky model
-- id: apparent_sky
-  type: File?
-  doc: Optional apparent sky model
-- id: strategy
+- id: settings
+  doc: Pipeline settings, used to generate a parset file
   type:
-  - File?
-  - string?
-  doc: |
-    Optional strategy; either a name (e.g., "selfcal"), or a path to a python
-    strategy file (e.g., "/path/to/my_fancy_strategy.py")
+    type: record
+    fields:
+      - name: global
+        type: Any
+      - name: calibration
+        type: Any
+      - name: imaging
+        type: Any
+      - name: cluster
+        type: Any
+- id: virtualenv
+  doc: Description of the virtual environment used to run Rapthor.
+  type: Any
 
 # Anything that the Rapthor pipeline will produce as outputs.
 outputs:
@@ -76,59 +74,36 @@ baseCommand:
 
 requirements:
 - class: InlineJavascriptRequirement
+  expressionLib:
+    - { $include: utils.js}
 - class: InitialWorkDirRequirement
   listing:
-  - entryname: json2ini.jq
+  - entryname: rapthor.parset
+    writable: True
     entry: |
-      # Convert JSON to INI using `jq`
-      # See: https://stackoverflow.com/a/50902853
-      def kv: to_entries[] | "\(.key)=\(.value)";
-      if type == "array" then .[] else . end
-      | to_entries[]
-      | "[\(.key)]", (.value|kv)
+        ${
+            var settings = inputs.settings;
+            settings.global.dir_working = runtime.outdir;
+            settings.global.input_ms = inputs.msin.path;
+            var result = "";
+            ["global", "calibration", "imaging", "cluster"].forEach(element => {
+                result += objectToParsetString(settings[element], element) + "\n\n"
+            });
+            return result;
+        }
 
   - entryname: runner.sh
     entry: |
       #!/bin/bash
       set -ex
 
-      # Compose a filter to update the relevant items in the input JSON file.
-      # NOTE: the JavaScript expressions are *always* evaluated. Hence, the
-      # extra check in the `then`-clause. Also note that `.global.strategy`
-      # can either be a string or a File, hence the ternary expression.
-      filter='.config
-        | .global.dir_working |= "$(runtime.outdir)"
-        | .global.input_ms |= "$(inputs.ms.path)"
-        | .global.input_skymodel |=
-            if $(inputs.skymodel != null)
-            then "$(inputs.skymodel && inputs.skymodel.path)"
-            else empty
-            end
-        | .global.apparent_skymodel |=
-            if $(inputs.apparent_sky != null)
-            then "$(inputs.apparent_sky && inputs.apparent_sky.path)"
-            else empty
-            end
-        | .global.strategy |=
-            if $(inputs.strategy != null)
-            then "$(inputs.strategy && inputs.strategy.path
-                    ? inputs.strategy.path
-                    : inputs.strategy)"
-            else empty
-            end
-      '
-
-      # Read input JSON file, update paths, and write to INI (parset) file
-      jq -c "\${filter}" "$(inputs.settings.path)" | \
-        jq -rf json2ini.jq > rapthor.parset
-
       # Create virtual environment and activate it.
-      python_version=`jq -r .virtualenv.python.version "$(inputs.settings.path)"`
+      python_version="$(inputs.virtualenv.python.version)"
       virtualenv --python="python\${python_version}" venv
       . venv/bin/activate
 
       # Install rapthor
-      rapthor_version=`jq -r .virtualenv.rapthor.version "$(inputs.settings.path)"`
+      rapthor_version="$(inputs.virtualenv.rapthor.version)"
       pip install --no-cache-dir \
         git+https://git.astron.nl/RD/rapthor.git@\${rapthor_version}
 

rapthor/pipeline/execution/tar_directory.cwl

+id: tar_directory
+label: Tar directory contents
+class: CommandLineTool
+cwlVersion: v1.2
+
+doc: |
+    Create a tar-ball of the contents of the input directory.
+
+inputs:
+- id: directory
+  type: Directory
+
+outputs:
+- id: tarball
+  type: File
+  outputBinding:
+    glob: $(inputs.directory.basename).tar.gz
+
+baseCommand:
+  - tar
+
+arguments:
+  - --create
+  - --dereference
+  - --gzip
+  - prefix: --directory
+    valueFrom: $(inputs.directory.dirname)
+  - prefix: --file
+    valueFrom: $(inputs.directory.basename).tar.gz
+  - valueFrom: $(inputs.directory.basename)

rapthor/pipeline/execution/utils.js

+function objectToParsetString(obj, section) {
+    var result = "[" + section + "]"
+    for(var item in obj) {
+        var value = obj[item]
+        var vType = typeof value
+        if(value === null) continue
+        try{
+            if (vType === 'object' && !Array.isArray(value) && value.hasOwnProperty('path')) {
+                value = value.path
+            } else if(vType === 'object' && Array.isArray(value) ) {
+                value = "[" +  value.join(',') + "]"
+            }
+            else if(vType === 'object' && !value.hasOwnProperty('path'))  {
+                throw "Unsupported type"
+            } else {
+                value = String(value)
+            }
+
+            result += "\n" + item + "=" + value
+        } catch(e) {
+            console.error(item, value, e)
+            throw e
+        }
+    }
+    return result
+}