flint.ms¶

Operations related to measurement sets

Measurement sets are represented by the MS class. It describes the path to a measurement set and any activate columns that should be used for certain operations. Throughout flint steps are generally carried out against the named column described by the column attribute.

Classes¶

MSSummary

Small structure to contain overview of a MS

Functions¶

`check_column_in_ms`(→ bool)	Checks to see whether a column exists in an MS. If column is provided this
`cli`(→ None)
`consistent_channelwise_frequencies`(→ numpy.ndarray)	Given a collection of frequencies in the form of
`consistent_ms`(→ bool)	Perform some basic consistency checks to ensure MS1 can
`consistent_ms_frequencies`(→ bool)	Given a set of measurement sets, inspect the frequencies
`copy_and_preprocess_casda_askap_ms`() → flint.options.MS)	Convert an ASKAP pipeline MS from CASDA into a FLINT form. This involves
`critical_ms_interaction`(input_ms[, copy, suffix])	A context manager that can be used to register a measurement set as
`describe_ms`(→ MSSummary)	Print some basic information from the inpute measurement set.
`find_mss`(→ tuple[flint.options.MS, Ellipsis])	Search a directory to find measurement sets via a simple
`get_beam_from_ms`(→ int)	Lookup the ASKAP beam number from a measurement set.
`get_field_id_for_field`(→ int \| None)	Return the FIELD_ID for an elected field in a measurement set
`get_freqs_from_ms`(→ numpy.ndarray)	Return the frequencies observed from an ASKAP Measurement set.
`get_parser`(→ argparse.ArgumentParser)
`get_phase_dir_from_ms`(→ astropy.coordinates.SkyCoord)	Extract the phase direction from a measurement set.
`get_pol_axis_from_ms`(→ astropy.units.Quantity)	Get the polarization axis from the ASKAP MS. Checks are performed
`get_telescope_location_from_ms`(...)	Return the telescope location from an ASKAP Measurement set.
`get_times_from_ms`(...)	Return the observation times from an ASKAP Measurement set.
`preprocess_askap_ms`(→ flint.options.MS)	The ASKAP MS stores its data in a way that is not immediately accessible
`remove_columns_from_ms`(→ list[str])	Attempt to remove a collection of columns from a measurement set.
`rename_column_in_ms`(→ flint.options.MS)	Rename a column in a measurement set. Optionally update the tracked
`rename_ms_and_columns_for_selfcal`(→ flint.options.MS)	Take an existing measurement set, rename it, and appropriately
`split_by_field`(→ list[flint.options.MS])	Attempt to split an input measurement set up by the unique FIELDs recorded
`subtract_model_from_data_column`(→ flint.options.MS)	Execute a `taql` query to subtract the MODEL_DATA from a nominated data column.

Module Contents¶

class flint.ms.MSSummary[source]¶

Bases: NamedTuple

Small structure to contain overview of a MS

ants: list[int][source]¶: Collection of unique antennas

beam: int[source]¶: The ASKAP beam number of the measurement set

fields: list[str][source]¶: Collection of unique field names from the FIELDS table

flag_spectrum: numpy.ndarray[source]¶: Flagged spectral channels

flagged: int[source]¶: Number of flagged records

path: pathlib.Path[source]¶: Path to the measurement set that is being represented

phase_dir: astropy.coordinates.SkyCoord[source]¶: The phase direction of the measurement set, which will be where the image will be centred

spw: int | None = None[source]¶: Intended to be used with ASKAP high-frequency resolution modes, where the MS is divided into SPWs

unflagged: int[source]¶: Number of unflagged records

flint.ms.check_column_in_ms(ms: flint.options.MS | str | os.PathLike, column: str | None = None, sub_table: str | None = None) → bool[source]¶

Checks to see whether a column exists in an MS. If column is provided this is checked. It column is None, then the MS.column is specified. If both are None then an error is raised.

Parameters:

ms (Union[MS, str, PathLike]) – The measurement set to check. Will attempt to cast to Path.
column (Optional[str], optional) – The column to check for. Defaults to None. sub_table (Optional[str], optional): A sub-table of the measurement set to inspect. If None the main table is examined. Defaults to None.

Raises:

ValueError – Raised when both column and ms.column are None.

Returns:

Whether the column exists in the measurement set.

Return type:

bool

flint.ms.cli() → None[source]¶

flint.ms.consistent_channelwise_frequencies(freqs: list[numpy.ndarray] | numpy.ndarray) → numpy.ndarray[source]¶

Given a collection of frequencies in the form of (N, frequencies), inspect the frequencies channelwise to ensure they are all the same.

This does not operate on MSs, just the collection of frequencies

