flint.imager.wsclean

Simple interface into wsclean

Some notes around the file naming.

A certain filenaming scheme is required for FITS files to perform leakage correction when co-adding them together in the yandasoft linmos application. The stokes field needs to be encoded as .i.. For example:

>>> `SB1234.RACS_0123+43.beam01.i.image.fits`

The wsclean formatted output string appends something denoted with -. Within this code there is and attempt to rename the wsclean outputs to replace the - with a ..

Classes

ResolvedCLIResult	Mapping results to provide to wsclean
WSCleanOptions	A basic container to handle WSClean options. These attributes should
WSCleanOutput	An image wsclean writes, with the fields of its name
WSCleanResult	Simple container for a wsclean command.

Functions

_resolve_wsclean_key_value_to_cli_str(→ ResolvedCLIResult)	An internal function intended to map a key-value pair to
_wsclean_output_callback(→ None)	Call back function used to detect clean divergence
clear_wsclean_outputs(→ list[pathlib.Path])	Delete the images this run of wsclean writes that already exist under prefix
cli(→ None)
collect_wsclean_outputs(→ flint.imageset.ImageSet)	Gather the images wsclean wrote under prefix into an ImageSet with flint names.
create_wsclean_cmd(→ WSCleanResult)	Create a wsclean command from a WSCleanOptions container
create_wsclean_name_argument(→ pathlib.Path)	Create the value that will be provided to wsclean -name argument. This has
expected_wsclean_outputs() → list[WSCleanOutput])	The images wsclean writes for image_types, each channel then the MFS image
find_wsclean_outputs(→ list[WSCleanOutput])	The images this run of wsclean writes that exist under prefix, found by exact name
flint_output_path(→ pathlib.Path)	The flint name of a wsclean image. Its polarisation, if tagged, replaces
get_parser(→ argparse.ArgumentParser)
get_wsclean_output_source_list_path(→ pathlib.Path)	WSClean can produce a text file that describes the components
image_set_from_result(→ flint.imageset.ImageSet | None)
merge_image_sets_from_results(→ flint.imageset.ImageSet)	Merge the image sets from multiple wsclean results into a single image set.
parse_pols(→ tuple[str, ...] | None)	The polarisations wsclean tags each image with, None when only one is imaged.
remove_wsclean_outputs() → list[pathlib.Path])	Delete the channel images of image_types, keeping the MFS images
run_wsclean_imager(→ flint.imageset.ImageSet)	Run a provided wsclean command. Optionally will clean up files,
verify_wsclean_outputs(→ None)	Raise FileNotFoundError if an image this run should write is not among
wsclean_imager(→ WSCleanResult)	Create and run a wsclean imager command against a measurement set.
wsclean_output_path(→ pathlib.Path)	The name wsclean gives an image, <prefix>[-<channel>|-MFS][-<pol>]-<type>.fits

Module Contents

class flint.imager.wsclean.ResolvedCLIResult

Bases: NamedTuple

Mapping results to provide to wsclean

bindpath: pathlib.Path | None = None

A path to bind to when called within a container

cmd: str | None = None

The argument value pair to place on the CLI.

ignore: bool = False

Ignore this CLIResult if True

unknown: Any | None = None

Unknown options that could not be converted

class flint.imager.wsclean.WSCleanOptions(/, **data: Any)

Bases: flint.options.BaseOptions

A basic container to handle WSClean options. These attributes should conform to the same option name in the calling signature of wsclean

Basic support for environment variables is available. Should a value start with $ it is assumed to be a environment variable, it is will be looked up. Some basic attempts to determine if it is a path is made.

Should flint_hold_dir (or, failing that, temp_dir) be specified then all images will be created in this location, and then moved over to the same parent directory as the imaged MS. This is done by setting the wsclean -name argument.

Options that start with flint_ are ignored when generating the wsclean command and generally denote some other flint specific operation.

check_pol() → WSCleanOptions

Refuse a pol flint can not image, and a source list asked of pol: xx,yy

abs_mem: int = 100

