flint.naming

Attempts to centralise components to do with naming of pipeline files and data products.

Attributes

LONG_FIELD_TO_SHORTHAND	Name mapping between the longform of ProcessedFieldComponents and shorthands used
PathStr
ResolutionModes

Classes

AegeanNames	Base names that would be used in various Aegean related tasks
CASDANameComponents	Container for the components of a CASDA MS. These are really those
FITSMaskNames	Contains the names of the FITS images created when creating a mask image/
LinmosNames	Creates expected output names for the yandasoft linmos task.
ProcessedNameComponents	Container for a file name derived from a MS flint name. Generally of the
RawNameComponents

Functions

_rename_linear_to_stokes(→ str)
add_timestamp_to_path(→ pathlib.Path)	Add a timestamp to a input path, where the timestamp is the
casda_ms_format(→ CASDANameComponents | None)	Break up a CASDA sty;e MS name (really the askap pipeline format) into its recognised parts.
create_aegean_names(→ AegeanNames)	Create the expected names for aegean and its tools.
create_fits_mask_names(→ FITSMaskNames)	Create the names that will be used when generate FITS mask products
create_image_cube_name(→ pathlib.Path)	Create a consistent naming scheme when combining images into cube images. Intended to
create_imaging_name_prefix(→ str)	Given a measurement set and a polarisation, create the naming prefix to be used
create_linmos_base_path(→ pathlib.Path)	Create the base path of a yandasoft linmos given a set of input images.
create_linmos_names(→ LinmosNames)	This creates the names that would be output but the yandasoft
create_ms_name(→ str)	Create a consistent naming scheme for measurement sets. At present
create_name_from_common_fields(→ pathlib.Path)	Attempt to craft a base name using the field elements that are in common.
create_path_from_processed_name_components(→ pathlib.Path)	Given an input ProcessedNameComponents create the corresponding path
extract_beam_from_name(→ int)	Attempts to extract the beam number from some input name should it follow a
extract_components_from_name(...)	Attempts to break down a file name of a recognised format into its principal compobnents.
get_aocalibrate_output_path(→ pathlib.Path)	Create a name for an aocalibrate style bandpass solution.
get_beam_resolution_str(→ str)	Map a beam resolution mode to an appropriate suffix. This
get_fits_cube_from_paths(→ list[pathlib.Path])	Given a list of files, find the ones that appear to be FITS files
get_potato_output_base_path(→ pathlib.Path)	Return the base name for potato peel.
get_sbid_from_path(→ int)	Attempt to extract the SBID of a observation from a path. It is a fairly simple ruleset
get_selfcal_ms_name(→ pathlib.Path)	Create the new output MS path that will be used for self-calibration. The
processed_ms_format(→ ProcessedNameComponents | None)	Will take a formatted name (i.e. one derived from the flint.naming.create_ms_name)
raw_ms_format(→ None | RawNameComponents)	The typical ASKAP measurement written to the ingest disks
rename_linear_to_stokes(→ PathStr)
split_and_get_images(→ list[pathlib.Path])	Split a list of images by a given field and return the images that match
split_images(→ dict[str, list[pathlib.Path]])	Split a list of images by a given field. This is intended to be used
update_beam_resolution_field_in_path(→ pathlib.Path)	Transition the resolution indicator in a processed name (either optimal or fixed)

Module Contents

class flint.naming.AegeanNames

Bases: NamedTuple

Base names that would be used in various Aegean related tasks

bkg_image: pathlib.Path

Background map computed by BANE

comp_cat: pathlib.Path

Component catalogue produced by the aegean source finder

ds9_region: pathlib.Path

DS9 region overlay file

resid_image: pathlib.Path

Residual map after subtracting component catalogue produced by AeRes

rms_image: pathlib.Path

RMS noise map computed by BANE

class flint.naming.CASDANameComponents

Bases: NamedTuple

Container for the components of a CASDA MS. These are really those processed by the ASKAP pipeline

alias: str | None = None

Older ASKAP MSs could be packed with multiple fields. The ASKAP pipeline holds this field as an alias. They are now the same in almost all cases as the field.

