Merge branch 'sort_phaseup' into 'master'

Restructured the sorting and concatenating of subbands.

See merge request !5

doc/pipeline.yaml

@@ -335,10 +335,9 @@ prep_delay_dir:
              - dpppconcat bands
              - "{{ best_delay_cat }} catalogue with information on in-field calibrator"
     Operation: >
-               get information from the catalogue
-               on the source name, RA, and DEC,
-               and return that information formatted
-               for use by NDPPP
+               get information from the catalogue on the 
+               source name, right ascension (RA), and declination (DEC),
+               and return that information formatted for use by NDPPP
     Software : plugins/Pipelinestep_TargetListToMapfile.py
     Output   : >
                variables that are the source name,
@@ -359,8 +358,8 @@ dppp_phaseup:
 dppp_phaseup_list:
     Input    : dppp_phaseup mapfile
     Operation: >
-               get a list with length 1 containing info
-               on all bands for the in-field calibrator
+               "[legacy] get a list with length 1 containing info
+               on all bands for the in-field calibrator"
     Software : prefactor/plugins/createMapfile.py
     Output   : >
                a mapfile with info on all bands
@@ -382,7 +381,8 @@ sort_phaseupmap:
 
 do_phaseup_maps:
     Input    : results of sort_phaseupmap
-    Operation: makes sensible files for future use out of the input
+    Operation: >
+               "[legacy] makes sensible files for future use out of the input"
     Software : prefactor/plugins/Pipelinestep_mapfilenamesFromMapfiles
     Output   : mapfiles for use in phaseup_concat
 

scripts/TargetListToCoords.py

+import os
+import numpy as np
+import pyrap.tables, math
+from astropy import units as u
+from astropy.coordinates import SkyCoord
+from astropy.table import Table
+
+def plugin_main(**kwargs):
+    """
+    Takes in a catalogue with a target and returns appropriate coordinates
+    
+    Parameters
+    ----------
+    filename: str
+        Name of output mapfile
+    target_file: str
+        file containing target info
+
+    Returns
+    -------
+    result : dict
+        Output coordinates
+    
+    """
+    # parse the input
+    target_file = kwargs['target_file']
+    
+    # read in the catalogue to get source_id, RA, and DEC
+    t = Table.read(target_file, format='csv')
+    RA_val = t['RA_LOTSS'].data[0]
+    DEC_val = t['DEC_LOTSS'].data[0]
+    Source_id = t['Source_id'].data[0]
+    if str(Source_id)[0:1] == 'I':
+        pass
+    elif str(Source_id)[0:1] == 'S':
+        pass
+    else:
+        Source_id = 'S' + str(Source_id)
+    # make a string of coordinates for the NDPPP command
+    ss = '["' + str(RA_val) + 'deg","' + str(DEC_val) + 'deg"]'
+    result = {'name' : Source_id, 'coords' : ss}
+    return result

steps/dp3_phaseup.cwl

