flint.naming ============ .. py:module:: flint.naming .. autoapi-nested-parse:: Attempts to centralise components to do with naming of pipeline files and data products. Attributes ---------- .. autoapisummary:: flint.naming.LONG_FIELD_TO_SHORTHAND flint.naming.PathStr flint.naming.ResolutionModes Classes ------- .. autoapisummary:: flint.naming.AegeanNames flint.naming.CASDANameComponents flint.naming.FITSMaskNames flint.naming.LinmosNames flint.naming.ProcessedNameComponents flint.naming.RawNameComponents Functions --------- .. autoapisummary:: flint.naming._rename_linear_to_stokes flint.naming.add_timestamp_to_path flint.naming.casda_ms_format flint.naming.create_aegean_names flint.naming.create_fits_mask_names flint.naming.create_image_cube_name flint.naming.create_imaging_name_prefix flint.naming.create_linmos_base_path flint.naming.create_linmos_names flint.naming.create_ms_name flint.naming.create_name_from_common_fields flint.naming.create_path_from_processed_name_components flint.naming.extract_beam_from_name flint.naming.extract_components_from_name flint.naming.get_aocalibrate_output_path flint.naming.get_beam_resolution_str flint.naming.get_fits_cube_from_paths flint.naming.get_potato_output_base_path flint.naming.get_sbid_from_path flint.naming.get_selfcal_ms_name flint.naming.processed_ms_format flint.naming.raw_ms_format flint.naming.rename_linear_to_stokes flint.naming.split_and_get_images flint.naming.split_images flint.naming.update_beam_resolution_field_in_path Module Contents --------------- .. py:class:: AegeanNames Bases: :py:obj:`NamedTuple` Base names that would be used in various Aegean related tasks .. py:attribute:: bkg_image :type: pathlib.Path Background map computed by BANE .. py:attribute:: comp_cat :type: pathlib.Path Component catalogue produced by the aegean source finder .. py:attribute:: ds9_region :type: pathlib.Path DS9 region overlay file .. py:attribute:: resid_image :type: pathlib.Path Residual map after subtracting component catalogue produced by AeRes .. py:attribute:: rms_image :type: pathlib.Path RMS noise map computed by BANE .. py:class:: CASDANameComponents Bases: :py:obj:`NamedTuple` Container for the components of a CASDA MS. These are really those processed by the ASKAP pipeline .. py:attribute:: alias :type: str | None :value: 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. .. py:attribute:: beam :type: str Beam number of the data .. py:attribute:: field :type: str The name of the field extracted .. py:attribute:: format :type: str :value: 'science' What the format / type of the data the MS is. .. py:attribute:: sbid :type: int The sbid of the observation .. py:attribute:: spw :type: str | None :value: None If multiple MS were written as the data were in a high-frequency resolution mode, which segment .. py:class:: FITSMaskNames Bases: :py:obj:`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. .. py:attribute:: mask_fits :type: pathlib.Path Name of the mask FITS file .. py:attribute:: scale_mask_fits :type: pathlib.Path | None :value: None Path to a FITS file that describes per-scale clean masks. Scales are represented with a bit-mapped values, .. py:attribute:: signal_fits :type: pathlib.Path | None :value: None Name of the signal FITS file .. py:class:: LinmosNames Bases: :py:obj:`NamedTuple` Creates expected output names for the yandasoft linmos task. .. py:attribute:: image_fits :type: pathlib.Path Path to the final fits co-added image .. py:attribute:: parset_output_path :type: pathlib.Path Path to the output parset generated .. py:attribute:: weight_fits :type: pathlib.Path Path to the weights fits image created when co-adding images .. py:class:: ProcessedNameComponents Bases: :py:obj:`NamedTuple` Container for a file name derived from a MS flint name. Generally of the form: SB.Field.Beam.Spw .. py:attribute:: beam :type: str | None :value: None The beam of the observation processed .. py:attribute:: channel_range :type: tuple[int, int] | None :value: 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. .. py:attribute:: field :type: str The name of the field extracted .. py:attribute:: pol :type: str | None :value: 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). .. py:attribute:: round :type: str | None :value: None The self-calibration round detected. This might be represented as 'noselfcal' in some image products, e.g. linmos. .. py:attribute:: sbid :type: str The sbid of the observation .. py:attribute:: scan_range :type: tuple[int, int] | None :value: 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. .. py:attribute:: spw :type: str | None :value: None The SPW of the observation. If there is only one spw this is None. .. py:class:: RawNameComponents Bases: :py:obj:`NamedTuple` .. py:attribute:: beam :type: str Beam number of the data .. py:attribute:: date :type: str Date that the data were taken, of the form YYYY-MM-DD .. py:attribute:: spw :type: str | None :value: None If multiple MS were written as the data were in a high-frequency resolution mode, which segment .. py:attribute:: time :type: str Time that the data were written .. py:function:: _rename_linear_to_stokes(linear_name_str: str, stokes: str) -> str .. py:function:: 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. :param input_path: Path that should have a timestamp added :type input_path: Union[Path, str] :param 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 :rtype: Path .. py:function:: 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` :param in_name: The path to or name of the MS to consider :type in_name: Union[str, Path] :returns: The returned components of the MS. If this fails a `None` is returned. :rtype: Union[CASDANameComponents, None] .. py:function:: create_aegean_names(base_output: str) -> AegeanNames Create the expected names for aegean and its tools. :param base_output: The base name that aegean outputs are built from. :type base_output: str :returns: A collection of names to be produced by Aegean related tasks :rtype: AegeanNames .. py:function:: 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 :param fits_image: Base name of the output files :type fits_image: Union[str,]Path :param include_signal_path: If True, also include ``signal_fits`` in the output. Defaults to False. :type include_signal_path: bool, optional :returns: collection of names used for the signal and mask FITS images :rtype: FITSMaskNames .. py:function:: 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. :param image_prefix: The unique path of the name. Generally this is the common part among the input planes :type image_prefix: Path :param mode: Additional mode/s to add to the file name. Defaults to None. :type mode: Optional[Union[str, List[str]]], optional :param suffix: Additional suffix/s to add before the final 'cube.fits'. Defaults to None. :type suffix: Optional[Union[str, List[str]]], optional :returns: The final path and file name :rtype: Path .. py:function:: 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 :param ms: The measurement set being considered :type ms: Union[MS,Path] :param pol: Whether a polarsation is being considered. Defaults to None. :type pol: Optional[str], optional :param channel_range: The channel range that is going to be imaged. Defaults to none. :type channel_range: Optional[Tuple[int,int]], optional :param scan_range: The scan range that is going to be imaged. Defaults to none. :type scan_range: Optional[Tuple[int,int]], optional :returns: The constructed string name :rtype: str .. py:function:: 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. :param input_images: If provided the common fields of the input images are used as basis of the path. Defaults to None. :type input_images: list[Path] | None, optional :param additional_suffixes: Any additional suffixes to append. Defaults to None. :type additional_suffixes: str | None, optional :returns: The full path of the parset file :rtype: Path .. py:function:: 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. :param name_prefix: The prefix of the filename that will be used to create the linmos and weight file names. :type name_prefix: str | Path :returns: Collection of expected filenames :rtype: LinmosNames .. py:function:: 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. :param ms_path: The measurement set being considered. A RawNameComponents will be constructed against it. :type ms_path: Path :param sbid: 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. :type sbid: Optional[int], optional :param field: The field that this measurement set will contain. Defaults to None. :type field: Optional[str], optional :returns: The filename of the measurement set :rtype: str .. py:function:: 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 :param in_paths: Collection of input paths to consider :type in_paths: Tuple[Path, ...] :param additional_suffixes: Add an additional set of suffixes before returning. Defaults to None. :type additional_suffixes: Optional[str], optional :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 :rtype: Path .. py:function:: 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 :param processed_name_components: The naming specification to create :type processed_name_components: ProcessedNameComponents :param parent_path: The parent directory of the output path. Defaults to None. :type parent_path: Path | None, optional :returns: A directory with following the specification of the input ProcessedNameComponents :rtype: Path .. py:function:: 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. :param name: The name to examine to search for the beam number. :type name: Union[str,Path] :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 :rtype: int .. py:function:: 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. :param name: The name to examine to search for the beam number. :type name: Union[str,Path] :raises ValueError: Raised if the name was not was not successfully matched against the known format :returns: The extracted name components within a name :rtype: Union[RawNameComponents,ProcessedNameComponents,CASDANamedComponents] .. py:function:: 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. :param ms_path: Path of the measurement set that the aocalibrate file will be created for :type ms_path: Path :param include_preflagger: Whether to include the ``.preflagged`` term to indicate that the preflagger has been executed :type include_preflagger: bool :param include_smoother: Whether to include the ``.smoothed`` term to included that bandpas smoothing has been performed :type include_smoother: bool :returns: The constructed output path of the solutions file. This include the parent directory of the input measurement set :rtype: Path .. py:function:: 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' :param mode: The mode of image resolution to use. :type mode: Literal["fixed","optimal"] :param marker: Append the marker to the end of the returned mode string. If None mode string is returned. Defaults to None. :type marker: Optional[str], optional :raises ValueError: Raised when an unrecognised mode is supplied :returns: The appropriate string for mapped mode :rtype: str .. py:function:: 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. :param paths: The set of paths to examine to identify potential cube fits images from :type paths: List[Path] :returns: Set of paths matching the search criteria :rtype: List[Path] .. py:function:: get_potato_output_base_path(ms_path: pathlib.Path) -> pathlib.Path Return the base name for potato peel. :param ms_path: Input measurement set that follows the FLINT style process name format :type ms_path: Path :returns: Output base name to use :rtype: Path .. py:function:: 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. :param path: The path that contains the sbid to extract. :type path: Path :raises ValueError: Raised when the SBID extracted from the path is not all digits :returns: The sbid extracted :rtype: int .. py:function:: 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. :param in_ms_path: The measurement set that will go through self-calibration :type in_ms_path: Path :param round: The self-calibration round number that is currently being used. Defaults to 1. :type round: int, optional :returns: Output measurement set path to use :rtype: Path .. py:function:: 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. :param in_name: The name that needs to be broken down into components :type in_name: Union[str, Path] :returns: A structure container the sbid, field, beam and spw. None is returned if can not be parsed. :rtype: Union[FormattedNameComponents,None' .. py:function:: 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. :param in_name: The name of a file, presumably a measurement set. The left-most part will be examined for to identify the raw ASKAP naming scheme. :type in_name: str :returns: None if the raw ASKAP measurement set naming scheme was not detected, otherwise a dictionary representing its parts. :rtype: Union[None,Dict[str,str]] .. py:function:: rename_linear_to_stokes(linear_name: PathStr, stokes: str) -> PathStr .. py:function:: 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. :param images: The images to split :type images: list[Path] :param get: The field to extract from the split images :type get: str :param by: How to split the images. Defaults to "pol". :type by: str, optional :raises ValueError: If the field to extract is not found in the split images :returns: The images that match the field of interest :rtype: list[Path] .. py:function:: 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. :param images: The images to split :type images: List[Path] :param by: The field to split the images by. Defaults to "pol". :type by: str, optional :returns: A dictionary of the images split by the field :rtype: Dict[str,List[Path]] .. py:function:: 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`` :param path: The path to inspect and update :type path: Path :param original_mode: The original mode :type original_mode: ResolutionModes :param updated_mode: The mode to move to :type updated_mode: ResolutionModes :param marker: The marker to separate the field. Defaults to None. :type marker: str | None, optional :returns: Updated path :rtype: Path .. py:data:: LONG_FIELD_TO_SHORTHAND Name mapping between the longform of ProcessedFieldComponents and shorthands used .. py:data:: PathStr .. py:data:: ResolutionModes