Memory wsclean should try to limit itself to

auto_mask: float | None = 3.5

How deep the construct clean mask is during each cycle

auto_threshold: float | None = 0.5

How deep to clean once initial clean threshold reached

beam_fitting_size: float | None = 1.25

Use a fitting box the size of <factor> times the theoretical beam size for fitting a Gaussian to the PSF.

channel_division_frequencies: tuple[float, ...] | None = None

Frequencies, in Hz, that divide the output channels into groups. Supplying channels_out - 1 values fully specifies the output frequency grid, see flint.imager.channel_division

channel_range: tuple[int, int] | None = None

Image a channel range between a lower (inclusive) and upper (exclusive) bound

channels_out: int = 4

Number of output channels

data_column: str = 'CORRECTED_DATA'

Which column in the MS to image

deconvolution_channels: int | None = None

The channels out will be averaged down to this many sub-band images during deconvolution

deconvolution_threads: int | None = None

Number of threads to use during the deconvolution. More may make things faster but can come with a memory hit. If None defaults to -j

fit_spectral_pol: int | None = None

Number of spectral terms to include during sub-band subtraction

fits_mask: pathlib.Path | None = None

Path to a FITS file that encodes a cleaning mask

flint_hold_dir: str | pathlib.Path | None = None

Directory images are written to before being moved beside the MS. Takes precedence over temp_dir for this purpose

flint_keep_linear_images: bool = False

Keep the XX and YY images when pol is xx,yy, rather than deleting them once Stokes I is formed

flint_name_suffix: str | None

An additional trailing token appended to the constructed wsclean -name, e.g. to disambiguate outputs from different pipelines imaging the same measurement set. Letters and digits only, as that is all the name parser reads back from the project field

flint_no_log_wsclean_output: bool = False

If True do not log the wsclean output

flint_save_mfs_products: bool = False

Save the MFS image, model and residual products (co-added and leakage-corrected the same way as the science image), rather than just the image

force_mask_rounds: int | None = None

Round of force masked derivation

gain: float = 0.1

Cleaning gain, ratio of peak that will be subtracted in each iteration

gridder: str | None = 'wgridder'

Use the wgridder kernel in wsclean (instead of the default w-stacking method)

interval: tuple[int, int] | None = None

Image a set of scans between a lower (inclusive) and upper (exclusive) bound

join_channels: bool = True

Collapse the sub-band images down to an MFS image when peak-finding

join_polarizations: bool = False

Perform deconvolution by searching for peaks in the sum of squares of the polarizations, but subtract components from the individual images. Only possible when imaging two or four Stokes or linear parameters. Default: off.

local_rms: bool = True

Whether a local rms map is computed

local_rms_window: int | None = None

Size of the window used to estimate rms noise

maxw: float | None = None

A percentage specifying the maximum w-term to be gridded, relative to the max w-term being considered

mgain: float = 0.7

Major cycle gain

minuv_l: float | None = None

The minimum lambda length that the visibility data needs to meet for it to be selected for imaging

minuvw_m: float | None = None

A (u,v) selection command, where any baselines shorter than this will be ignored during imaging

multiscale: bool = False

Enable multiscale deconvolution

multiscale_fits_mask: pathlib.Path | None = None

Path to a FITS file that contains per-scale clean masks encoded as per-pixel bit values

multiscale_gain: float | None = None

Size of step made in the subminor loop of multi-scale. Default currently 0.2, but shows sign of instability. A value of 0.1 might be more stable.

multiscale_scale_bias: float = 0.75

Multiscale bias term

multiscale_scales: tuple[int, ...] | None = None

Scales used for multi-scale deconvolution

niter: int = 750000

Maximum number of minor cycles

nmiter: int = 15

Maximum number of major cycles to perform

no_mf_weighting: bool = False

Opposite of -ms-weighting; can be used to turn off MF weighting in -join-channels mode

no_reorder: bool = False

If True turn off the reordering of the MS at the beginning of wsclean

no_small_inversion: bool = False

Disables an optimisation of wsclean’s w-gridder mode. This might improve accuracy of the w-gridder.