+class: CommandLineTool
+cwlVersion: v1.2
+id: dp3_phaseup
+label: dp3_phaseup
+
+baseCommand: DP3
+
+arguments:
+    - 'steps=[shift, average1, applybeam, average2]'
+    - shift.type=phaseshift
+    - average1.type=averager
+    - applybeam.type=applybeam
+    - average2.type=averager
+    - numthreads=5
+    - msout.overwrite=True
+    - applybeam.beammode=full
+    - average2.freqresolution=390.56kHz
+    - average2.timeresolution=32.0
+
+inputs:
+    - id: msin
+      type: Directory
+      doc: Input measurement set.
+      inputBinding:
+        position: 0
+        prefix: msin=
+        separate: false
+        shellQuote: false
+    - id: msout_name
+      type: string?
+      default: "dp3-phaseup-"
+      inputBinding:
+        position: 0
+        prefix: msout=
+        separate: false
+        valueFrom: |
+          $(self + inputs.msin.basename)
+    - id: storagemanager
+      type: string?
+      doc:
+      default: 'dysco'
+      inputBinding:
+        position: 1
+        prefix: msout.storagemanager=
+        separate: false
+        shellQuote: false
+    - id: datacolumn_in
+      type: string?
+      default: 'DATA'
+      doc: Data column input measurement set.
+      inputBinding:
+        position: 1
+        prefix: msin.datacolumn=
+        separate: false
+        shellQuote: false
+    - id: datacolumn_out
+      type: string?
+      default: 'DATA'
+      doc: Data column output measurement set.
+      inputBinding:
+        position: 1
+        prefix: msout.datacolumn=
+        separate: false
+        shellQuote: false
+    - id: phase_center
+      type: string
+      doc: 'source RA and DEC.'
+      inputBinding:
+        position: 1
+        prefix: shift.phasecenter=
+    - id: freqresolution
+      type: string?
+      default: '48.82kHz'
+      inputBinding:
+        position: 1
+        prefix: average1.freqresolution=
+    - id: timeresolution
+      type: float?
+      default: 4.0
+      inputBinding:
+        position: 1
+        prefix: average1.timeresolution=
+    - id: beam_direction
+      type: string
+      doc: 'source RA and DEC.'
+      inputBinding:
+        position: 1
+        prefix: applybeam.direction=
+
+outputs:
+    - id: msout
+      type: Directory
+      outputBinding:
+        glob: $("dp3-phaseup-" + inputs.msin.basename)
+    - id: logfile
+      type: File
+      outputBinding:
+        glob: 'dp3_phaseup.log'
+    - id: errorfile
+      type: File
+      outputBinding:
+        glob: 'dp3_phaseup_err.log'
+
+hints:
+  DockerRequirement:
+    dockerPull: vlbi-cwl
+
+requirements:
+  - class: InlineJavascriptRequirement
+
+stdout: dp3_phaseup.log
+stderr: dp3_phaseup_err.log

steps/prep_delay.cwl

+class: CommandLineTool
+cwlVersion: v1.2
+id: prep_delay
+label: prep_delay
+
+baseCommand:
+  - python3
+  - prep_delay.py
+
+inputs:
+    - id: delay_calibrator
+      type: File
+      doc: file containing target info.
+
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: prep_delay.py
+        entry: |
+          import sys
+          import json
+          from TargetListToCoords import plugin_main as targetListToCoords
+
+          mss = sys.argv[1:]
+          inputs = json.loads(r"""$(inputs)""")
+
+          target_file = inputs['delay_calibrator']['path']
+
+          output = targetListToCoords(target_file=target_file)
+
+          coords = output['coords']
+          name = output['name']
+
+          cwl_output = {}
+          cwl_output['coords'] = coords
+          cwl_output['name'] = name
+
+          with open('./out.json', 'w') as fp:
+              json.dump(cwl_output, fp)
+
+outputs:
+    - id: coordinates
+      type: string
+      doc: Catalogue source coordinates.
+      outputBinding:
+        loadContents: true
+        glob: 'out.json'
+        outputEval: $(JSON.parse(self[0].contents).coords)
+    - id: logfile
+      type: File[]
+      outputBinding:
+        glob: 'prep_delay*.log'
+
+hints:
+  DockerRequirement:
+    dockerPull: vlbi-cwl
+
+stdout: prep_delay.log
+stderr: prep_delay_err.log

workflows/concatenate-flag.cwl

@@ -6,9 +6,33 @@ label: sort-concat-flag
 inputs:
   - id: msin
     type: Directory[]
+  - id: numbands
+    type: int?
+    default: 10
+    doc: The number of files that have to be grouped together.
+  - id: DP3fill
+    type: boolean?
+    default: True
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: stepname
+    type: string?
+    default: '.dp3-concat'
+    doc: Add this stepname into the file names of the output files.
+  - id: mergeLastGroup
+    type: boolean?
+    default: False
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: truncateLastSBs
+    type: boolean?
+    default: True
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: firstSB
+    type: int?
+    default: null
+    doc: If set, reference the grouping of files to this station subband.
 
 steps:
-  - id: sort_concatmap
+  - id: sort_concatenate
     in:
       - id: msin
         source: msin
@@ -24,9 +48,9 @@ steps:
         source:
           - msin
       - id: group_id
-        source: sort_concatmap/groupnames
+        source: sort_concatenate/groupnames
       - id: groups_specification
-        source: sort_concatmap/filenames
+        source: sort_concatenate/filenames
     out:
       - id: msout
       - id: aoflag_logfile
@@ -61,7 +85,7 @@ steps:
       - id: files
         linkMerge: merge_flattened
         source:
-            - sort_concatmap/logfile
+            - sort_concatenate/logfile
             - concatenate_logfiles_concatenate/output
             - concatenate_logfiles_aoflagging/output
       - id: sub_directory_name