beam: str

Beam number of the data

field: str

The name of the field extracted

format: str = 'science'

What the format / type of the data the MS is.

sbid: int

The sbid of the observation

spw: str | None = None

If multiple MS were written as the data were in a high-frequency resolution mode, which segment

class flint.naming.FITSMaskNames

Bases: NamedTuple

Contains the names of the FITS images created when creating a mask image/ These are only the names, and do not mean that they are necessarily created.

mask_fits: pathlib.Path

Name of the mask FITS file

scale_mask_fits: pathlib.Path | None = None

Path to a FITS file that describes per-scale clean masks. Scales are represented with a bit-mapped values,

signal_fits: pathlib.Path | None = None

Name of the signal FITS file

class flint.naming.LinmosNames

Bases: NamedTuple

Creates expected output names for the yandasoft linmos task.

image_fits: pathlib.Path

Path to the final fits co-added image

parset_output_path: pathlib.Path

Path to the output parset generated

weight_fits: pathlib.Path

Path to the weights fits image created when co-adding images

class flint.naming.ProcessedNameComponents

Bases: NamedTuple

Container for a file name derived from a MS flint name. Generally of the form: SB.Field.Beam.Spw

beam: str | None = None

The beam of the observation processed

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

The channel range encoded in a file name. Generally are zero-padded, and are two fields of the form ch1234-1235, where the upper bound is exclusive. Defaults to None.

field: str

The name of the field extracted

pol: str | None = None

The polarisation component, if it exists, in a filename. Examples are ‘i’,’q’,’u’,’v’. Could be combinations in some cases depending on how it was created (e.g. based on wsclean pol option).

round: str | None = None

The self-calibration round detected. This might be represented as ‘noselfcal’ in some image products, e.g. linmos.

sbid: str

The sbid of the observation

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

The scane range encoded in a file name. Generally are zero-padded and are two fields of the form scan1234-1235, where the epper bound is exclusive. Defaults to None.

spw: str | None = None

The SPW of the observation. If there is only one spw this is None.

class flint.naming.RawNameComponents

Bases: NamedTuple

beam: str

Beam number of the data

date: str

Date that the data were taken, of the form YYYY-MM-DD

spw: str | None = None

If multiple MS were written as the data were in a high-frequency resolution mode, which segment

time: str

Time that the data were written

flint.naming._rename_linear_to_stokes(linear_name_str: str, stokes: str) → str

flint.naming.add_timestamp_to_path(input_path: pathlib.Path | str, timestamp: datetime.datetime | None = None) → pathlib.Path

Add a timestamp to a input path, where the timestamp is the current data and time. The time will be added to the name component before the file suffix. If the name component of the input_path has multiple suffixes than the timestamp will be added before the last.

Parameters:

• input_path (Union[Path, str]) – Path that should have a timestamp added

• timestamp – (Optional[datetime], optional): The date-time to add. If None the current time is used. Defaults to None.

Returns:

Updated path with a timestamp in the file name

Return type:

Path

flint.naming.casda_ms_format(in_name: str | pathlib.Path) → CASDANameComponents | None

Break up a CASDA sty;e MS name (really the askap pipeline format) into its recognised parts. if a match fails a None is returned.

Example of a CASDA style MS:

• scienceData.RACS_1237+00.SB40470.RACS_1237+00.beam35_averaged_cal.leakage.ms

Parameters:

in_name (Union[str, Path]) – The path to or name of the MS to consider

Returns:

The returned components of the MS. If this fails a None is returned.

Return type:

Union[CASDANameComponents, None]

flint.naming.create_aegean_names(base_output: str) → AegeanNames

Create the expected names for aegean and its tools.

Parameters:

base_output (str) – The base name that aegean outputs are built from.

Returns:

A collection of names to be produced by Aegean related tasks

Return type:

AegeanNames

flint.naming.create_fits_mask_names(fits_image: str | pathlib.Path, include_signal_path: bool = False) → FITSMaskNames

Create the names that will be used when generate FITS mask products

Parameters:

• fits_image (Union[str,]Path) – Base name of the output files