no_update_model_required: bool = False

Will instruct wsclean not to create the MODEL_DATA column

nwlayers: int | None = None

Number of w-layers to use if the gridder mode is w-stacking

parallel_deconvolution: int | None = None

If not none, the maximum sub-image size in pixels used during deconvolution (suggestion between to between 1024-4096)

parallel_gridding: int | None = None

If not none, then this is the number of channel images that will be gridded in parallel

pol: str = 'i'

The polarisation to be imaged

save_source_list: bool = False

Saves the found clean components as a BBS/DP3 text sky model

scale: str = '2.5asec'

Pixel scale size

size: int = 10128

Image size, only a single dimension is required. Note that this means images will be squares.

squared_channel_joining: bool = False

Use with -join-channels to perform peak finding in the sum of squared values over channels, instead of the normal sum. This is useful for imaging QU polarizations with non-zero rotation measures, for which the normal sum is insensitive.

taper_gaussian: float | None = None

The size of a Gaussian function applied to the weights, assuming units of arcseconds. Defaults to None.

temp_dir: str | pathlib.Path | None = None

The path to a temporary directory where files will be written.

threshold: float | None = None

Threshold in Jy to stop cleaning

verbose: bool = False

Increase logging output

weight: str = 'briggs -0.5'

Robustness of the weighting used

wgridder_accuracy: float | None = None

The accuracy requested of the wgridder (should it be used), compared as the RMS error when compred to a DFT

class flint.imager.wsclean.WSCleanOutput

Bases: NamedTuple

An image wsclean writes, with the fields of its name

channel: int | None = None

Output channel index. None when a single channel is imaged, or for an MFS image

image_type: str

One of image, dirty, model, residual or psf

mfs: bool = False

Whether this is the multi-frequency synthesis image over all channels

path: pathlib.Path

Where the image is

pol: str | None = None

wsclean’s polarisation token, e.g. XX or Q. None when a single polarisation is imaged

class flint.imager.wsclean.WSCleanResult(/, **data: Any)

Bases: flint.options.BaseOptions

Simple container for a wsclean command.

bind_dirs: tuple[pathlib.Path, ...]

Paths that should be binded to when executing the command

cleanup: bool = True

Will clean up the dirty images/psfs/residuals/models when the imaging has completed

cmd: str

The constructede wsclean command that would be executed.

image_prefix_str: str | None = None

The prefix of the images that will be created

image_set: flint.imageset.ImageSet | None = None

The set of images produced by wsclean

move_hold_directories: tuple[pathlib.Path, pathlib.Path]

Paths that should be moved to and from when executing the command

ms: flint.ms.MS

The base measurement sets that have been included in the wsclean command.

ms_list: list[flint.ms.MS]

List of measurement sets uused in the wsclean command

options: WSCleanOptions

The set of wslean options used for imaging

flint.imager.wsclean._resolve_wsclean_key_value_to_cli_str(key: str, value: Any) → ResolvedCLIResult

An internal function intended to map a key-value pair to the appropriate form to pass to a CLI call into wsclean.

Parameters:

• key (str) – The wsclean argument name to consider. Underscores will be converted to hyphens, as expected by wsclean

• value (Any) – The value of the argument that should be converted to the appropriately formatted string

Returns:

Converted CLI output, including paths to bind to and unknown conversions

Return type:

ResolvedCLIResult

flint.imager.wsclean._wsclean_output_callback(line: str) → None

Call back function used to detect clean divergence

flint.imager.wsclean.clear_wsclean_outputs(prefix: str | pathlib.Path, channels_out: int, pols: tuple[str, ...] | None) → list[pathlib.Path]

Delete the images this run of wsclean writes that already exist under prefix

flint.imager.wsclean.cli() → None

flint.imager.wsclean.collect_wsclean_outputs(wsclean_result: WSCleanResult, prefix: str, make_cube_from_subbands: bool = True, fitscube_options: flint.options.FitsCubeOptions | None = None, extra_output_types: tuple[str, ...] | None = None) → flint.imageset.ImageSet

Gather the images wsclean wrote under prefix into an ImageSet with flint names.

With pol: xx,yy the XX and YY images become Stokes I, and are deleted unless flint_keep_linear_images is set. Images not gathered are renamed too, and left on disk with a warning. Channel images are cubed when make_cube_from_subbands is set and one polarisation remains. Raises FileNotFoundError if an expected image is missing.

flint.imager.wsclean.create_wsclean_cmd(ms_list: list[flint.ms.MS], wsclean_options: WSCleanOptions) → WSCleanResult

Create a wsclean command from a WSCleanOptions container

For the most part these are one-to-one mappings to the wsclean CLI with the exceptions being: #. the -name argument will be generated and supplied to the CLI string and will default to the parent directory and name of the supplied measurement set #. If wsclean_options.flint_hold_dir or wsclean_options.temp_dir is specified this directory is used in place of the measurement sets parent directory

If container is supplied to immediately execute this command then the output wsclean image products will be moved from the hold directory to the same directory as the measurement set.

Parameters:

• ms_list (list[MS]) – The measurement sets to be imaged

• wsclean_options (WSCleanOptions) – WSClean options to image with

• container (Optional[Path], optional) – If a path to a container is provided the command is executed immediately. Defaults to None.

Raises:

ValueError – Raised when a option has not been successfully processed

Returns:

The wsclean command to run

Return type:

WSCleanResult

flint.imager.wsclean.create_wsclean_name_argument(wsclean_options: WSCleanOptions, ms: flint.ms.MS | list[flint.ms.MS]) → pathlib.Path

Create the value that will be provided to wsclean -name argument. This has to be generated. Among things to consider is the desired output directory of imaging files. This by default will be alongside the measurement set, and is overridden by flint_hold_dir, or failing that temp_dir.

If the input ms is an instance of MSs then the first measurement set will be used to base the name from.

Parameters:

• wsclean_options (WSCleanOptions) – Set of wsclean options to consider

• ms (MS | list[MS]) – The measurement set to be imaged

Returns:

Value of the -name argument to provide to wsclean

Return type:

Path

flint.imager.wsclean.expected_wsclean_outputs(prefix: str | pathlib.Path, channels_out: int, pols: tuple[str, ...] | None, image_types: collections.abc.Collection[str] = ('image',)) → list[WSCleanOutput]

The images wsclean writes for image_types, each channel then the MFS image when there are several. The PSF is shared by every polarisation, so carries none.

flint.imager.wsclean.find_wsclean_outputs(prefix: str | pathlib.Path, channels_out: int, pols: tuple[str, ...] | None) → list[WSCleanOutput]

The images this run of wsclean writes that exist under prefix, found by exact name

flint.imager.wsclean.flint_output_path(output: WSCleanOutput, prefix: str | pathlib.Path, pol_token: str) → pathlib.Path

The flint name of a wsclean image. Its polarisation, if tagged, replaces pol_token in prefix, and its channel becomes a ch<lo>-<hi> field.

flint.imager.wsclean.get_parser() → argparse.ArgumentParser

flint.imager.wsclean.get_wsclean_output_source_list_path(name_path: str | pathlib.Path, pol: str | None = None) → pathlib.Path

WSClean can produce a text file that describes the components that it cleaned, their type, scale and brightness. These are placed in a file that is:

>> {name}.{pol}-sources.txt

where name represented the -name component. Given an input measurement set path or this -name value return the expected source list text file. pol is the stokes that the source is expected.

Parameters:

• name_path (Union[str,Path]) – Value of the -name option. If str converted to a Path

• pol (Optional[str], optional) – The polarisation to add to the name. If None the -source.txt suffix is simply appended. Defaults to None.

Returns:

Path to the source list text file

Return type:

Path

flint.imager.wsclean.image_set_from_result(wsclean_result: WSCleanResult) → flint.imageset.ImageSet | None

flint.imager.wsclean.merge_image_sets_from_results(wsclean_results: list[WSCleanResult]) → flint.imageset.ImageSet

Merge the image sets from multiple wsclean results into a single image set.

Parameters:

wsclean_results (List[WSCleanResult]) – The results of multiple wsclean imaging runs

Returns:

The merged image set

Return type:

ImageSet

flint.imager.wsclean.parse_pols(pol_str: str) → tuple[str, ...] | None

The polarisations wsclean tags each image with, None when only one is imaged. Raises NotSupportedError for an empty or unknown part, or a feed product other than xx,yy.

flint.imager.wsclean.remove_wsclean_outputs(outputs: list[WSCleanOutput], image_types: collections.abc.Collection[str] = ('dirty', 'psf', 'model', 'residual')) → list[pathlib.Path]

Delete the channel images of image_types, keeping the MFS images

flint.imager.wsclean.run_wsclean_imager(wsclean_result: WSCleanResult, container: pathlib.Path, make_cube_from_subbands: bool = True, fitscube_options: flint.options.FitsCubeOptions | None = None, extra_output_types: tuple[str, ...] | None = None) → flint.imageset.ImageSet

Run a provided wsclean command. Optionally will clean up files, including the dirty beams, psfs and other assorted things.

An ImageSet is constructed that attempts to capture the output wsclean image products. If image_prefix_str is specified the image set will be created by (ordered by preference): #. Adding the image_prefix_str to the move_directory #. Guessing it from the path of the measurement set from wsclean_result.ms.path

Parameters:

• wsclean_result (WSCleanResult) – The command to run, and other properties (cleanup.)

• container (Path) – Path to the container with wsclean available in it

• bind_dirs (Optional[Tuple[Path, ...]], optional) – Additional directories to include when binding to the wsclean container. Defaults to None.

• move_hold_directories (Optional[Tuple[Path,Optional[Path]]], optional) – The move_directory and hold_directory passed to the temporary context manager. If None no hold_then_move_into manager is used. Defaults to None.

• make_cube_from_subbands (bool, optional) – Form a single FITS cube from the set of sub-band images wsclean produces. Defaults to False.

• image_prefix_str (Optional[str], optional) – The name used to search for wsclean outputs. If None, it is guessed from the name and location of the MS. Defaults to None.

• extra_output_types (tuple[str, ...] | None, optional) – Additional output types (beyond image/residual) to search for and attach to the returned ImageSet, e.g. ("model",). Defaults to None.

Returns:

The executed wsclean output products.

Return type:

ImageSet

flint.imager.wsclean.verify_wsclean_outputs(outputs: list[WSCleanOutput], prefix: str | pathlib.Path, channels_out: int, pols: tuple[str, ...] | None) → None

Raise FileNotFoundError if an image this run should write is not among outputs. Other types are not checked, as cleaning up deletes them.

flint.imager.wsclean.wsclean_imager(ms: pathlib.Path | flint.ms.MS | tuple[flint.ms.MS | pathlib.Path, ...] | list[flint.ms.MS | pathlib.Path], wsclean_container: pathlib.Path, update_wsclean_options: dict[str, Any] | None = None, update_fitscube_options: dict[str, Any] | None = None, make_cube_from_subbands: bool = True, extra_output_types: tuple[str, ...] | None = None) → WSCleanResult

Create and run a wsclean imager command against a measurement set.

Parameters:

• ms (Path | MS | tuple[MS | Path, ...] | list[MS | Path]) – Path to the measurement set that will be imaged

• wsclean_container (Path) – Path to the container with wsclean installed

• update_wsclean_options (Optional[Dict[str, Any]], optional) – Additional options to update the generated WscleanOptions with. Keys should be attributes of WscleanOptions. Defaults to None.

Returns:

_description_

Return type:

WSCleanResult

flint.imager.wsclean.wsclean_output_path(prefix: str | pathlib.Path, image_type: str, channel: int | None = None, mfs: bool = False, pol: str | None = None) → pathlib.Path

The name wsclean gives an image, <prefix>[-<channel>|-MFS][-<pol>]-<type>.fits