workflows/phaseup-concat.cwl

+class: Workflow
+cwlVersion: v1.2
+id: phaseup-concat
+label: phaseup-concat
+
+inputs:
+  - id: msin
+    type: Directory[]
+    doc: Input measurement sets.
+  - id: delay_calibrator
+    type: File
+    doc: Coordinates of best delay calibrator.
+  - id: numbands
+    type: int?
+    default: -1
+    doc: The number of files that have to be grouped together.
+  - id: DP3fill
+    type: boolean?
+    default: True
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: stepname
+    type: string?
+    default: '.dp3-phaseup-concat'
+    doc: Add this stepname into the file names of the output files.
+  - id: mergeLastGroup
+    type: boolean?
+    default: False
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: truncateLastSBs
+    type: boolean?
+    default: False
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: firstSB
+    type: int?
+    default: null
+    doc: If set, reference the grouping of files to this station subband.
+  - id: do_flagging
+    type: boolean?
+    default: false
+
+steps:
+  - id: prep_delay
+    in:
+      - id: delay_calibrator
+        source: delay_calibrator
+    out:
+      - id: coordinates
+      - id: logfile
+    run: ../steps/prep_delay.cwl
+    label: prep_delay
+  - id: dp3_phaseup
+    in:
+      - id: msin
+        source: msin
+      - id: phase_center
+        source: prep_delay/coordinates
+      - id: beam_direction
+        source: prep_delay/coordinates
+    out:
+      - id: msout
+      - id: logfile
+      - id: errorfile
+    run: ../steps/dp3_phaseup.cwl
+    scatter: msin
+    label: dp3_phaseup
+  - id: sort_concatenate
+    in:
+      - id: msin
+        source: dp3_phaseup/msout
+      - id: numbands
+        source: numbands
+      - id: DP3fill
+        source: DP3fill
+      - id: stepname
+        source: stepname
+      - id: mergeLastGroup
+        source: mergeLastGroup
+      - id: truncateLastSBs
+        source: truncateLastSBs
+      - id: firstSB
+        source: firstSB
+    out:
+      - id: filenames
+      - id: groupnames
+      - id: logfile
+    run: ../steps/sort_concatmap.cwl
+    label: sort_concatmap
+  - id: phaseup_concatenate
+    in:
+      - id: msin
+        source:
+          - dp3_phaseup/msout
+      - id: group_id
+        source: sort_concatenate/groupnames
+      - id: groups_specification
+        source: sort_concatenate/filenames
+      - id: do_flagging
+        source: do_flagging
+    out:
+      - id: msout
+      - id: concatenate_logfile
+      - id: aoflag_logfile
+    run: ./subworkflows/concatenation.cwl
+    scatter: group_id
+    label: phaseup_concatenate
+  - id: concat_logfiles_phaseup
+    label: concat_logfiles_phaseup
+    in:
+      - id: file_list
+        linkMerge: merge_flattened
+        source:
+          - dp3_phaseup/logfile
+      - id: file_prefix
+        default: dp3_phaseup
+    out:
+      - id: output
+    run: ../steps/concatenate_files.cwl
+
+
+  - id: concat_logfiles_concatenate
+    label: concat_logfiles_concatenate
+    in:
+      - id: file_list
+        linkMerge: merge_flattened
+        source:
+          - phaseup_concatenate/concatenate_logfile
+      - id: file_prefix
+        default: phaseup_concatenate
+    out:
+      - id: output
+    run: ../steps/concatenate_files.cwl
+
+  - id: save_logfiles
+    in:
+      - id: files
+        linkMerge: merge_flattened
+        source:
+          - prep_delay/logfile
+          - concat_logfiles_phaseup/output 
+          - sort_concatenate/logfile
+          - concat_logfiles_phaseup/output
+      - id: sub_directory_name
+        default: phaseup
+    out:
+      - id: dir
+    run: ../steps/collectfiles.cwl
+    label: save_logfiles
+
+outputs:
+  - id: msout
+    outputSource: phaseup_concatenate/msout
+    type: Directory[]
+  - id: logdir
+    outputSource: save_logfiles/dir
+    type: Directory
+
+requirements:
+  - class: SubworkflowFeatureRequirement
+  - class: ScatterFeatureRequirement
+  #- class: StepInputExpressionRequirement
+  #- class: InlineJavascriptRequirement
+  - class: MultipleInputFeatureRequirement