• include_signal_path (bool, optional) – If True, also include signal_fits in the output. Defaults to False.

Returns:

collection of names used for the signal and mask FITS images

Return type:

FITSMaskNames

flint.naming.create_image_cube_name(image_prefix: pathlib.Path, mode: str | list[str] | None = None, suffix: str | list[str] | None = None) → pathlib.Path

Create a consistent naming scheme when combining images into cube images. Intended to be used when combining many subband images together into a single cube.

The name returned will be: >>> {image_prefix}.{mode}.{suffix}.cube.fits

Should mode or suffix be a list, they will be joined with ‘.’ separators. Hence, no ‘.’ should be added.

This function will always output ‘cube.fits’ at the end of the returned file name.

Parameters:

• image_prefix (Path) – The unique path of the name. Generally this is the common part among the input planes

• mode (Optional[Union[str, List[str]]], optional) – Additional mode/s to add to the file name. Defaults to None.

• suffix (Optional[Union[str, List[str]]], optional) – Additional suffix/s to add before the final ‘cube.fits’. Defaults to None.

Returns:

The final path and file name

Return type:

Path

flint.naming.create_imaging_name_prefix(ms_path: pathlib.Path, pol: str | None = None, channel_range: tuple[int, int] | None = None, scan_range: tuple[int, int] | None = None) → str

Given a measurement set and a polarisation, create the naming prefix to be used by some imager

Parameters:

• ms (Union[MS,Path]) – The measurement set being considered

• pol (Optional[str], optional) – Whether a polarsation is being considered. Defaults to None.

• channel_range (Optional[Tuple[int,int]], optional) – The channel range that is going to be imaged. Defaults to none.

• scan_range (Optional[Tuple[int,int]], optional) – The scan range that is going to be imaged. Defaults to none.

Returns:

The constructed string name

Return type:

str

flint.naming.create_linmos_base_path(input_images: list[pathlib.Path], additional_suffixes: str | None = None) → pathlib.Path

Create the base path of a yandasoft linmos given a set of input images. The default operation is to form the name from the common processed name fields amount all of the input images.

Parameters:

• input_images (list[Path] | None, optional) – If provided the common fields of the input images are used as basis of the path. Defaults to None.

• additional_suffixes (str | None, optional) – Any additional suffixes to append. Defaults to None.

Returns:

The full path of the parset file

Return type:

Path

flint.naming.create_linmos_names(name_prefix: str | pathlib.Path, parset_output_path: pathlib.Path | None = None) → LinmosNames

This creates the names that would be output but the yandasoft linmos task. It returns the names for the linmos and weight maps that linmos would create. These names will have the .fits extension with them, but be aware that when the linmos parset if created these are omitted.

Parameters:

name_prefix (str | Path) – The prefix of the filename that will be used to create the linmos and weight file names.

Returns:

Collection of expected filenames

Return type:

LinmosNames

flint.naming.create_ms_name(ms_path: pathlib.Path, sbid: int | None = None, field: str | None = None) → str

Create a consistent naming scheme for measurement sets. At present it is intended to be used for splitting fields from raw measurement sets, but can be expanded.

Parameters:

• ms_path (Path) – The measurement set being considered. A RawNameComponents will be constructed against it.

• sbid (Optional[int], optional) – An explicit SBID to include in the name, otherwise one will attempted to be extracted the the ms path. If these fail the sbid is set of 00000. Defaults to None.

• field (Optional[str], optional) – The field that this measurement set will contain. Defaults to None.

Returns:

The filename of the measurement set

Return type:

str

flint.naming.create_name_from_common_fields(in_paths: tuple[pathlib.Path, Ellipsis], additional_suffixes: str | None = None) → pathlib.Path

Attempt to craft a base name using the field elements that are in common. The expectation that these are paths that can be processed by the processed_name_format handler. Resulting fields that are common across all in_paths are preserved.

Only fields that are recognised as a known property are retained. Suffixes that do not form a component are ignored. For example:

>>> "59058/SB59058.RACS_1626-84.ch0287-0288.linmos.fits"

the linmos.fits would be ignored.