Parameters:: freqs (Union[List[np.ndarray], np.ndarray]) – The collection of frequencies to be inspected
Returns:: Same length as the frequencies. True if for a single channel all frequencies are the same. False otherwise.
Return type:: np.ndarray

flint.ms.consistent_ms(ms1: flint.options.MS, ms2: flint.options.MS) → bool[source]¶

Perform some basic consistency checks to ensure MS1 can be combined with MS2. This is important when considering candidate AO-style calibration solution files.

Parameters:

ms1 (MS) – The first MS to consider
ms2 (MS) – The second MS to consider

Returns:

Whether MS1 is consistent with MS2

Return type:

bool

flint.ms.consistent_ms_frequencies(mss: tuple[flint.options.MS, Ellipsis]) → bool[source]¶

Given a set of measurement sets, inspect the frequencies to ensure they are all the same

See the get_freqs_from_ms function, which is used internally.

Parameters:: mss (Tuple[MS, ...]) – Collection of MSs to inspect the frequencies of
Returns:: Whether all the frequencies and their order are the same
Return type:: bool

flint.ms.copy_and_preprocess_casda_askap_ms(casda_ms: flint.options.MS | pathlib.Path, casa_container: pathlib.Path, data_column: str = 'DATA', instrument_column: str = 'INSTRUMENT_DATA', output_directory: pathlib.Path = Path('./')) → flint.options.MS[source]¶

Convert an ASKAP pipeline MS from CASDA into a FLINT form. This involves making a copy of it, updating its name, and then preprocessing.

Typically these MSs are:

bandpass calibrated
self-calibrated
in the instrument frame
has factor of 2 scaling

This function attempts to craft the MS so that the data column has had visibilities rotated and scaled to make them compatible with certain imaging packages (e.g. wsclean).

Parameters:

casda_ms (Union[MS, Path]) – The measurement set to preprocess
data_column (str, optional) – The column with data to preprocess. Defaults to “DATA”.
instrument_column (str, optional) – The name of the column to be created with data in the instrument frame. Defaults to “INSTRUMENT_DATA”.
fix_stokes_factor (bool, optional) – Whether to scale the visibilities to account for the factor of 2 error. Defaults to True.
output_directory (Path, optional) – The output directory that the preprocessed MS will be placed into. Defaults to Path(“./”).

Returns:

a corrected and preprocessed measurement set

Return type:

flint.ms.critical_ms_interaction(input_ms: pathlib.Path, copy: bool = False, suffix: str = '.critical')[source]¶

A context manager that can be used to register a measurement set as entering a critical segment, i.e. phase rotation. If this stage were to fail, then future operations against said measurement set might be nonsense. This mechanism is intended to make it clear that the measurement set is in a dangerous part of code.

Failure to return the MS to its original name (or rename the copy) highlights this failed stage.

Parameters:

input_ms (Path) – The measurement set to monitor.
copy (bool, optional) – If True, a copy of the MS is made with the suffix supplied. Otherwise, the MS is siomply renamed. Defaults to False.
suffix (str, optional) – Suffix indicating the MS is in the dangerous stage. Defaults to ‘.critical’.

Yields:

Path – Resource location of the measurement set being processed

flint.ms.describe_ms(ms: flint.options.MS | pathlib.Path, verbose: bool = False) → MSSummary[source]¶

Print some basic information from the inpute measurement set.

Parameters:

ms (Union[MS,Path]) – Measurement set to inspect
verbose (bool, optional) – Log MS options to the flint logger. Defaults to False.

Returns:

Brief overview of the MS.

Return type:

MSSummary

flint.ms.find_mss(mss_parent_path: pathlib.Path, expected_ms_count: int | None = 36, data_column: str | None = None, model_column: str | None = None) → tuple[flint.options.MS, Ellipsis][source]¶

Search a directory to find measurement sets via a simple *.ms glob expression. An expected number of MSs can be enforced via the expected_ms_count option.

Parameters:

mss_parent_path (Path) – The parent directory that will be globbed to search for MSs.
expected_ms_count (Optional[int], optional) – The number of MSs that should be there. If None no check is performed. Defaults to 36.
data_column (Optional[str], optional) – Set the column attribute of each MS to this (no checks to ensure it exists). If None use default of MS. Defaults to None.
model_column (Optional[str], optional) – Set the model column attribute of each MS to this (no checks to ensure it exists). If None use default of MS. Defaults to None.

Returns:

Collection of found MSs

Return type:

Tuple[MS, …]

flint.ms.get_beam_from_ms(ms: flint.options.MS | pathlib.Path) → int[source]¶

Lookup the ASKAP beam number from a measurement set.

Parameters:: ms (Union[MS, Path]) – The measurement set to inspect. If MS, the attached path is used.
Returns:: The beam ID number
Return type:: int