workflows/subworkflows/concatenation.cwl

@@ -10,6 +10,9 @@ inputs:
     type: string
   - id: groups_specification
     type: File
+  - id: do_flagging
+    type: boolean?
+    default: true
 
 steps:
   - id: filter_ms_group
@@ -41,9 +44,12 @@ steps:
     in:
       - id: msin
         source: dp3_concat/msout
+      - id: do_flagging
+        source: do_flagging
     out:
       - id: msout
       - id: logfile
+    when: $(inputs.do_flagging)
     run: ../../steps/aoflagger.cwl
     label: AOflagging
   - id: concat_logfiles_aoflagging
@@ -53,8 +59,11 @@ steps:
         source: AOflagging/logfile
       - id: file_prefix
         default: AOflagging
+      - id: do_flagging
+        source: do_flagging
     out:
       - id: output
+    when: $(inputs.do_flagging)
     run: ../../steps/concatenate_files.cwl
     label: concat_logfiles_AOflagging
   - id: dp3_concatenate_logfiles
@@ -71,11 +80,16 @@ steps:
 
 outputs:
   - id: msout
-    outputSource: AOflagging/msout
+    outputSource:
+        - AOflagging/msout
+        - dp3_concat/msout
+    pickValue: first_non_null
     type: Directory
   - id: concatenate_logfile
     outputSource: dp3_concatenate_logfiles/output
     type: File
   - id: aoflag_logfile
-    outputSource: concat_logfiles_aoflagging/output
+    outputSource:
+        - concat_logfiles_aoflagging/output
+    pickValue: all_non_null
     type: File

workflows/subworkflows/sort-concatenate.cwl

+class: Workflow
+cwlVersion: v1.2
+id: sort-concatenate
+label: sort-concatenate
+
+inputs:
+  - id: msin
+    type: Directory[]
+  - id: numbands
+    type: int?
+    default: 10
+    doc: The number of files that have to be grouped together.
+  - id: DP3fill
+    type: boolean?
+    default: True
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: stepname
+    type: string?
+    default: '.dp3-concat'
+    doc: Add this stepname into the file names of the output files.
+  - id: mergeLastGroup
+    type: boolean?
+    default: False
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: truncateLastSBs
+    type: boolean?
+    default: True
+    doc: Add dummy file names for missing frequencies, so that DP3 can fill the data with flagged dummy data.
+  - id: firstSB
+    type: int?
+    default: null
+    doc: If set, reference the grouping of files to this station subband.
+
+steps:
+  - id: sort_concatmap
+    in:
+      - id: msin
+        source: msin
+      - id: numbands
+        source: numbands
+      - id: DP3fill
+        source: DP3fill
+      - id: stepname
+        source: stepname
+      - id: mergeLastGroup
+        source: mergeLastGroup
+      - id: truncateLastSBs
+        source: truncateLastSBs
+      - id: firstSB
+        source: firstSB
+    out:
+      - id: filenames
+      - id: groupnames
+      - id: logfile
+    run: ../../steps/sort_concatmap.cwl
+    label: sort_concatmap
+  - id: concatenate
+    in:
+      - id: msin
+        source:
+          - msin
+      - id: group_id
+        source: sort_concatmap/groupnames
+      - id: groups_specification
+        source: sort_concatmap/filenames
+    out:
+      - id: msout
+      - id: logfile
+    run: ./concatenation.cwl
+    scatter: group_id
+    label: concatenation
+  - id: concatenate_logfiles_concatenate
+    in:
+      - id: file_list
+        source:
+          - concatenate/logfile
+      - id: file_prefix
+        default: concatenate
+    out:
+      - id: output
+    run: ../../steps/concatenate_files.cwl
+    label: concatenate_logfiles_concatenate
+
+outputs:
+    - id: logfile_concatenate
+      outputSource: concatenate_logfiles_concatenate/output
+      type: File
+    - id: logfile_sortconcat
+      outputSource: sort_concatmap/logfile
+      type: File
+    - id: msout
+      outputSource: concatenate/msout
+      type: Directory[]
+
+requirements:
+    - class: SubworkflowFeatureRequirement
+    - class: ScatterFeatureRequirement
+    - class: MultipleInputFeatureRequirement