Merge multiscale and non-multiscale steps

docs/source/parset.rst

@@ -117,7 +117,7 @@ The available options are described below under their respective sections.
 
     llssolver
         The linear least-squares solver to use (one of "qr", "svd", or "lsmr";
-        default = "qr")
+        default = ``qr``)
 
     maxiter
         Maximum number of iterations to perform during calibration (default = 50).
@@ -127,7 +127,8 @@ The available options are described below under their respective sections.
 
     solveralgorithm
         The algorithm used for solving (one of "directionsolve", "directioniterative",
-        "lbfgs", or "hybrid"; default = "hybrid")? When using "lbfgs", the "stepsize" should be set to a small value like 0.001.
+        "lbfgs", or "hybrid"; default = ``hybrid``)? When using "lbfgs", the :term:`stepsize`
+        should be set to a small value like 0.001.
 
     onebeamperpatch
         Calculate the beam correction once per calibration patch (default =
@@ -140,7 +141,8 @@ The available options are described below under their respective sections.
         Parallelize model calculation over baselines, instead of parallelizing over directions (default = ``False``).
 
     stepsize
-        Size of steps used during calibration (default = 0.02).
+        Size of steps used during calibration (default = 0.02). When using
+        ``solveralgorithm = lbfgs``, the stepsize should be set to a small value like 0.001.
 
     tolerance
         Tolerance used to check convergence during calibration (default = 1e-3).
@@ -214,24 +216,23 @@ The available options are described below under their respective sections.
     taper_arcsec
         Taper to apply when imaging, in arcsec (default = 0).
 
-    multiscale_scales_pixel
-        Scale sizes in pixels to use during multiscale clean (default = ``[0, 5, 10, 15]``).
+    do_multiscale_clean
+        Use multiscale cleaning (default = ``True``)?
 
-    do_multiscale
-        Use multiscale cleaning (default = auto)?
-
-    use_screens
-        Use screens during imaging (default = ``True``)? If ``False``, the
-        solutions closest to the image centers will be used.
+    dde_method
+        Method to use to correct for direction-dependent effects during imaging: "none",
+        "facets", or "screens" (default = ``facets``). If "none", the solutions closest to the image centers
+        will be used. If "facets", Voronoi faceting is used. If "screens", smooth 2-D
+        screens are used.
 
     screen_type
-        Type of screen to use (default = "tessellated"), if use_screens = ``True``:
+        Type of screen to use (default = ``tessellated``), if ``dde_method = screens``:
         "tessellated" (simple, smoothed Voronoi tessellated screens) or
         "kl" (Karhunen-Lo`eve screens).
 
     idg_mode
         IDG (image domain gridder) mode to use in WSClean (default = "hybrid").
-        The mode can be "cpu" or "hybrid"".
+        The mode can be "cpu" or "hybrid".
 
     mem_fraction
         Fraction of the total memory (per node) to use for WSClean jobs (default = 0.9).
@@ -240,7 +241,8 @@ The available options are described below under their respective sections.
         Use MPI to distribute WSClean jobs over multiple nodes (default =
         ``False``)? If ``True`` and more than one node can be allocated to each
         WSClean job (i.e., max_nodes / num_images >= 2), then distributed
-        imaging will be used (only available if batch_system = slurm).
+        imaging will be used (only available if ``batch_system = slurm`` and
+        ``dde_method = screens``).
 
         .. note::
 
@@ -249,7 +251,10 @@ The available options are described below under their respective sections.
             it is on a shared filesystem.
 
     reweight
-        Reweight the visibility data before imaging (default = ``True``).
+        Reweight the visibility data before imaging (default = ``False``). If
+        ``True``, data with high residuals (compared to the predicted model
+        visibilities) are down-weighted. This feature is experimental and
+        should be used with caution.
 
     grid_width_ra_deg
         Size of area to image when using a grid (default = mean FWHM of the
@@ -287,11 +292,6 @@ The available options are described below under their respective sections.
     sector_width_dec_deg_list
         List of image  widths, in degrees (default = ``[]``).
 
-    sector_do_multiscale_list
-        List of multiscale flags, one per sector (default = ``[]``). ``None``
-        indicates that multiscale clean should be activated automatically if a
-        large source is detected in the sector.
-
     max_peak_smearing
         Max desired peak flux density reduction at center of the image edges due
         to bandwidth smearing (at the mean frequency) and time smearing (default
@@ -334,7 +334,12 @@ The available options are described below under their respective sections.
         all).
 
     deconvolution_threads
-        Number of threads to use by WSClean during deconvolution (default = 0 = all).
+        Number of threads to use by WSClean during deconvolution (default = 0 = 2/5
+        of ``max_threads``).
+
+    parallel_gridding_threads
+        Number of threads to use by WSClean during parallel gridding (default = 0 = 2/5
+        of ``max_threads``).
 
     dir_local
         Full path to a local disk on the nodes for IO-intensive processing (default =

examples/rapthor.parset

@@ -82,19 +82,18 @@ download_initial_skymodel_server = TGSS
 # slow_smoothnessconstraint_joint = 3e6
 # slow_smoothnessconstraint_separate = 3e6
 
-# Do extra solve for debugging purposes
-# debug = False
-
 
 [imaging]
 # Imaging parameters: pixel size in arcsec (default = 1.25, suitable for HBA data), Briggs
-# robust parameter (default = -0.5) and minimum uv distance in lambda (default = 80)
+# robust parameter (default = -0.5), min and max uv distance in lambda (default = 80, none),
+# taper in arcsec (default = none), and whether multiscale clean should be used (default =
+# True)
 # cellsize_arcsec = 1.25
 # robust = -0.5
 # min_uv_lambda = 80.0
 # max_uv_lambda = 0.0
 # taper_arcsec = 0.0
-# multiscale_scales_pixel = [0, 5, 10, 15]
+# do_multiscale_clean = True
 
 # Method to use to correct for direction-dependent effects during imaging: "none",
 # "facets", or "screens". If "none", the solutions closest to the image centers
@@ -135,27 +134,24 @@ download_initial_skymodel_server = TGSS
 # grid_nsectors_ra = 3
 
 # Instead of a grid, imaging sectors can be defined individually by specifying
-# their centers and widths. Multiscale clean can also be set (with None indicating that
-# multiscale clean should be activated automatically if a large source is detected in the
-# sector). If sectors are specified in this way, they will be
+# their centers and widths. If sectors are specified in this way, they will be
 # used instead of the sector grid. Note that the sectors should not overlap
 # sector_center_ra_list = [14h41m01.884, 14h13m23.234]
 # sector_center_dec_list = [+35d30m31.52, +37d21m56.86]
 # sector_width_ra_deg_list = [0.532, 0.127]
 # sector_width_dec_deg_list = [0.532, 0.127]
-# sector_do_multiscale_list = [None, True]
 
 # Max desired peak flux density reduction at center of the image edges due to
 # bandwidth smearing (at the mean frequency) and time smearing (default = 0.15 =
-# 15% reduction in peak flux). Higher values result in shorter run times but
+# 15% reduction in peak flux). Higher values can result in shorter run times but
 # more smearing away from the sector centers
 # max_peak_smearing = 0.15
 
 
 [cluster]
-# Cluster batch system (default = singleMachine). Use batch_system = slurm to
+# Cluster batch system (default = single_machine). Use batch_system = slurm to
 # use a SLURM-based cluster
-# batch_system = singleMachine
+# batch_system = single_machine
 
 # For batch_system = slurm, the maximum number of nodes of the cluster to use at
 # once can be specified with the max_nodes option (default = 12), the number of
@@ -168,16 +164,15 @@ download_initial_skymodel_server = TGSS
 # cpus_per_task = 0
 # mem_per_node_gb = 0
 
-# TODO: Slurm partition names?
-# cpu_partition = batch
-# gpu_partition = batch
-
 # Maximum number of cores and threads per task to use on each node (default = 0 = all)
 # max_cores = 0
 # max_threads = 0
 
-# Number of threads to use by WSClean during deconvolution (default = 0 = all)
+# Number of threads to use by WSClean during deconvolution and parallel gridding
+# (default = 0 = 2/5 of max_threads). Higher values will speed up imaging at the
+# expense of higher memory usage
 # deconvolution_threads = 0
+# parallel_gridding_threads = 0
 
 # Full path to a local disk on the nodes for IO-intensive processing (default =
 # not used). The path must exist on all nodes (but does not have to be on a

rapthor/lib/parset.py

@@ -524,40 +524,12 @@ def get_imaging_options(parset):
         len_list.append(len(val_list))
     else:
         parset_dict['sector_width_dec_deg_list'] = []
-    if 'sector_do_multiscale_list' in parset_dict:
-        val_list = parset_dict['sector_do_multiscale_list'].strip('[]').split(',')
-        if val_list[0] == '':
-            val_list = []
-        bool_list = []
-        for v in val_list:
-            if v.lower().strip() == 'true':
-                b = True
-            elif v.lower().strip() == 'false':
-                b = False
-            elif v.lower().strip() == 'none':
-                b = None
-            else:
-                log.error('The entry "{}" in sector_do_multiscale_list is invalid. It must '
-                          'be one of True, False, or None'.format(v))
-                sys.exit(1)
-            bool_list.append(b)
-        parset_dict['sector_do_multiscale_list'] = bool_list
-        len_list.append(len(bool_list))
-    else:
-        parset_dict['sector_do_multiscale_list'] = []
-    if (len(parset_dict['sector_do_multiscale_list']) == 0 or
-            (True not in parset_dict['sector_do_multiscale_list'] and
-             None not in parset_dict['sector_do_multiscale_list'])):
-        parset_dict['do_multiscale_clean'] = False
-    else:
-        parset_dict['do_multiscale_clean'] = True
 
     # Check that all the above options have the same number of entries
     if len(set(len_list)) > 1:
         log.error('The options sector_center_ra_list, sector_center_dec_list, '
-                  'sector_width_ra_deg_list, sector_width_dec_deg_list, and '
-                  'sector_do_multiscale_list (if specified) must all have the same number of '
-                  'entires')
+                  'sector_width_ra_deg_list, and sector_width_dec_deg_list '
+                  'must all have the same number of entries')
         sys.exit(1)
 
     # IDG (image domain gridder) mode to use in WSClean (default = cpu). The mode can
@@ -613,24 +585,10 @@ def get_imaging_options(parset):
     else:
         parset_dict['max_peak_smearing'] = 0.15
 
-    # List of scales in pixels to use when multiscale clean is activated (default =
-    # auto). Note that multiscale clean is activated for a direction only when the
-    # calibrator or a source in the facet is determined to be larger than 4 arcmin,
-    # the facet contains the target (specified below with target_ra and target_dec),
-    # or mscale_selfcal_do / mscale_facet_do is set for the direction in the
-    # directions file
-    if 'multiscale_scales_pixel' in parset_dict:
-        val_list = parset_dict['multiscale_scales_pixel'].strip('[]').split(',')
-        str_list = ','.join([v.strip() for v in val_list])
-        parset_dict['multiscale_scales_pixel'] = str_list
-    else:
-        parset_dict['multiscale_scales_pixel'] = None
-
-    # Selfcal imaging parameters: pixel size in arcsec (default = 1.25), Briggs
-    # robust parameter (default = -0.5) and minimum uv distance in lambda
-    # (default = 80). These settings apply both to selfcal images and to the
-    # full facet image used to make the improved facet model that is subtracted
-    # from the data
+    # Imaging parameters: pixel size in arcsec (default = 1.25, suitable for HBA data), Briggs
+    # robust parameter (default = -0.5), min and max uv distance in lambda (default = 80, none),
+    # taper in arcsec (default = none), and whether multiscale clean should be used (default =
+    # True)
     if 'cellsize_arcsec' in parset_dict:
         parset_dict['cellsize_arcsec'] = parset.getfloat('imaging', 'cellsize_arcsec')
     else:
@@ -651,15 +609,19 @@ def get_imaging_options(parset):
         parset_dict['taper_arcsec'] = parset.getfloat('imaging', 'taper_arcsec')
     else:
         parset_dict['taper_arcsec'] = 0.0
+    if 'do_multiscale_clean' in parset_dict:
+        parset_dict['do_multiscale_clean'] = parset.getboolean('imaging', 'do_multiscale_clean')
+    else:
+        parset_dict['do_multiscale_clean'] = True
 
     # Check for invalid options
     allowed_options = ['max_peak_smearing', 'cellsize_arcsec', 'robust', 'reweight',
-                       'multiscale_scales_pixel', 'grid_center_ra', 'grid_center_dec',
+                       'grid_center_ra', 'grid_center_dec',
                        'grid_width_ra_deg', 'grid_width_dec_deg', 'grid_nsectors_ra',
                        'min_uv_lambda', 'max_uv_lambda', 'mem_fraction', 'screen_type',
                        'robust', 'sector_center_ra_list', 'sector_center_dec_list',
                        'sector_width_ra_deg_list', 'sector_width_dec_deg_list',
-                       'idg_mode', 'sector_do_multiscale_list', 'use_mpi',
+                       'idg_mode', 'do_multiscale_clean', 'use_mpi',
                        'dde_method', 'skip_corner_sectors']
     for option in given_options:
         if option not in allowed_options:
@@ -738,13 +700,21 @@ def get_cluster_options(parset):
     if parset_dict['max_threads'] == 0:
         parset_dict['max_threads'] = multiprocessing.cpu_count()
 
-    # Number of threads to use by WSClean during deconvolution (default = 0 = all)
+    # Number of threads to use by WSClean during deconvolution and parallel gridding
+    # (default = 0 = 2/5 of max_threads). Higher values will speed up imaging at the
+    # expense of higher memory usage
     if 'deconvolution_threads' in parset_dict:
         parset_dict['deconvolution_threads'] = parset.getint('cluster', 'deconvolution_threads')
     else:
         parset_dict['deconvolution_threads'] = 0
     if parset_dict['deconvolution_threads'] == 0:
-        parset_dict['deconvolution_threads'] = multiprocessing.cpu_count()
+        parset_dict['deconvolution_threads'] = max(1, int(parset_dict['max_threads'] * 2 / 5))
+    if 'parallel_gridding_threads' in parset_dict:
+        parset_dict['parallel_gridding_threads'] = parset.getint('cluster', 'parallel_gridding_threads')
+    else:
+        parset_dict['parallel_gridding_threads'] = 0
+    if parset_dict['parallel_gridding_threads'] == 0:
+        parset_dict['parallel_gridding_threads'] = max(1, int(parset_dict['max_threads'] * 2 / 5))
 
     # Full path to a local disk on the nodes for I/O-intensive processing. The path
     # must be the same for all nodes
@@ -768,7 +738,7 @@ def get_cluster_options(parset):
     if 'cwl_runner' not in parset_dict:
         parset_dict['cwl_runner'] = 'toil'
     cwl_runner = parset_dict['cwl_runner']
-    supported_cwl_runners = ('cwltool', 'toil') #, 'toil-cwl-runner')
+    supported_cwl_runners = ('cwltool', 'toil')
     if cwl_runner not in supported_cwl_runners:
         log.critical("CWL runner '%s' is not supported; select one of: %s",
                      cwl_runner, ', '.join(supported_cwl_runners))

rapthor/lib/sector.py

@@ -82,15 +82,14 @@ class Sector(object):
             obs.set_prediction_parameters(self.name, self.patches,
                                           os.path.join(self.field.working_dir, 'scratch'))
 
-    def set_imaging_parameters(self, do_multiscale=None, recalculate_imsize=False):
+    def set_imaging_parameters(self, do_multiscale=False, recalculate_imsize=False):
         """
         Sets the parameters needed for the imaging pipeline
 
         Parameters
         ----------
         do_multiscale : bool, optional
-            If True, multiscale clean is done. If None, multiscale clean is done only
-            when a large source is detected
+            If True, multiscale clean is done
         recalculate_imsize : bool, optional
             If True, the image size is recalculated based on the current sector region
         """
@@ -179,21 +178,8 @@ class Sector(object):
             self.wsclean_niter = int(1e7)  # set to high value and just use nmiter to limit clean
             self.wsclean_nmiter = max(2, int(round(self.wsclean_nmiter * 0.75)))
 
-        # Set multiscale: get source sizes and check for large sources
+        # Set multiscale clean
         self.multiscale = do_multiscale
-        if self.multiscale is None:
-            # TODO: figure out good way to determine whether multiscale should be used
-            # and the scales, maybe using the presence of Gaussians on larger wavelet
-            # scales? For now, force it to off.
-            self.multiscale = False
-
-#             largest_scale = np.max(self.source_sizes) / self.cellsize_deg / 3.0
-#             large_size_arcmin = 4.0  # threshold source size for multiscale to be activated
-#             sizes_arcmin = self.source_sizes * 60.0
-#             if sizes_arcmin is not None and any([s > large_size_arcmin for s in sizes_arcmin]):
-#                 self.multiscale = True
-#             else:
-#                 self.multiscale = False
         if self.multiscale:
             self.wsclean_niter = int(self.wsclean_niter/1.5)  # fewer iterations are needed
             self.log.debug("Will do multiscale cleaning.")

rapthor/operations/calibrate.py

@@ -34,7 +34,6 @@ class Calibrate(Operation):
                              'do_slowgain_solve': self.field.do_slowgain_solve,
                              'use_scalarphase': self.field.use_scalarphase,
                              'max_cores': max_cores,
-                             'max_threads': self.field.parset['cluster_specific']['max_threads'],
                              'debug': self.field.debug}
 
     def set_input_parameters(self):
@@ -196,7 +195,8 @@ class Calibrate(Operation):
                             'combined_h5parms2': combined_h5parms2,
                             'solverlbfgs_dof': solverlbfgs_dof,
                             'solverlbfgs_iter': solverlbfgs_iter,
-                            'solverlbfgs_minibatches': solverlbfgs_minibatches}
+                            'solverlbfgs_minibatches': solverlbfgs_minibatches,
+                            'max_threads': self.field.parset['cluster_specific']['max_threads']}
 
         if self.field.debug:
             output_slow_h5parm_debug = ['slow_gain_{}_debug.h5parm'.format(i)

rapthor/operations/image.py

@@ -43,9 +43,6 @@ class Image(Operation):
                              'use_facets': use_facets,
                              'peel_bright_sources': self.field.peel_bright_sources,
                              'max_cores': max_cores,
-                             'max_threads': self.field.parset['cluster_specific']['max_threads'],
-                             'deconvolution_threads': self.field.parset['cluster_specific']['deconvolution_threads'],
-                             'do_multiscale_clean': self.field.do_multiscale_clean,
                              'use_mpi': self.field.use_mpi,
                              'toil_version': self.toil_major_version}
 
@@ -75,15 +72,8 @@ class Image(Operation):
             # Generally, this should work fine, since we do not expect large changes in
             # the size of the sector from iteration to iteration (small changes are OK,
             # given the padding we use during imaging)
-            if self.field.do_multiscale_clean:
-                sector_do_multiscale_list = self.field.parset['imaging_specific']['sector_do_multiscale_list']
-                if len(sector_do_multiscale_list) == nsectors:
-                    do_multiscale = sector_do_multiscale_list[i]
-                else:
-                    do_multiscale = None
-            else:
-                do_multiscale = False
-            sector.set_imaging_parameters(do_multiscale=do_multiscale, recalculate_imsize=False)
+            sector.set_imaging_parameters(do_multiscale=self.field.do_multiscale_clean,
+                                          recalculate_imsize=False)
 
             # Set input MS filenames
             if self.field.do_predict:
@@ -149,7 +139,10 @@ class Image(Operation):
                             'idg_mode': [sector.idg_mode for sector in self.field.imaging_sectors],
                             'wsclean_mem': [sector.mem_percent for sector in self.field.imaging_sectors],
                             'threshisl': [sector.threshisl for sector in self.field.imaging_sectors],
-                            'threshpix': [sector.threshpix for sector in self.field.imaging_sectors]}
+                            'threshpix': [sector.threshpix for sector in self.field.imaging_sectors],
+                            'do_multiscale': [sector.multiscale for sector in self.field.imaging_sectors],
+                            'max_threads': self.field.parset['cluster_specific']['max_threads'],
+                            'deconvolution_threads': self.field.parset['cluster_specific']['deconvolution_threads']}
 
         if self.field.peel_bright_sources:
             self.input_parms.update({'bright_skymodel_pb': CWLFile(self.field.bright_source_skymodel_file).to_json()})
@@ -197,6 +190,8 @@ class Image(Operation):
                     self.input_parms.update({'soltabs': 'amplitude000,phase000'})
                 else:
                     self.input_parms.update({'soltabs': 'phase000'})
+                self.input_parms.update({'parallel_gridding_threads':
+                                         self.field.parset['cluster_specific']['parallel_gridding_threads']})
             else:
                 self.input_parms.update({'central_patch_name': central_patch_name})
 

rapthor/operations/mosaic.py

@@ -32,7 +32,6 @@ class Mosaic(Operation):
             max_cores = self.field.parset['cluster_specific']['max_cores']
         self.parset_parms = {'rapthor_pipeline_dir': self.rapthor_pipeline_dir,
                              'max_cores': max_cores,
-                             'max_threads': self.field.parset['cluster_specific']['max_threads'],
                              'skip_processing': self.skip_processing,
                              'do_slowgain_solve': self.field.do_slowgain_solve}
 

rapthor/operations/predict.py

@@ -28,7 +28,6 @@ class Predict(Operation):
             max_cores = self.field.parset['cluster_specific']['max_cores']
         self.parset_parms = {'rapthor_pipeline_dir': self.rapthor_pipeline_dir,
                              'max_cores': max_cores,
-                             'max_threads': self.field.parset['cluster_specific']['max_threads'],
                              'do_slowgain_solve': self.field.do_slowgain_solve}
 
     def set_input_parameters(self):
@@ -110,7 +109,8 @@ class Predict(Operation):
                             'peel_outliers': peel_outliers,
                             'nr_bright': nr_bright,
                             'peel_bright': peel_bright,
-                            'reweight': reweight}
+                            'reweight': reweight,
+                            'max_threads': self.field.parset['cluster_specific']['max_threads']}
 
     def finalize(self):
         """

rapthor/pipeline/parsets/calibrate_pipeline.cwl

@@ -220,6 +220,12 @@ inputs:
       The screen type to use to derive the a-term images (length = 1).
     type: string
 
+  - id: max_threads
+    label: Max number of threads
+    doc: |
+      The maximum number of threads to use for a job (length = 1).
+    type: int
+
 {% if do_slowgain_solve %}
 # start do_slowgain_solve
   - id: freqchunk_filename
@@ -395,7 +401,7 @@ steps:
   - id: solve_fast_phases
     label: Solve for fast phases
     doc: |
-      This step uses DDECal (in DPPP) to solve for phase corrections on short
+      This step uses DDECal (in DP3) to solve for phase corrections on short
       timescales (< 1 minute), using the input MS files and sourcedb. These
       corrections are used to correct primarily for ionospheric effects.
 {% if use_scalarphase %}
@@ -457,7 +463,7 @@ steps:
       - id: antennaconstraint
         source: fast_antennaconstraint
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [msin, starttime, ntimes, h5parm, solint, nchan, smoothnessreffrequency]
     scatterMethod: dotproduct
     out:
@@ -498,7 +504,7 @@ steps:
   - id: solve_slow_gains1
     label: Solve for slow gains 1
     doc: |
-      This step uses DDECal (in DPPP) to solve for gain corrections on long
+      This step uses DDECal (in DP3) to solve for gain corrections on long
       timescales (> 10 minute), using the input MS files and sourcedb. These
       corrections are used to correct primarily for beam errors. The fast-
       phase solutions are preapplied and all stations are constrained to
@@ -560,7 +566,7 @@ steps:
       - id: antennaconstraint
         source: slow_antennaconstraint
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [msin, starttime, ntimes, startchan, nchan, h5parm, solint, solve_nchan]
     scatterMethod: dotproduct
     out:
@@ -621,7 +627,7 @@ steps:
   - id: solve_slow_gains2
     label: Solve for slow gains 2
     doc: |
-      This step uses DDECal (in DPPP) to solve for gain corrections on long
+      This step uses DDECal (in DP3) to solve for gain corrections on long
       timescales (> 10 minute), using the input MS files and sourcedb. These
       corrections are used to correct primarily for beam errors. The fast-
       phase solutions and first slow-gain solutions are preapplied and stations
@@ -682,7 +688,7 @@ steps:
       - id: smoothnessconstraint
         source: slow_smoothnessconstraint2
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [msin, starttime, ntimes, startchan, nchan, h5parm, solint, solve_nchan]
     scatterMethod: dotproduct
     out:
@@ -865,7 +871,7 @@ steps:
       - id: sector_bounds_mid_deg
         source: sector_bounds_mid_deg
       - id: ncpu
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [h5parm, outroot]
     scatterMethod: dotproduct
     out:
@@ -935,7 +941,7 @@ steps:
       - id: smoothnessconstraint
         source: slow_smoothnessconstraint
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [msin, starttime, ntimes, startchan, nchan, h5parm, solint, solve_nchan]
     scatterMethod: dotproduct
     out:
@@ -1011,7 +1017,7 @@ steps:
       - id: sector_bounds_mid_deg
         source: sector_bounds_mid_deg
       - id: ncpu
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [h5parm, outroot]
     scatterMethod: dotproduct
     out:

rapthor/pipeline/parsets/image_pipeline.cwl

@@ -3,7 +3,7 @@ class: Workflow
 label: Rapthor imaging pipeline
 doc: |
   This workflow performs imaging with direction-dependent corrections. The
-  imaging data are generated (and averaged if possible) and WSClean+IDG is
+  imaging data are generated (and averaged if possible) and WSClean is
   used to perform the imaging. Masking and sky model filtering is then done
   using PyBDSF.
 
@@ -229,6 +229,12 @@ inputs:
       The name of the calibration solution table (length = 1).
     type: string
 
+  - id: parallel_gridding_threads
+    label: Max number of gridding threads
+    doc: |
+      The maximum number of threads to use during parallel gridding (length = 1).
+    type: int
+
 {% else %}
 # start not use_facets
 
@@ -274,17 +280,23 @@ inputs:
     type: float[]
 
   - id: min_uv_lambda
-    label: Minimum us distance
+    label: Minimum uv distance
     doc: |
       The WSClean minimum uv distance in lambda (length = n_sectors).
     type: float[]
 
   - id: max_uv_lambda
-    label: Maximum us distance
+    label: Maximum uv distance
     doc: |
       The WSClean maximum uv distance in lambda (length = n_sectors).
     type: float[]
 
+  - id: do_multiscale
+    label: Activate multiscale
+    doc: |
+      Activate multiscale clean (length = n_sectors).
+    type: boolean[]
+
   - id: taper_arcsec
     label: Taper value
     doc: |
@@ -329,6 +341,19 @@ inputs:
     type: File
 {% endif %}
 
+  - id: max_threads
+    label: Max number of threads
+    doc: |
+      The maximum number of threads to use for a job (length = 1).
+    type: int
+
+  - id: deconvolution_threads
+    label: Max number of deconvolution threads
+    doc: |
+      The maximum number of threads to use during deconvolution (length = 1).
+    type: int
+
+
 outputs:
   - id: filtered_skymodels
     outputSource:
@@ -452,6 +477,8 @@ steps:
         source: min_uv_lambda
       - id: max_uv_lambda
         source: max_uv_lambda
+      - id: do_multiscale
+        source: do_multiscale
       - id: taper_arcsec
         source: taper_arcsec
       - id: wsclean_mem
@@ -464,6 +491,14 @@ steps:
         source: threshisl
       - id: threshpix
         source: threshpix
+      - id: max_threads
+        source: max_threads
+      - id: deconvolution_threads
+        source: deconvolution_threads
+{% if use_facets %}
+      - id: parallel_gridding_threads
+        source: parallel_gridding_threads
+{% endif %}
 {% if peel_bright_sources %}
       - id: bright_skymodel_pb
         source: bright_skymodel_pb
@@ -479,7 +514,7 @@ steps:
 {% endif %}
               channels_out, deconvolution_channels, wsclean_niter,
               wsclean_nmiter, robust, min_uv_lambda,
-              max_uv_lambda, taper_arcsec, wsclean_mem,
+              max_uv_lambda, do_multiscale, taper_arcsec, wsclean_mem,
               auto_mask, idg_mode, threshisl, threshpix]
 {% else %}
 # start not use_screens
@@ -497,7 +532,7 @@ steps:
 {% endif %}
               channels_out, deconvolution_channels, wsclean_niter,
               wsclean_nmiter, robust, min_uv_lambda,
-              max_uv_lambda, taper_arcsec, wsclean_mem,
+              max_uv_lambda, do_multiscale, taper_arcsec, wsclean_mem,
               auto_mask, idg_mode, threshisl, threshpix]
 {% endif %}
 # end use_screens / not use_screens

rapthor/pipeline/parsets/image_sector_pipeline.cwl

@@ -3,7 +3,7 @@ class: Workflow
 label: Rapthor imaging subpipeline
 doc: |
   This subworkflow performs imaging with direction-dependent corrections. The
-  imaging data are generated (and averaged if possible) and WSClean+IDG is
+  imaging data are generated (and averaged if possible) and WSClean is
   used to perform the imaging. Masking and sky model filtering is then done
   using PyBDSF.
 
@@ -194,6 +194,12 @@ inputs:
       The names of the calibration solution tables (length = 1).
     type: string
 
+  - id: parallel_gridding_threads
+    label: Max number of gridding threads
+    doc: |
+      The maximum number of threads to use during parallel gridding (length = 1).
+    type: int
+
 {% else %}
 # start not use_facets
 
@@ -239,17 +245,23 @@ inputs:
     type: float
 
   - id: min_uv_lambda
-    label: Minimum us distance
+    label: Minimum uv distance
     doc: |
       The WSClean minimum uv distance in lambda (length = 1).
     type: float
 
   - id: max_uv_lambda
-    label: Maximum us distance
+    label: Maximum uv distance
     doc: |
       The WSClean maximum uv distance in lambda (length = 1).
     type: float
 
+  - id: do_multiscale
+    label: Activate multiscale
+    doc: |
+      Activate multiscale clean (length = 1).
+    type: boolean
+
   - id: taper_arcsec
     label: Taper value
     doc: |
@@ -294,6 +306,18 @@ inputs:
     type: File
 {% endif %}
 
+  - id: max_threads
+    label: Max number of threads
+    doc: |
+      The maximum number of threads to use for a job (length = 1).
+    type: int
+
+  - id: deconvolution_threads
+    label: Max number of deconvolution threads
+    doc: |
+      The maximum number of threads to use during deconvolution (length = 1).
+    type: int
+
 outputs:
   - id: filtered_skymodels
     outputSource:
@@ -326,7 +350,7 @@ steps:
   - id: prepare_imaging_data
     label: Prepare imaging data
     doc: |
-      This step uses DPPP to prepare the input data for imaging. This involves
+      This step uses DP3 to prepare the input data for imaging. This involves
       averaging, phase shifting, and optionally the application of the
       calibration solutions at the center.
 {% if use_screens or use_facets %}
@@ -368,7 +392,7 @@ steps:
       - id: beamdir
         source: phasecenter
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
 {% if use_screens or use_facets %}
     scatter: [msin, msout, starttime, ntimes, freqstep, timestep]
 {% else %}
@@ -434,60 +458,25 @@ steps:
     label: Make an image
     doc: |
       This step makes an image using WSClean. Direction-dependent effects
-      can be corrected for using a-term images.
+      can be corrected for using a-term images or facet-based corrections.
 {% if use_screens %}
 # start use_screens
-{% if use_mpi %}
-# start use_mpi
-{% if do_multiscale_clean %}
-# start do_multiscale_clean
-
-{% if toil_version < 5 %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_mpi_image_multiscale_toil4.cwl
-{% else %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_mpi_image_multiscale.cwl
-{% endif %}
-{% else %}
-# start not do_multiscale_clean
 
-{% if toil_version < 5 %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_mpi_image_toil4.cwl
-{% else %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_mpi_image.cwl
-{% endif %}
-{% endif %}
-# end do_multiscale_clean / not do_multiscale_clean
-
-{% else %}
-# start not use_mpi
-
-{% if do_multiscale_clean %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_multiscale.cwl
+{% if use_mpi %}
+    run: {{ rapthor_pipeline_dir }}/steps/wsclean_mpi_image_screens.cwl
 {% else %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image.cwl
-{% endif %}
+    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_screens.cwl
 {% endif %}
-# end use_mpi / not use_mpi
 
 {% else %}
 # start not use_screens
 
 {% if use_facets %}
-# start use_facets
-{% if do_multiscale_clean %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_facets_multiscale.cwl
-{% else %}
     run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_facets.cwl
-{% endif %}
-{% else %}
-# start not use_facets
-{% if do_multiscale_clean %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_no_screens_multiscale.cwl
 {% else %}
-    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_no_screens.cwl
-{% endif %}
+    run: {{ rapthor_pipeline_dir }}/steps/wsclean_image_no_dde.cwl
 {% endif %}
-# end use_facets / not use_facets
+
 {% endif %}
 # end use_screens / not use_screens
 
@@ -521,6 +510,8 @@ steps:
         source: soltabs
       - id: region_file
         source: make_region_file/region_file
+      - id: num_gridding_threads
+        source: parallel_gridding_threads
 {% endif %}
       - id: wsclean_imsize
         source: wsclean_imsize
@@ -534,6 +525,8 @@ steps:
         source: min_uv_lambda
       - id: max_uv_lambda
         source: max_uv_lambda
+      - id: multiscale
+        source: do_multiscale
       - id: cellsize_deg
         source: cellsize_deg
       - id: channels_out
@@ -549,9 +542,9 @@ steps:
       - id: idg_mode
         source: idg_mode
       - id: num_threads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
       - id: num_deconvolution_threads
-        valueFrom: '{{ deconvolution_threads }}'
+        source: deconvolution_threads
     out:
       - id: image_nonpb_name
       - id: image_pb_name
@@ -581,7 +574,7 @@ steps:
         source: image/image_pb_name
         valueFrom: $(self.basename)
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     out:
       - id: restored_image
 
@@ -606,7 +599,7 @@ steps:
         source: image/image_nonpb_name
         valueFrom: $(self.basename)
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     out:
       - id: restored_image
 {% endif %}
@@ -616,7 +609,8 @@ steps:
     label: Filter sources
     doc: |
       This step uses PyBDSF to filter artifacts from the sky model and make
-      a clean mask for the next iteration.
+      a clean mask for the next iteration, as well as derive various image
+      diagnostics.
     run: {{ rapthor_pipeline_dir }}/steps/filter_skymodel.cwl
     in:
 {% if peel_bright_sources %}

rapthor/pipeline/parsets/predict_pipeline.cwl

@@ -152,6 +152,13 @@ inputs:
       The flag that sets reweighting of uv data (length = 1).
     type: boolean
 
+  - id: max_threads
+    label: Max number of threads
+    doc: |
+      The maximum number of threads to use for a job (length = 1).
+    type: int
+
+
 outputs:
   - id: subtract_models
     outputSource:
@@ -166,7 +173,7 @@ steps:
   - id: predict_model_data
     label: Predict the model uv data
     doc: |
-      This step uses DPPP to predict uv data (using the input sourcedb) from the
+      This step uses DP3 to predict uv data (using the input sky model) from the
       input MS files. It also corrupts the model data with the calibration
       solutions. For each sector, prediction is done for all observations.
 {% if do_slowgain_solve %}
@@ -200,7 +207,7 @@ steps:
       - id: directions
         source: sector_patches
       - id: numthreads
-        valueFrom: '{{ max_threads }}'
+        source: max_threads
     scatter: [msin, msout, starttime, ntimes, sourcedb, directions]
     scatterMethod: dotproduct
     out:

rapthor/pipeline/steps/ddecal_solve_complexgain.cwl

@@ -153,7 +153,7 @@ inputs:
       prefix: solve.smoothnessconstraint=
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/ddecal_solve_complexgain1.cwl

@@ -158,7 +158,7 @@ inputs:
       prefix: solve.antennaconstraint=
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/ddecal_solve_complexgain2.cwl

@@ -154,7 +154,7 @@ inputs:
       prefix: solve.smoothnessconstraint=
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/ddecal_solve_complexgain_debug.cwl

@@ -140,7 +140,7 @@ inputs:
       prefix: solve.smoothnessconstraint=
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/ddecal_solve_scalarcomplexgain.cwl

@@ -146,7 +146,7 @@ inputs:
       prefix: solve.antennaconstraint=
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/ddecal_solve_scalarphase.cwl

@@ -232,7 +232,7 @@ inputs:
     label: Number of threads
     doc: |
       The number of threads to use during solve (0 = all).
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/predict_model_data.cwl

@@ -115,7 +115,7 @@ inputs:
     label: Number of threads
     doc: |
       The number of threads for DPPP.
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False

rapthor/pipeline/steps/predict_model_data_phase_only.cwl

@@ -77,7 +77,7 @@ inputs:
       itemSeparator: ','
       separate: False
   - id: numthreads
-    type: string
+    type: int
     inputBinding:
       prefix: numthreads=
       separate: False