flint.ms.get_field_id_for_field(ms: flint.options.MS | pathlib.Path, field_name: str) → int | None[source]¶

Return the FIELD_ID for an elected field in a measurement set

Parameters:

ms (Union[MS, Path]) – The measurement set to inspect
field_name (str) – The desired name of the field to find the FIELD_ID for

Raises:

ValueError – Raised when more than one unique FIELD_ID is found for a single field

Returns:

The FIELD_ID as an int if the field is found. None if it is not found.

Return type:

Union[int, None]

flint.ms.get_freqs_from_ms(ms: flint.options.MS | pathlib.Path) → numpy.ndarray[source]¶

Return the frequencies observed from an ASKAP Measurement set. Some basic checks are performed to ensure they conform to some expectations.

Parameters:: ms (Union[MS, Path]) – Measurement set to inspect
Returns:: A squeeze array of frequencies, in Hertz.
Return type:: np.ndarray

flint.ms.get_parser() → argparse.ArgumentParser[source]¶

flint.ms.get_phase_dir_from_ms(ms: flint.options.MS | pathlib.Path) → astropy.coordinates.SkyCoord[source]¶

Extract the phase direction from a measurement set.

If more than one phase direction is found an AssertError will be raised.

Parameters:: ms (Union[MS, Path]) – The measurement set to get the phase direction from
Returns:: The phase direction the measurement set is directed towards
Return type:: SkyCoord

flint.ms.get_pol_axis_from_ms(ms: flint.options.MS | pathlib.Path, feed_idx: int | None = None, col: str = 'RECEPTOR_ANGLE') → astropy.units.Quantity[source]¶

Get the polarization axis from the ASKAP MS. Checks are performed to ensure this polarisation axis angle is constant throughout the observation.

Parameters:

ms (Union[MS, Path]) – The path to the measurement set that will be inspected
feed_idx (Optional[int], optional) – Specify the entry in the FEED
a (table of ms to return. This might be required when a subset of)
varying (measurement set has been extracted from an observation with a)
orientation.
col (str, optional) – The column to extract the polarization angle from.

Returns:

The rotation of the PAF throughout the observing.

Return type:

astropy.units.Quantity

flint.ms.get_telescope_location_from_ms(ms: flint.options.MS | pathlib.Path) → astropy.coordinates.EarthLocation[source]¶

Return the telescope location from an ASKAP Measurement set.

Parameters:: ms (Union[MS, Path]) – Measurement set to inspect
Returns:: The telescope location
Return type:: EarthLocation

flint.ms.get_times_from_ms(ms: flint.options.MS | pathlib.Path, unique: bool = False, sort: bool = False, return_raw: float = False) → astropy.time.Time | numpy.typing.NDArray[numpy.floating][source]¶

Return the observation times from an ASKAP Measurement set.

Parameters:

ms (Union[MS, Path]) – Measurement set to inspect
unique (bool, optional) – return only the unique times. Defaults to False.
sort (bool, optional) – return times in ascending order, otherwise they are returned in the order the MS has them in. Defaults to False.
rutern_raw (bool, optional) – If True return the times as they are from the MS. Otherwise return as astropy.time.Time. Defaults to False.

Returns:

The observation times. If return_raw is True these are floats, otherwise Time.

Return type:

Time | np.typing.NDArray[np.floating]

flint.ms.preprocess_askap_ms(ms: flint.options.MS | pathlib.Path, data_column: str = 'DATA', instrument_column: str = 'INSTRUMENT_DATA', overwrite: bool = True, skip_rotation: bool = False, fix_stokes_factor: bool = False, apply_ms_transform: bool = False, casa_container: pathlib.Path | None = None, rename: bool = False) → flint.options.MS[source]¶

The ASKAP MS stores its data in a way that is not immediately accessible to other astronomical software, like wsclean or casa. For each measurement set the centre of the field is stored, and beam offsets are stored in a separate table.

Additionally, the correlations stored are more akin to (P, Q) – they are not (X, Y) in the sky reference frame. This function does two things:

updates the positions stored so when data are imaged/calibrated the correlations are directed to the correct position
will apply a rotation to go from (P, Q) -> (X, Y)

These corrections are applied to the original MS, and should be able to be executed multiple times without accumulating changes.

Parameters:

ms (Union[MS, Path]) – The measurement set to update
data_column (str, optional) – The name of the data column to correct. This will first be renamed to the value specified by instrument_column before being corrected. Defaults to ‘DATA’.
instrument_column (str, optional) – The name of the column that will hold the original data_column data. Defaults to ‘INSTRUMENT_DATA’
overwrite (bool, optional) – If the instrument_column and data_column both exist and overwrite=True the data_column will be overwritten. Otherwise, a ValueError is raised. Defaults to True.
skip_rotation (bool, optional) – If true, the visibilities are not rotated Defaults to False.
fix_stokes_factor (bool, optional) – Apply the stokes scaling factor (aruses in different definition of Stokes between Ynadasoft and other applications) when rotation visibilities. This should be set to False is the bandpass solutions have already absorded this scaling term. Defaults to False.
apply_ms_transform (bool, optional) – If True, the MS will be transformed using the casa_container provided. Defaults to False.
casa_container (Path, optional) – The path to the CASA container that will be used to transform the MS. Defaults to None.
rename (bool, optional) – If True, the MS will be renamed to a Flint-processed name. Defaults to False.

Returns:

An updated measurement set with the corrections applied.

Return type:

flint.ms.remove_columns_from_ms(ms: flint.options.MS | pathlib.Path, columns_to_remove: str | list[str]) → list[str][source]¶

Attempt to remove a collection of columns from a measurement set. If any of the provided columns do not exist they are ignored.

Parameters:

ms (Union[MS, Path]) – The measurement set to inspect and remove columns from
columns_to_remove (Union[str, List[str]]) – Collection of column names to remove. If a single column internally it is cast to a list of length 1.

Returns:

Collection of column names removed

Return type:

List[str]

flint.ms.rename_column_in_ms(ms: flint.options.MS, original_column_name: str, new_column_name: str, update_tracked_column: bool = False) → flint.options.MS[source]¶

Rename a column in a measurement set. Optionally update the tracked data column attribute of the input measurement set.

Parameters:

ms (MS) – Measurement set with the column to rename
original_column_name (str) – The name of the column that will be changed
new_column_name (str) – The new name of the column set in original_column_name
update_tracked_column (bool, optional) – Whether the data attribute of ms will be updated to new_column_name. Defaults to False.

Returns:

The measurement set operated on

Return type:

flint.ms.rename_ms_and_columns_for_selfcal(ms: flint.options.MS, target: str | pathlib.Path, corrected_data: str = 'CORRECTED_DATA', data: str = 'DATA') → flint.options.MS[source]¶

Take an existing measurement set, rename it, and appropriately rename the “DATA” and “CORRECTED_DATA” columns to support a new round of imaging and self-calibration.

This could be considered for larger measurement sets where holding multiple copies throughout rounds of self-calibration is not advisable.

Parameters:

ms (MS) – The subject measurement set to rename
target (Union[str, Path]) – The targett path the measurement set will be renamed to. This should not already exist.
corrected_data (str, optional) – The name of the column with the latest calibrated data. This becomes the data column. Defaults to “CORRECTED_DATA”.
data (str, optional) – The name of the column that will be subsequently processed. If it exists it will be removed. Defaults to “DATA”.

Raises:

FileExistsError – Raised if target already exists
FileNotFoundError – Raise if ms.path does not exist

Returns:

Updated MS container with new path and appropriate data column

Return type:

flint.ms.split_by_field(ms: flint.options.MS | pathlib.Path, field: str | None = None, out_dir: pathlib.Path | None = None, column: str | None = None) → list[flint.options.MS][source]¶

Attempt to split an input measurement set up by the unique FIELDs recorded

Parameters:

ms (Union[MS, Path]) – Input measurement sett to split into smaller MSs by field name
field (Optional[str], optional) – Desired field to extract. If None, all are split. Defaults to None.
out_dir (Optional[Path], optional) – Output directory to write the fresh MSs to. If None, write to same directory as
None. (parent MS. Defaults to)
column (Optional[str], optional) – If not None, set the column attribute of the output MS instance to this. Defaults to None.

Returns:

The output MSs split by their field name.

Return type:

List[MS]

flint.ms.subtract_model_from_data_column(ms: flint.options.MS, model_column: str = 'MODEL_DATA', data_column: str | None = None, output_column: str | None = None, update_tracked_column: bool = False, chunk_size: int | None = None) → flint.options.MS[source]¶

Execute a taql query to subtract the MODEL_DATA from a nominated data column. This requires the model_column to already be inserted into the MS. Internally the critical_ms_interaction context manager is used to highlight that the MS is being modified should things fail when subtracting.

Parameters:

ms (MS) – The measurement set instance being considered
model_column (str, optional) – The column with representing the model. Defaults to “MODEL_DATA”.
data_column (Optional[str], optional) – The column where the column will be subtracted. If None it is taken from the column nominated by the input MS instance. Defaults to None.
output_column (Optional[str], optional) – The output column that will be created. If None it defaults to data_column. Defaults to None.
update_tracked_column (bool, optional) – If True, update ms.column to the column with subtracted data. Defaults to False.
chunk_size (int, optional) – The number of rows to manage at a time. If None, taql is used. If an int iteration with getcol/putcol is used. Defaults to None.

Returns:

The updated MS

Return type: