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

ImageSet	A structure to represent the images and auxiliary products produced by
ResolvedCLIResult	Mapping results to provide to wsclean
WSCleanOptions	A basic container to handle WSClean options. These attributes should
WSCleanResult	Simple container for a wsclean command.

Functions

_make_pols(→ tuple[str, Ellipsis] | None)	Create a tuple of polarisations from a polarisation string
_rename_wsclean_file(→ pathlib.Path)	Rename a file with wsclean appended string information to convert its
_rename_wsclean_title(→ str)	Construct and apply a regular expression that aims to identify
_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
cli(→ None)
combine_image_set_to_cube(→ ImageSet)	Combine wsclean subband channel images into a cube. Each collection attribute
combine_images_to_cube(→ pathlib.Path)	Combine wsclean subband channel images into a cube. Each collection attribute
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
delete_wsclean_outputs(→ list[pathlib.Path])	Attempt to remove elected wsclean output files
get_parser(→ argparse.ArgumentParser)
get_wsclean_output_names(, check_exists_when_adding)	Attempt to generate the file names and paths that would be
get_wsclean_output_source_list_path(→ pathlib.Path)	WSClean can produce a text file that describes the components
image_set_from_result(→ ImageSet | None)
merge_image_sets(→ ImageSet)	Merge the image sets from multiple wsclean results into a single image set.
merge_image_sets_from_results(→ ImageSet)	Merge the image sets from multiple wsclean results into a single image set.
rename_wsclean_prefix_in_image_set(→ ImageSet)	Given an input image_set, rename the files contained in it to
rotate_cube(→ pathlib.Path)	Rotate the FITS cube axes to a shape of (chan, pol, dec, ra)
run_wsclean_imager(→ ImageSet)	Run a provided wsclean command. Optionally will clean up files,
split_and_get_image_set(→ list[pathlib.Path])	Split an ImageSet by a given field and return the images that match the field of interest.
split_image_set(→ dict[str, list[pathlib.Path]])	Split an ImageSet by a given field.
wsclean_cleanup_files(, single_channel, Ellipsis])	Clean up (i.e. delete) files from wsclean.
wsclean_imager(→ WSCleanResult)	Create and run a wsclean imager command against a measurement set.

Module Contents

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

Bases: flint.options.BaseOptions

A structure to represent the images and auxiliary products produced by wsclean

dirty: list[pathlib.Path] | None = None

Dirty images.

image: list[pathlib.Path]

Images produced.

model: list[pathlib.Path] | None = None

Model images.

prefix: str

Prefix of the images and other output products. This should correspond to the -name argument from wsclean

psf: list[pathlib.Path] | None = None

References to the PSFs produced by wsclean.

residual: list[pathlib.Path] | None = None

Residual images.

source_list: pathlib.Path | None = None

Path to a source list that accompanies the image data

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 the temp_dir options 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.

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_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

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_make_cube_inplace: bool = True

Rotate the cube for the linmos axis ordering in place, or do it via a temporary file that then gets deleted. Good thing to turn off when getting weird OSErrors on file writing

flint_no_log_wsclean_output: bool = False

If True do not log the wsclean output

force_mask_rounds: int | None = None

Round of force masked derivation

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, Ellipsis] | 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.

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

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.WSCleanResult(/, **data: Any)

Bases: flint.options.BaseOptions

Simple container for a wsclean command.

bind_dirs: tuple[pathlib.Path, Ellipsis]

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: 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.options.MS

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

options: WSCleanOptions

The set of wslean options used for imaging

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

Create a tuple of polarisations from a polarisation string

Parameters:

pol_str (str) – Polarisation string to convert

Returns:

Tuple of polarisations

Return type:

tuple[str, …] | None

flint.imager.wsclean._rename_wsclean_file(input_path: pathlib.Path, rename_file: bool = False) → pathlib.Path

Rename a file with wsclean appended string information to convert its - separation markers with .. This should handle skipping the field name of the target field observed.

Parameters:

• input_path (Path) – The file path that would be examined and modified

• clean_parts (Union[int, Tuple[int, ...]], optional) – Which parts of a split string will be modified. Defaults to -2.

• rename_file (bool, optional) – Whether the file should be moved / renamed. Defaults to False.

Returns:

Path to the renamed file

Return type:

Path

flint.imager.wsclean._rename_wsclean_title(name_str: str) → str

Construct and apply a regular expression that aims to identify the wsclean appended properties string within a file and replace the - separator with a ..

A simple replace of all - with . may not be ideal if the character has been used on purpose.

Parameters:

name_str (str) – The name that will be extracted and modified

Returns:

The modified string if a wsclean string was matched, otherwise the input name-str

Return type:

str

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.cli() → None

flint.imager.wsclean.combine_image_set_to_cube(image_set: ImageSet, remove_original_images: bool = False, inplace: bool = True) → ImageSet

Combine wsclean subband channel images into a cube. Each collection attribute of the input image_set will be inspected. The MFS images will be ignored.

A output file name will be generated based on the prefix and mode (e.g. image, residual, psf, dirty).

Parameters:

• image_set (ImageSet) – Collection of wsclean image productds

• remove_original_images (bool, optional) – If True, images that went into the cube are removed. Defaults to False.

• inplace (bool, optional) – If True, modify the file in-place. If False, write to a temporary file and

• True (then replace the original. Default)

Returns:

Updated iamgeset describing the new outputs

Return type:

ImageSet

flint.imager.wsclean.combine_images_to_cube(images: list[pathlib.Path], prefix: str, mode: str, remove_original_images: bool = False, inplace: bool = True) → pathlib.Path

Combine wsclean subband channel images into a cube. Each collection attribute of the input image_set will be inspected. The MFS images will be ignored.

A output file name will be generated based on the prefix and mode (e.g. image, residual, psf, dirty).

Parameters:

• image_set (ImageSet) – Collection of wsclean image productds

• remove_original_images (bool, optional) – If True, images that went into the cube are removed. Defaults to False.

• inplace (bool, optional) – If True, modify the file in-place. If False, write to a temporary file and

• True (then replace the original. Default)

Returns:

Updated iamgeset describing the new outputs

Return type:

ImageSet

flint.imager.wsclean.create_wsclean_cmd(ms: flint.options.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.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 temp-dir to the same directory as the measurement set.

Parameters:

• ms (MS) – The measurement set 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.options.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. If a temp_dir has been specified then output files will be written here.

Parameters:

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

• ms (MS) – The measurement set to be imaged

Returns:

Value of the -name argument to provide to wsclean

Return type:

Path

flint.imager.wsclean.delete_wsclean_outputs(prefix: str, output_type: str = 'image', ignore_mfs: bool = True, no_subbands: bool = False) → list[pathlib.Path]

Attempt to remove elected wsclean output files

If no_subbands is True (as in channels_out is 1) then nothing is deleted.

Parameters:

• prefix (str) – The prefix of the files to remove. This would correspond to the -name of wsclean.

• output_type (str, optional) – What type of wsclean output to try to remove. Defaults to ‘image’.

• ignore_mfs (bool, optional) – If True, do not remove MFS outputs (attempt to, at least). Defaults to True.

• no_subbands (bool, Optional) – If True, nothing is deleted. Defaults to False.

Returns:

The paths that were removed (or at least attempted to be removed)/

Return type:

List[Path]

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

flint.imager.wsclean.get_wsclean_output_names(prefix: str, subbands: int | None = None, pols: str | tuple[str] | tuple[str, Ellipsis] | None = None, verify_exists: bool = False, include_mfs: bool = True, output_types: str | Collection[str] = ('image', 'dirty', 'residual', 'model', 'psf'), check_exists_when_adding: bool = False) → ImageSet

Attempt to generate the file names and paths that would be created by an imaging run of wsclean. This is done using a the known (expected) naming format of modern wsclean versions.

Checks can be made to ensure a file exists before adding it to the output ImageSet. This might be important as some wsclean image products might be deleted in order to preserve disk space.

Parameters:

• prefix (str) – The prefix of the imaging run (akin to -name option in wsclean call)

• subbands (int) – Number of subbands that were imaged

• pols (Optional[Union[str,Tuple[str]]], optional) – The polarisation of the image. If None are provided then this is not used. Multiple polarisation may be supplied. If multiple pols are given in an iterable, each will be produced. Defaults to None.

• verify_exists (bool, optional) – Ensures that each generated path corresponds to an actual file. Defaults to False.

• include_mfs (bool, optional) – Include the MFS images produced by wsclean. Defaults to True.

• output_types (Union[str,Collection[str]]) – Include files of this type, including image, dirty, residual, model, psf. Defaults to (‘image’,’dirty’,’residual’,’model’, ‘psf’).

• check_exists_when_adding (bool, optional) – Only add a generated file if it exists. Although related to the verify_exists, when this optional is True files will silently be ignored if they are not found. if this and verify_exists are both True then no errors on missing files will be raised. Defaults to False.

Raises:

FileExistsError – Raised when a file does not exist and verify_exists is True.

Returns:

The file paths that wsclean should create/has created.

Return type:

ImageSet

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) → ImageSet | None

flint.imager.wsclean.merge_image_sets(image_sets: list[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.merge_image_sets_from_results(wsclean_results: list[WSCleanResult]) → 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.rename_wsclean_prefix_in_image_set(input_image_set: ImageSet) → ImageSet

Given an input image_set, rename the files contained in it to remove the - separator that wsclean uses and replace it with ..

Files will be renamed on disk appropriately.

Parameters:

input_image_set (ImageSet) – The collection of output wsclean products

Returns:

The updated image_set after replacing the separator and renaming files

Return type:

ImageSet

flint.imager.wsclean.rotate_cube(output_cube_path: str | pathlib.Path, inplace: bool = True) → pathlib.Path

Rotate the FITS cube axes to a shape of (chan, pol, dec, ra) which is what yandasoft linmos tasks expect.

Parameters:

• output_cube_path (str | Path) – Path to the FITS cube to rotate.

• inplace (bool, optional) – If True, modify the file in-place. If False, write to a temporary file and then replace the original. Default True

Returns:

Path to the rotated FITS cube.

Return type:

Path

flint.imager.wsclean.run_wsclean_imager(wsclean_result: WSCleanResult, container: pathlib.Path, make_cube_from_subbands: bool = True) → 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.

Returns:

The executed wsclean output products.

Return type:

ImageSet

flint.imager.wsclean.split_and_get_image_set(image_set: ImageSet, get: str, by: str = 'pol', mode: str = 'image') → list[pathlib.Path]

Split an ImageSet by a given field and return the images that match the field of interest.

Parameters:

• image_set (ImageSet) – The ImageSet to split

• get (str) – The field to extract from the split images

• by (str, optional) – The field to split by. Defaults to “pol”.

• mode (str, optional) – The mode to extract from ImageSet. Defaults to “image”.

Raises:

NamingException – If the field to split by is not found in the ImageSet

Returns:

The images that match the field of interest

Return type:

list[Path]

flint.imager.wsclean.split_image_set(image_set: ImageSet, by: str = 'pol', mode: str = 'image') → dict[str, list[pathlib.Path]]

Split an ImageSet by a given field.

Parameters:

• image_set (ImageSet) – The ImageSet to split

• by (str, optional) – Field to split images by. Defaults to “pol”.

• mode (str, optional) – Type of images to extract from ImageSet. Defaults to “image”.

Raises:

NamingException – If the field to split by is not found in the ImageSet

Returns:

_description_

Return type:

dict[str, list[Path]]

flint.imager.wsclean.wsclean_cleanup_files(prefix: str | pathlib.Path, output_types: tuple[str, Ellipsis] | None = ('dirty', 'psf', 'model', 'residual'), single_channel: bool = False) → tuple[pathlib.Path, Ellipsis]

Clean up (i.e. delete) files from wsclean.

Parameters:

• prefix (Union[str, Path]) – The prefix used to search for files. This is generally the -name

• output_types (Optional[Tuple[str]], optional) – Which type of output wsclean products to delete. Defaults to (“dirty”, “psf”, “model”, “residual”).

• single_channel (bool, optional) – Whether there is the subband part of the wsclean file names to consider. Defaults to False.

Returns:

Set of files that were deleted

Return type:

Tuple[Path, …]

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

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

Parameters:

• ms (Union[Path,MS]) – 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