flint.ms ======== .. py:module:: flint.ms .. autoapi-nested-parse:: 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 ------- .. autoapisummary:: flint.ms.MSSummary Functions --------- .. autoapisummary:: flint.ms.check_column_in_ms flint.ms.cli flint.ms.consistent_channelwise_frequencies flint.ms.consistent_ms flint.ms.consistent_ms_frequencies flint.ms.copy_and_preprocess_casda_askap_ms flint.ms.critical_ms_interaction flint.ms.describe_ms flint.ms.find_mss flint.ms.get_beam_from_ms flint.ms.get_field_id_for_field flint.ms.get_freqs_from_ms flint.ms.get_parser flint.ms.get_phase_dir_from_ms flint.ms.get_pol_axis_from_ms flint.ms.get_telescope_location_from_ms flint.ms.get_times_from_ms flint.ms.preprocess_askap_ms flint.ms.remove_columns_from_ms flint.ms.rename_column_in_ms flint.ms.rename_ms_and_columns_for_selfcal flint.ms.split_by_field flint.ms.subtract_model_from_data_column Module Contents --------------- .. py:class:: MSSummary Bases: :py:obj:`NamedTuple` Small structure to contain overview of a MS .. py:attribute:: ants :type: list[int] Collection of unique antennas .. py:attribute:: beam :type: int The ASKAP beam number of the measurement set .. py:attribute:: fields :type: list[str] Collection of unique field names from the FIELDS table .. py:attribute:: flag_spectrum :type: numpy.ndarray Flagged spectral channels .. py:attribute:: flagged :type: int Number of flagged records .. py:attribute:: path :type: pathlib.Path Path to the measurement set that is being represented .. py:attribute:: phase_dir :type: astropy.coordinates.SkyCoord The phase direction of the measurement set, which will be where the image will be centred .. py:attribute:: spw :type: int | None :value: None Intended to be used with ASKAP high-frequency resolution modes, where the MS is divided into SPWs .. py:attribute:: unflagged :type: int Number of unflagged records .. py:function:: check_column_in_ms(ms: flint.options.MS | str | os.PathLike, column: str | None = None, sub_table: str | None = None) -> bool 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. :param ms: The measurement set to check. Will attempt to cast to Path. :type ms: Union[MS, str, PathLike] :param column: 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. :type column: Optional[str], optional :raises ValueError: Raised when both `column` and `ms.column` are None. :returns: Whether the column exists in the measurement set. :rtype: bool .. py:function:: cli() -> None .. py:function:: consistent_channelwise_frequencies(freqs: list[numpy.ndarray] | numpy.ndarray) -> numpy.ndarray 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 :param freqs: The collection of frequencies to be inspected :type freqs: Union[List[np.ndarray], np.ndarray] :returns: Same length as the frequencies. True if for a single channel all frequencies are the same. False otherwise. :rtype: np.ndarray .. py:function:: consistent_ms(ms1: flint.options.MS, ms2: flint.options.MS) -> bool Perform some basic consistency checks to ensure MS1 can be combined with MS2. This is important when considering candidate AO-style calibration solution files. :param ms1: The first MS to consider :type ms1: MS :param ms2: The second MS to consider :type ms2: MS :returns: Whether MS1 is consistent with MS2 :rtype: bool .. py:function:: consistent_ms_frequencies(mss: tuple[flint.options.MS, Ellipsis]) -> bool 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. :param mss: Collection of MSs to inspect the frequencies of :type mss: Tuple[MS, ...] :returns: Whether all the frequencies and their order are the same :rtype: bool .. py:function:: 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 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). :param casda_ms: The measurement set to preprocess :type casda_ms: Union[MS, Path] :param data_column: The column with data to preprocess. Defaults to "DATA". :type data_column: str, optional :param instrument_column: The name of the column to be created with data in the instrument frame. Defaults to "INSTRUMENT_DATA". :type instrument_column: str, optional :param fix_stokes_factor: Whether to scale the visibilities to account for the factor of 2 error. Defaults to True. :type fix_stokes_factor: bool, optional :param output_directory: The output directory that the preprocessed MS will be placed into. Defaults to Path("./"). :type output_directory: Path, optional :returns: a corrected and preprocessed measurement set :rtype: MS .. py:function:: critical_ms_interaction(input_ms: pathlib.Path, copy: bool = False, suffix: str = '.critical') 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. :param input_ms: The measurement set to monitor. :type input_ms: Path :param copy: If True, a copy of the MS is made with the suffix supplied. Otherwise, the MS is siomply renamed. Defaults to False. :type copy: bool, optional :param suffix: Suffix indicating the MS is in the dangerous stage. Defaults to '.critical'. :type suffix: str, optional :Yields: *Path* -- Resource location of the measurement set being processed .. py:function:: describe_ms(ms: flint.options.MS | pathlib.Path, verbose: bool = False) -> MSSummary Print some basic information from the inpute measurement set. :param ms: Measurement set to inspect :type ms: Union[MS,Path] :param verbose: Log MS options to the flint logger. Defaults to False. :type verbose: bool, optional :returns: Brief overview of the MS. :rtype: MSSummary .. py:function:: 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] 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. :param mss_parent_path: The parent directory that will be globbed to search for MSs. :type mss_parent_path: Path :param expected_ms_count: The number of MSs that should be there. If None no check is performed. Defaults to 36. :type expected_ms_count: Optional[int], optional :param data_column: Set the column attribute of each MS to this (no checks to ensure it exists). If None use default of MS. Defaults to None. :type data_column: Optional[str], optional :param model_column: 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. :type model_column: Optional[str], optional :returns: Collection of found MSs :rtype: Tuple[MS, ...] .. py:function:: get_beam_from_ms(ms: flint.options.MS | pathlib.Path) -> int Lookup the ASKAP beam number from a measurement set. :param ms: The measurement set to inspect. If `MS`, the attached path is used. :type ms: Union[MS, Path] :returns: The beam ID number :rtype: int .. py:function:: get_field_id_for_field(ms: flint.options.MS | pathlib.Path, field_name: str) -> int | None Return the FIELD_ID for an elected field in a measurement set :param ms: The measurement set to inspect :type ms: Union[MS, Path] :param field_name: The desired name of the field to find the FIELD_ID for :type field_name: str :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. :rtype: Union[int, None] .. py:function:: get_freqs_from_ms(ms: flint.options.MS | pathlib.Path) -> numpy.ndarray Return the frequencies observed from an ASKAP Measurement set. Some basic checks are performed to ensure they conform to some expectations. :param ms: Measurement set to inspect :type ms: Union[MS, Path] :returns: A squeeze array of frequencies, in Hertz. :rtype: np.ndarray .. py:function:: get_parser() -> argparse.ArgumentParser .. py:function:: get_phase_dir_from_ms(ms: flint.options.MS | pathlib.Path) -> astropy.coordinates.SkyCoord Extract the phase direction from a measurement set. If more than one phase direction is found an AssertError will be raised. :param ms: The measurement set to get the phase direction from :type ms: Union[MS, Path] :returns: The phase direction the measurement set is directed towards :rtype: SkyCoord .. py:function:: get_pol_axis_from_ms(ms: flint.options.MS | pathlib.Path, feed_idx: int | None = None, col: str = 'RECEPTOR_ANGLE') -> astropy.units.Quantity Get the polarization axis from the ASKAP MS. Checks are performed to ensure this polarisation axis angle is constant throughout the observation. :param ms: The path to the measurement set that will be inspected :type ms: Union[MS, Path] :param feed_idx: Specify the entry in the FEED :type feed_idx: Optional[int], optional :param table of `ms` to return. This might be required when a subset of a: :param measurement set has been extracted from an observation with a varying: :param orientation.: :param col: The column to extract the polarization angle from. :type col: str, optional :returns: The rotation of the PAF throughout the observing. :rtype: astropy.units.Quantity .. py:function:: get_telescope_location_from_ms(ms: flint.options.MS | pathlib.Path) -> astropy.coordinates.EarthLocation Return the telescope location from an ASKAP Measurement set. :param ms: Measurement set to inspect :type ms: Union[MS, Path] :returns: The telescope location :rtype: EarthLocation .. py:function:: 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] Return the observation times from an ASKAP Measurement set. :param ms: Measurement set to inspect :type ms: Union[MS, Path] :param unique: return only the unique times. Defaults to False. :type unique: bool, optional :param sort: return times in ascending order, otherwise they are returned in the order the MS has them in. Defaults to False. :type sort: bool, optional :param rutern_raw: If True return the times as they are from the MS. Otherwise return as astropy.time.Time. Defaults to False. :type rutern_raw: bool, optional :returns: The observation times. If `return_raw` is True these are floats, otherwise `Time`. :rtype: Time | np.typing.NDArray[np.floating] .. py:function:: 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 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. :param ms: The measurement set to update :type ms: Union[MS, Path] :param data_column: 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'. :type data_column: str, optional :param instrument_column: The name of the column that will hold the original `data_column` data. Defaults to 'INSTRUMENT_DATA' :type instrument_column: str, optional :param overwrite: 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. :type overwrite: bool, optional :param skip_rotation: If true, the visibilities are not rotated Defaults to False. :type skip_rotation: bool, optional :param fix_stokes_factor: 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. :type fix_stokes_factor: bool, optional :param apply_ms_transform: If True, the MS will be transformed using the `casa_container` provided. Defaults to False. :type apply_ms_transform: bool, optional :param casa_container: The path to the CASA container that will be used to transform the MS. Defaults to None. :type casa_container: Path, optional :param rename: If True, the MS will be renamed to a Flint-processed name. Defaults to False. :type rename: bool, optional :returns: An updated measurement set with the corrections applied. :rtype: MS .. py:function:: remove_columns_from_ms(ms: flint.options.MS | pathlib.Path, columns_to_remove: str | list[str]) -> list[str] Attempt to remove a collection of columns from a measurement set. If any of the provided columns do not exist they are ignored. :param ms: The measurement set to inspect and remove columns from :type ms: Union[MS, Path] :param columns_to_remove: Collection of column names to remove. If a single column internally it is cast to a list of length 1. :type columns_to_remove: Union[str, List[str]] :returns: Collection of column names removed :rtype: List[str] .. py:function:: rename_column_in_ms(ms: flint.options.MS, original_column_name: str, new_column_name: str, update_tracked_column: bool = False) -> flint.options.MS Rename a column in a measurement set. Optionally update the tracked `data` column attribute of the input measurement set. :param ms: Measurement set with the column to rename :type ms: MS :param original_column_name: The name of the column that will be changed :type original_column_name: str :param new_column_name: The new name of the column set in `original_column_name` :type new_column_name: str :param update_tracked_column: Whether the `data` attribute of `ms` will be updated to `new_column_name`. Defaults to False. :type update_tracked_column: bool, optional :returns: The measurement set operated on :rtype: MS .. py:function:: 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 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. :param ms: The subject measurement set to rename :type ms: MS :param target: The targett path the measurement set will be renamed to. This should not already exist. :type target: Union[str, Path] :param corrected_data: The name of the column with the latest calibrated data. This becomes the `data` column. Defaults to "CORRECTED_DATA". :type corrected_data: str, optional :param data: The name of the column that will be subsequently processed. If it exists it will be removed. Defaults to "DATA". :type data: str, optional :raises FileExistsError: Raised if `target` already exists :raises FileNotFoundError: Raise if `ms.path` does not exist :returns: Updated MS container with new path and appropriate data column :rtype: MS .. py:function:: 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] Attempt to split an input measurement set up by the unique FIELDs recorded :param ms: Input measurement sett to split into smaller MSs by field name :type ms: Union[MS, Path] :param field: Desired field to extract. If None, all are split. Defaults to None. :type field: Optional[str], optional :param out_dir: Output directory to write the fresh MSs to. If None, write to same directory as :type out_dir: Optional[Path], optional :param parent MS. Defaults to None.: :param column: If not None, set the column attribute of the output MS instance to this. Defaults to None. :type column: Optional[str], optional :returns: The output MSs split by their field name. :rtype: List[MS] .. py:function:: 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 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. :param ms: The measurement set instance being considered :type ms: MS :param model_column: The column with representing the model. Defaults to "MODEL_DATA". :type model_column: str, optional :param data_column: 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. :type data_column: Optional[str], optional :param output_column: The output column that will be created. If ``None`` it defaults to ``data_column``. Defaults to None. :type output_column: Optional[str], optional :param update_tracked_column: If True, update ``ms.column`` to the column with subtracted data. Defaults to False. :type update_tracked_column: bool, optional :param chunk_size: 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. :type chunk_size: int, optional :returns: The updated MS :rtype: MS