All in_paths should be detected, otherwise an ValueError is raised

Parameters:

• in_paths (Tuple[Path, ...]) – Collection of input paths to consider

• additional_suffixes (Optional[str], optional) – Add an additional set of suffixes before returning. Defaults to None.

Raises:

ValueError – Raised if any of the in_paths fail to conform to flint processed name format

Returns:

Common fields with the same base parent path

Return type:

Path

flint.naming.create_path_from_processed_name_components(processed_name_components: ProcessedNameComponents, parent_path: pathlib.Path | None = None) → pathlib.Path

Given an input ProcessedNameComponents create the corresponding path

Parameters:

• processed_name_components (ProcessedNameComponents) – The naming specification to create

• parent_path (Path | None, optional) – The parent directory of the output path. Defaults to None.

Returns:

A directory with following the specification of the input ProcessedNameComponents

Return type:

Path

flint.naming.extract_beam_from_name(name: str | pathlib.Path) → int

Attempts to extract the beam number from some input name should it follow a known naming convention.

Parameters:

name (Union[str,Path]) – The name to examine to search for the beam number.

Raises:

ValueError – Raised if the name was not was not successfully matched against the known format

Returns:

Beam number that extracted from the input name

Return type:

int

flint.naming.extract_components_from_name(name: str | pathlib.Path) → RawNameComponents | ProcessedNameComponents | CASDANameComponents

Attempts to break down a file name of a recognised format into its principal compobnents. Presumably this is a measurement set or something derived from it (i.e. images).

There are two formats currently recognised: - the raw measurement set format produced by the ASKAP ingest system (date, time, beam, spw, underscore delimited) - a formatted name produced by flint (SBID, field, beam, spw, dot delimited)

Internally this function attempts to run two regular expression filters against the input, and returns to the set of components that a filter has matched.

Parameters:

name (Union[str,Path]) – The name to examine to search for the beam number.

Raises:

ValueError – Raised if the name was not was not successfully matched against the known format

Returns:

The extracted name components within a name

Return type:

Union[RawNameComponents,ProcessedNameComponents,CASDANamedComponents]

flint.naming.get_aocalibrate_output_path(ms_path: pathlib.Path, include_preflagger: bool, include_smoother: bool) → pathlib.Path

Create a name for an aocalibrate style bandpass solution.

Parameters:

• ms_path (Path) – Path of the measurement set that the aocalibrate file will be created for

• include_preflagger (bool) – Whether to include the .preflagged term to indicate that the preflagger has been executed

• include_smoother (bool) – Whether to include the .smoothed term to included that bandpas smoothing has been performed

Returns:

The constructed output path of the solutions file. This include the parent directory of the input measurement set

Return type:

Path

flint.naming.get_beam_resolution_str(mode: ResolutionModes, marker: str | None = None) → str

Map a beam resolution mode to an appropriate suffix. This is located her in anticipation of other imaging modes.

Supported modes are: ‘optimal’, ‘fixed’, ‘raw’

Parameters:

• mode (Literal["fixed","optimal"]) – The mode of image resolution to use.

• marker (Optional[str], optional) – Append the marker to the end of the returned mode string. If None mode string is returned. Defaults to None.

Raises:

ValueError – Raised when an unrecognised mode is supplied

Returns:

The appropriate string for mapped mode

Return type:

str

flint.naming.get_fits_cube_from_paths(paths: list[pathlib.Path]) → list[pathlib.Path]

Given a list of files, find the ones that appear to be FITS files and contain the .cube. field indicator. A regular expression searching for both the .cube. and .fits file type is used.

Parameters:

paths (List[Path]) – The set of paths to examine to identify potential cube fits images from

Returns:

Set of paths matching the search criteria

Return type:

List[Path]

flint.naming.get_potato_output_base_path(ms_path: pathlib.Path) → pathlib.Path

Return the base name for potato peel.

Parameters:

ms_path (Path) – Input measurement set that follows the FLINT style process name format

Returns:

Output base name to use

Return type:

Path

flint.naming.get_sbid_from_path(path: pathlib.Path) → int

