delay-calibration.cwl

class: Workflow
cwlVersion: v1.2
id: delay-calibration
label: VLBI delay calibration
doc: |
    The delay calibration pipeline does the following:
    * [If LINC target solutions are supplied]
      applies LINC solutions to target data and flags
      A-team sources,
    * [If LINC target solutions are supplied]
      concatenates the data in groups of 10, performs
      flagging on the international stations.
    * [If a DDF pipeline SOLSDIR directory is supplied]
      corrects direction-dependent effects for the Dutch stations and
      optionally subtracts the LoTSS model outside a user-specifiable region.
    * creates a MeasurementSet with data phase-shifted
      to a given delay calibrator, calibrated for direction-
      independent effects.

requirements:
  - class: SubworkflowFeatureRequirement
  - class: MultipleInputFeatureRequirement
  - class: StepInputExpressionRequirement
  - class: InlineJavascriptRequirement
  - class: ScatterFeatureRequirement

inputs:
    - id: msin
      type: Directory[]
      doc: The raw data in a MeasurementSet version 2.0 format.

    - id: solset
      type: File?
      doc: |
        The solution tables generated by the LINC target pipeline
        in an HDF5 format.

    - id: delay_calibrator
      type: File?
      doc: |
        A delay calibrator catalogue in CSV format.
        The input CSV should contain the following columns: Observation,Source_id,RA,DEC,Total_flux,Peak_flux
        If not provided, lofar-vlbi-plot will be run to generate the CSV (requires internet access).

    - id: image_catalogue
      type: File?
      doc: An image catalogue file in CSV format.

    - id: Ateam_skymodel
      doc: The skymodel to use in clipping bright sources.
      type: File

    - id: rfi_strategy
      doc: The RFI strategy to use in AOflagging.
      type: File

    - id: filter_baselines
      type: string?
      default: "*&"
      doc: The default filter constraints for the dp3_prep_target step.

    - id: flag_baselines
      type: string[]?
      default: []
      doc: |
        The baselines to be flagged by DP3.
        Can be a pattern, e.g. [ CS013HBA*&&* ].

    - id: phasesol
      type: string?
      default: TGSSphase
      doc: The name of the target solution table to use from the solset input.

    - id: reference_stationSB
      type: int?
      default: 104
      doc: |
        Subbands are concatenated in the concatenate-flag
        workflow relative to this station subband.

    - id: number_cores
      type: int?
      default: 12
      doc: |
        Number of cores to use per job for tasks with
        high I/O or memory.

    - id: max_dp3_threads
      type: int?
      default: 5
      doc: The number of threads per DP3 process.

    - id: ddf_solsdir
      type: Directory?
      doc: |
        [Required if subtracting LoTSS] Path to the SOLSDIR directory
        of the DDF-pipeline run, where most of the calibration solutions
        are stored.

    - id: ddf_rundir
      type: Directory?
      doc: |
        [Required if subtracting LoTSS] Path to the directory of the
        DDF-pipeline run where files required for the subtract can be found.

    - id: box_size
      type: float?
      default: 2.5
      doc: |
        [Required if subtracting LoTSS] Box size, in degrees, outside of which to subtract
        the LoTSS model from the data.

    - id: subtract_chunk_hours
      type: float?
      default: 0.5
      doc: |
        The range of time to predict the LoTSS model for at once. Lowering this value reduces
        memory footprint at the (possible) cost of increased runtime and vice versa.

    - id: do_subtraction
      type: boolean?
      default: false
      doc: When set to true, the LoTSS model will be subtracted from the DDF corrected data.

    - id: model_image
      type: File?
      doc: Image to generate an initial delay calibration model from.

    - id: apply_delay_solutions
      type: boolean?
      default: true
      doc: When set to true, the delay calibration solutions will be applied on the full MS.

    - id: do_validation
      type: boolean?
      default: true
      doc: Validate DI calibration, using merged output calibration solutions.

    - id: rm_correction
      type:
        type: enum
        symbols:
          - "RMextract"
          - "spinifex"
      default: "spinifex"
      doc: |
        The name of the target solution table to use from the solset input for
        rotation measure corrections.

    - id: select_best_n_delay_calibrators
      type: int?
      default: 1
      doc: Select this number of top-scoring delay calibrator candidates to attempt to calibrate.

    - id: starting_skymodel
      type:
        - File?
        - File[]?
      doc: |
        Optional starting model(s) in BBS-compatible text format used to kickstart the delay calibration. If given and `do_auto_delay_selection` is enabled, the number of skymodels must be equal to `select_best_n_delay_calibrators`. Additionally, they should be named in such a way that when sorted by name, the delay calibrator MSes and skymodels end up in the same order.

    - id: do_auto_delay_selection
      type: boolean?
      default: false
      doc: |
        Automatically select the best delay calibrator based on phasediff scores.

steps:
    - id: setup
      label: setup
      in:
        - id: msin
          source: msin
        - id: solset
          source: solset
          valueFrom: $(self)
        - id: filter_baselines
          source: filter_baselines
        - id: flag_baselines
          source: flag_baselines
        - id: phasesol
          source: phasesol
        - id: number_cores
          source: number_cores
        - id: Ateam_skymodel
          source: Ateam_skymodel
        - id: rm_correction
          source: rm_correction
      out:
        - id: logdir
        - id: msout
        - id: summary_file
      run: ./setup.cwl
      when: $(inputs.solset != null)

    - id: lofar_vlbi_plot
      in:
        - id: msin
          source: msin
          valueFrom: $(self[0])
        - id: delay_calibrator
          source: delay_calibrator
      out:
        - id: delay_calibrator_pf
        - id: image_catalogue_pf
      run: ../steps/lofar_vlbi_plot.cwl
      when: $(inputs.delay_calibrator == null)

    - id: sort-concatenate-flag
      in:
        - id: msin
          source:
           - setup/msout
          pickValue: the_only_non_null
        - id: firstSB
          source: reference_stationSB
        - id: max_dp3_threads
          source: max_dp3_threads
        - id: rfi_strategy
          source: rfi_strategy
      out:
        - id: logdir
        - id: msout
        - id: summary_file
      run: ./concatenate-flag.cwl
      label: sort-concatenate-flag
      when: $(inputs.msin != null)

    - id: process_ddf
      in:
        - id: msin
          source:
            - sort-concatenate-flag/msout
            - msin
          pickValue: first_non_null
          valueFrom: $(self)
        - id: solsdir
          source: ddf_solsdir
          valueFrom: $(self)
        - id: ddf_rundir
          source: ddf_rundir
          valueFrom: $(self)
        - id: box_size
          source: box_size
        - id: ncpu
          source: number_cores
        - id: chunkhours
          source: subtract_chunk_hours
        - id: do_subtraction
          source: do_subtraction
      out:
        - id: msout
      run: ./process-ddf.cwl
      when: $(inputs.ddf_rundir != null && inputs.solsdir != null)

    - id: phaseup
      in:
        - id: msin
          source:
            - process_ddf/msout
            - sort-concatenate-flag/msout
            - msin
          linkMerge: merge_nested
          pickValue: first_non_null
          valueFrom: $(self)
        - id: delay_calibrator
          source:
            - delay_calibrator
            - lofar_vlbi_plot/delay_calibrator_pf
          pickValue: first_non_null
          valueFrom: $(self)
        - id: image_catalogue
          source: image_catalogue
        - id: max_dp3_threads
          source: max_dp3_threads
        - id: model_image
          source: model_image
        - id: number_cores
          source: number_cores
        - id: do_auto_delay_selection
          source: do_auto_delay_selection
      out:
        - id: msout
        - id: solutions
        - id: pictures
        - id: facetselfcal_config
        - id: starting_skymodel
        - id: logdir
        - id: summary_file
      run: ./phaseup-concat.cwl
      label: phaseup
      when: $(!inputs.do_auto_delay_selection)

    - id: validation
      in:
        - id: h5parm
          source:
            - select_best_delay_cal/solutions
            - phaseup/solutions
          linkMerge: merge_flattened
          pickValue: all_non_null
          valueFrom: $(self)
        - id: do_validation
          source: do_validation
      out:
        - validate_csv
      run: ./subworkflows/dical_validation.cwl
      when: $(inputs.do_validation)

    - id: select_best_delay_cal
      in:
        - id: msin
          source:
            - process_ddf/msout
            - sort-concatenate-flag/msout
            - msin
          linkMerge: merge_nested
          pickValue: first_non_null
          valueFrom: $(self)
        - id: delay_calibrator
          source:
            - delay_calibrator
            - lofar_vlbi_plot/delay_calibrator_pf
          pickValue: first_non_null
          valueFrom: $(self)
        - id: image_catalogue
          source:
            - image_catalogue
            - lofar_vlbi_plot/image_catalogue_pf
          pickValue: first_non_null
          valueFrom: $(self)
        - id: select_best_n_delay_calibrators
          source: select_best_n_delay_calibrators
        - id: do_auto_delay_selection
          source: do_auto_delay_selection
        - id: starting_skymodel
          source: starting_skymodel
          # This valueFrom suppresses CWL's warning about potentially incompatible types.
          # There can technically be a single skymodel, but that means there is one source
          # and thus nothing to automatically select the best from.
          # Practically, we should never be in a situation of having File here.
          valueFrom: $(self)
      out:
        - id: msout
        - id: starting_skymodels
        - id: pictures
        - id: phasediff_score_csv
        - id: solutions
      run: ./subworkflows/find-best-delay-calibrator.cwl
      when: $(inputs.do_auto_delay_selection)

    - id: store_logs
      in:
        - id: files
          linkMerge: merge_flattened
          source:
            - setup/logdir
            - sort-concatenate-flag/logdir
            - phaseup/logdir
          pickValue: all_non_null
        - id: sub_directory_name
          default: logs
      out:
        - id: dir
      run: ../steps/collectfiles.cwl
      label: store_logs

    # Selection of the concatenated MSs, as pickValue doesn't allow
    # us to do this in the msouts output of the workflow. The reasoning
    # is the following:
    # If process_ddf was run, take that output.
    # If process_ddf was not run but sort-concatenate-flag was, take that instead.
    # Otherwise, don't collect anything.
    - id: select_concatenated_mss
      in:
        - id: input1
          source: sort-concatenate-flag/msout
        - id: input2
          source: process_ddf/msout
      out:
        - id: output
      when: $(inputs.input1 != null || inputs.input2 != null)
      run: ../utils/select_input.cwl

    - id: apply_delay_allms
      in:
        - id: ms
          source:
            - select_concatenated_mss/output
            - msin
          pickValue: first_non_null
        - id: h5parm
          source:
            - phaseup/solutions
            - select_best_delay_cal/solutions
          pickValue: first_non_null
          linkMerge: merge_flattened
          # This valueFrom suppresses CWL's warning about potentially incompatible types.
          # select_best_delay_cal can technically return either File or File[], but the when
          # clause for this step prevents from running if we have more than 1 candidate, so
          # we should never be in a situation of having File[] here.
          valueFrom: $(self)
        - id: apply_delay_solutions
          source: apply_delay_solutions
        - id: select_best_n_delay_calibrators
          source: select_best_n_delay_calibrators
      out:
        - id: ms_out
      run: ../steps/applycal.cwl
      scatter: ms
      label: apply_delay_allms
      when: $(inputs.apply_delay_solutions && (inputs.select_best_n_delay_calibrators == 1))