Attempt to extract the SBID of a observation from a path. It is a fairly simple ruleset that follows the typical use cases that are actually in practise. There is no mechanism to get the SBID from the measurement set meta-data.

If the path provided ends in a .ms suffix, the parent directory is assumed to be named the sbid. Otherwise, the name of the subject directory is. A test is made to ensure the sbid is made up of integers only.

Parameters:

path (Path) – The path that contains the sbid to extract.

Raises:

ValueError – Raised when the SBID extracted from the path is not all digits

Returns:

The sbid extracted

Return type:

int

flint.naming.get_selfcal_ms_name(in_ms_path: pathlib.Path, round: int = 1) → pathlib.Path

Create the new output MS path that will be used for self-calibration. The output measurement set path will include a roundN.ms suffix, where N is the round. If such a suffix already exists from an earlier self-calibration round, it will be removed and replaced.

Parameters:

• in_ms_path (Path) – The measurement set that will go through self-calibration

• round (int, optional) – The self-calibration round number that is currently being used. Defaults to 1.

Returns:

Output measurement set path to use

Return type:

Path

flint.naming.processed_ms_format(in_name: str | pathlib.Path) → ProcessedNameComponents | None

Will take a formatted name (i.e. one derived from the flint.naming.create_ms_name) and attempt to extract its main components. This includes the SBID, field, beam and spw.

Parameters:

in_name (Union[str, Path]) – The name that needs to be broken down into components

Returns:

A structure container the sbid, field, beam and spw. None is returned if can not be parsed.

Return type:

Union[FormattedNameComponents,None’

flint.naming.raw_ms_format(in_name: str) → None | RawNameComponents

The typical ASKAP measurement written to the ingest disks has the form:

>>> 2022-04-14_100122_1.ms

and in the case of a multiple beams written out (in high frequency resolution mode)

>>> 2022-04-14_100122_1_1.ms

This function will attempt to break it up into its main parts and return the mapping.

Parameters:

in_name (str) – The name of a file, presumably a measurement set. The left-most part will be examined for to identify the raw ASKAP naming scheme.

Returns:

None if the raw ASKAP measurement set naming scheme was not detected, otherwise a dictionary representing its parts.

Return type:

Union[None,Dict[str,str]]

flint.naming.rename_linear_to_stokes(linear_name: PathStr, stokes: str) → PathStr

flint.naming.split_and_get_images(images: list[pathlib.Path], get: str, by: str = 'pol') → list[pathlib.Path]

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

Parameters:

• images (list[Path]) – The images to split

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

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

Raises:

ValueError – If the field to extract is not found in the split images

Returns:

The images that match the field of interest

Return type:

list[Path]

flint.naming.split_images(images: list[pathlib.Path], by: str = 'pol') → dict[str, list[pathlib.Path]]

Split a list of images by a given field. This is intended to be used when a set of images are to be split by a common field, such as polarisation.

Parameters:

• images (List[Path]) – The images to split

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

Returns:

A dictionary of the images split by the field

Return type:

Dict[str,List[Path]]

flint.naming.update_beam_resolution_field_in_path(path: pathlib.Path, original_mode: ResolutionModes, updated_mode: ResolutionModes, marker: str | None = None) → pathlib.Path

Transition the resolution indicator in a processed name (either optimal or fixed) to another state. For example:

>>> 'SB57516.RACS_0929-81.round4.i.optimal.round4.residual.linmos.fits'

to

>>> 'SB57516.RACS_0929-81.round4.i.fixed.round4.residual.linmos.fits'

See get_beam_resolution_str for addition information. Supported modes are fixed and optimal

Parameters:

• path (Path) – The path to inspect and update

• original_mode (ResolutionModes) – The original mode

• updated_mode (ResolutionModes) – The mode to move to

• marker (str | None, optional) – The marker to separate the field. Defaults to None.

Returns:

Updated path

Return type:

Path

flint.naming.LONG_FIELD_TO_SHORTHAND

Name mapping between the longform of ProcessedFieldComponents and shorthands used

flint.naming.PathStr

flint.naming.ResolutionModes