outputs:
  - id: msout
    outputSource: 
      - select_best_delay_cal/msout
      - phaseup/msout
    pickValue: all_non_null
    linkMerge: merge_flattened
    type: Directory[]
    doc: |
        The fully concatenated data in MeasurementSet
        format, phase-shifted to the delay calibrator.

  - id: msouts
    outputSource:
      - apply_delay_allms/ms_out
      - select_concatenated_mss/output
    pickValue: first_non_null
    type: Directory[]?
    doc: |
        The concatenated data in MeasurementSet format after
        A-team clipping and optional DDF solutions applied
        and/or delay solutions applied.

  - id: logs
    outputSource: store_logs/dir
    type: Directory
    doc: |
        The logfiles generated by the pipeline steps,
        sorted per subworkflow.

  - id: pictures
    outputSource: 
      - select_best_delay_cal/pictures
      - phaseup/pictures
    type: File[]
    pickValue: all_non_null
    linkMerge: merge_flattened
    doc: Inspection plots generated by lofar_facet_selfcal.

  - id: facetselfcal_config
    outputSource: 
      - phaseup/facetselfcal_config
    type: File
    pickValue: the_only_non_null
    doc: Config file with settings used for the delay calibration.

  - id: facetselfcal_starting_model
    outputSource: 
      - phaseup/starting_skymodel
      - select_best_delay_cal/starting_skymodels
    type: File[]
    pickValue: first_non_null
    doc: |
        The starting model(s) that were used to kick start the delay calibration.

  - id: solutions
    outputSource:
      - select_best_delay_cal/solutions
      - phaseup/solutions
    pickValue: all_non_null
    type:
      - File?
      - File[]?
    doc: |
        The calibrated data solutions generated by lofar_facet_selfcal
        in HDF5 format.

  - id: validation_csv
    outputSource: validation/validate_csv
    type: File?
    doc: CSV with validation scores

  - id: phasediff_score_csv
    outputSource:
      - select_best_delay_cal/phasediff_score_csv
    type:
      - File?
    pickValue: the_only_non_null
    doc: |
        A CSV file containing the phasediff scores for each of the calibrators that were split out.

  - id: summary_files
    outputSource:
      - setup/summary_file
      - sort-concatenate-flag/summary_file
      - phaseup/summary_file
    pickValue: all_non_null
    type: File[]
    doc: Pipeline summary statistics in JSON format.