flint.utils =========== .. py:module:: flint.utils .. autoapi-nested-parse:: Collection of functions and tooling intended for general usage. Classes ------- .. autoapisummary:: flint.utils.SlurmInfo Functions --------- .. autoapisummary:: flint.utils._signal_timelimit_handler flint.utils.copy_directory flint.utils.create_directory flint.utils.estimate_image_centre flint.utils.estimate_skycoord_centre flint.utils.flatten_items flint.utils.generate_strict_stub_wcs_header flint.utils.generate_stub_wcs_header flint.utils.get_beam_shape flint.utils.get_job_info flint.utils.get_packaged_resource_path flint.utils.get_pixels_per_beam flint.utils.get_slurm_info flint.utils.hold_then_move_into flint.utils.log_job_environment flint.utils.parse_environment_variables flint.utils.remove_files_folders flint.utils.rsync_copy_directory flint.utils.temporarily_move_into flint.utils.timelimit_on_context flint.utils.zip_folder Module Contents --------------- .. py:class:: SlurmInfo Bases: :py:obj:`NamedTuple` .. py:attribute:: hostname :type: str The hostname of the slurm job .. py:attribute:: job_id :type: str | None :value: None The job ID of the slurm job .. py:attribute:: task_id :type: str | None :value: None The task ID of the slurm job .. py:attribute:: time :type: str | None :value: None The time time the job information was gathered .. py:function:: _signal_timelimit_handler(*args) .. py:function:: copy_directory(input_directory: pathlib.Path, output_directory: pathlib.Path, verify: bool = False, overwrite: bool = False) -> pathlib.Path Copy a directory into a new location. :param input_directory: The source directory to copy :type input_directory: Path :param output_directory: The location of the source directory to copy to :type output_directory: Path :param verify: Attempt to run `rsync` to verify copy worked. Defaults to False. :type verify: bool, optional :param overwrite: Remove the target direcrtory if it exists. Defaults to False. :type overwrite: bool, optional :returns: Location of output directory :rtype: Path .. py:function:: create_directory(directory: pathlib.Path, parents: bool = True) -> pathlib.Path Will attempt to safely create a directory. Should it not exist it will be created. if this creates an exception, which might happen in a multi-process environment, it is reported and discarded. :param directory: Path to directory to create :type directory: Path :param parents: Create parent directories if necessary. Defaults to True. :type parents: bool, optional :returns: The directory created :rtype: Path .. py:function:: estimate_image_centre(image_path: pathlib.Path) -> astropy.coordinates.SkyCoord .. py:function:: estimate_skycoord_centre(sky_positions: astropy.coordinates.SkyCoord, final_frame: str = 'fk5') -> astropy.coordinates.SkyCoord Estimate the centre position of (RA, Dec) positions stored in a input `SkyCoord`. Internally the mean of the cartesian (X,Y,Z) is calculated, which is then transformed back to a sky position,. :param sky_positions: Set of input positions to consider :type sky_positions: SkyCoord :param final_frame: The frame of the returned `SkyCoord` objects set using `.transform_to`. Defaults to fk5. :type final_frame: str, optional :returns: Centre position :rtype: SkyCoord .. py:function:: flatten_items(items: list[Any]) -> list[Any] Recursively flatten a collection (list or tuple) of items. :param items: Items to flatten. Can be arbitrarily nested :type items: list[Any] :returns: Flattened items :rtype: list[Any] .. py:function:: generate_strict_stub_wcs_header(position_at_image_center: astropy.coordinates.SkyCoord, image_shape: tuple[int, int], pixel_scale: astropy.units.Quantity | str, image_shape_is_center: bool = False) -> astropy.wcs.WCS Create a WCS object using some strict quantities. There are no attempts to cast values appropriately, exception being calling `astropy.units.Quantity` on the `pixel_scale` input should it not be a quantity. The supplied `image_size` is used to calculate the center of the image and set the reference pixel value. The output projection is SIN. :param position_at_image_center: The position that will be at the reference pixel :type position_at_image_center: SkyCoord :param image_shape: The size of the image :type image_shape: Tuple[int, int] :param pixel_scale: Size of the square pixels. If `str` passed will be cast to `Quantity`. :type pixel_scale: Union[u.Quantity,str] :param image_shape_is_center: It True the position specified by `image_shape` is the center reference position. if False, `image_shape` is assumed to be the size of the image, and the center is computed from this. Defaults to False. :type image_shape_is_center: bool, optional :raises TypeError: Raised when pixel scale it not a str or astropy.units.Quantity :returns: A WCS header matching the input specs :rtype: WCS .. py:function:: generate_stub_wcs_header(ra: float | astropy.units.Quantity | None = None, dec: float | astropy.units.Quantity | None = None, image_shape: tuple[int, int] | None = None, pixel_scale: astropy.units.Quantity | str | float | None = None, projection: str = 'SIN', base_wcs: pathlib.Path | astropy.wcs.WCS | None = None) -> astropy.wcs.WCS Create a basic WSC header object that can be used to calculate sky positions for an example image. Care should be taken when using this function as it tries to be too smart for its own good. :param ra: The RA at the reference pixel. if a float is provided it is assumed to be in degrees. :type ra: fUnion[loat,u.Quantity] :param dec: The Dec at the reference pizel. if a float is provided it is assumed to be in degrees. :type dec: Union[float,u.Quantity] :param image_shape: Size of the representative image :type image_shape: Tuple[int, int] :param pixel_scale: The size of the square pixels. if a `float` it is assumed to be arcseconds. If `str`, parsing is hangled by `astropy.units.Quantity`. :type pixel_scale: Union[u.Quantity, str, float] :param projection: Project scheme to encode in the header. Defaults to "SIN". :type projection: str, optional :param base_wcs: Overload an existing WCS object with argument properties. If a `Path` the WCS is obtained from the fits file. If `None` WCS is built from arguments. Defaults to None. :type base_wcs: Optional[Union[Path, WCS]], optional :returns: The representative WCS objects :rtype: WCS .. py:function:: get_beam_shape(fits_path: pathlib.Path) -> flint.convol.BeamShape | None Construct and return a beam shape from the fields in a FITS image :param fits_path: FITS image to extract the beam information from :type fits_path: Path :returns: Shape of the beam stored in the FITS image. None is returned if the beam is not found. :rtype: Optional[BeamShape] .. py:function:: get_job_info(mode: str = 'slurm') -> SlurmInfo Get the job information for the supplied mode :param mode: Which mode to poll information for. Defaults to "slurm". :type mode: str, optional :raises ValueError: Raised if the mode is not supported :returns: The specified mode :rtype: Union[SlurmInfo] .. py:function:: get_packaged_resource_path(package: str, filename: str) -> pathlib.Path Load in the path of a package sources. The `package` argument is passed as a though the module is being specified as an import statement: `flint.data.aoflagger`. :param package: The module path to the resources :type package: str :param filename: Filename of the datafile to load :type filename: str :returns: The absolute path to the packaged resource file :rtype: Path .. py:function:: get_pixels_per_beam(fits_path: pathlib.Path) -> float | None Given a image with beam information, return the number of pixels per beam. The beam is taken from the FITS header. This is evaluated for pixels at the reference pixel position. :param fits_path: FITS image to consideer :type fits_path: Path :returns: Number of pixels per beam. If beam is not in header then None is returned. :rtype: Optional[float] .. py:function:: get_slurm_info() -> SlurmInfo Collect key slurm attributes of a job :returns: Collection of slurm items from the job environment :rtype: SlurmInfo .. py:function:: hold_then_move_into(move_directory: pathlib.Path, hold_directory: pathlib.Path | None, delete_hold_on_exit: bool = True, overwrite_if_exists: bool = False, append_uuid: bool = False) -> Generator[pathlib.Path, None, None] Create a temporary directory such that anything within it on the exit of the context manager is copied over to `move_directory`. If `hold_directory` and `move_directory` are the same or `hold_directory` is None, then `move_directory` is immediately returned and no output files are copied or deleted. `move_directory` will be created if it does not exist. If `append_uuid` is `True` then the returned Path will have a folder whose name is based on a UUID. This UUID will automatically be derived and the output directory will be created. Consider using this is `delete_hold_on_exit` is `True`, especially if running in a multi-threaded context and the `hold_directory` is based on an environment variable (e.g. such as on SLURM). :param move_directory: Final directory location to move items into :type move_directory: Path :param hold_directory: Location of directory to temporarily base work from. If None provided `move_directory` is returned and no copying/deleting is performed on exit. Defaults to None. :type hold_directory: Optional[Path], optional :param delete_hold_on_exit: Whether `hold_directory` is deleted on exit of the context. Defaults to True. :type delete_hold_on_exit: bool, optional :param overwrite_if_exists: If a file already exists in the move directory overwrite it with a new copy. Defaults to False. :type overwrite_if_exists: bool, optional :param append_uuid: add a folder whose name is a `uuid` to the returned Path. Defaults to False. :type append_uuid: bool, optional :returns: Path to the temporary folder :rtype: Path :Yields: *Iterator[Path]* -- Path to the temporary folder .. py:function:: log_job_environment() -> SlurmInfo Log components of the slurm environment. Currently only support slurm :returns: Collection of slurm items from the job environment :rtype: SlurmInfo .. py:function:: parse_environment_variables(variable: str | None, default: str | None = None) -> str | None Expand a $VARIABLE environment variable to obtain its value from the environment. The dollar-character is required in its input to be expanded. Expanding an environment variable embedded within a Path-like expression is supported. Each “/”-delimited segment as an env-var if it exists. Splits `variable` on "/", and for each segment: 1. Strips any leading "$". 2. If the remaining name matches an env-var, substitutes its value. 3. If it started with "$" but the env-var is unset, returns `default`. 4. Otherwise leaves the segment literal. Rejoins the segments with "/", preserves a trailing slash if present. Should a $VARIABLE be specified but is unresolved, the `default` value is returned. Some variables can be used to trigger some operation: - $FLINT_UUID: resolves to a hex-formatted UUID :param variable: e.g. "TEST1/$SLURM_TMPDIR" or "$HOME/subdir" or "literal/path" :param default: returned if any "$VAR" lookup fails :returns: Expanded path string, `default` on lookup failure, or None if `variable` is None. .. py:function:: remove_files_folders(*paths_to_remove: pathlib.Path) -> list[pathlib.Path] Will remove a set of paths from the file system. If a Path points to a folder, it will be recursively removed. Otherwise it is simply unlinked. :param paths_to_remove: Set of Paths that will be removed :type paths_to_remove: Path :returns: Set of Paths that were removed :rtype: List[Path] .. py:function:: rsync_copy_directory(target_path: pathlib.Path, out_path: pathlib.Path) -> pathlib.Path A small attempt to rsync a directory from one location to another. This is an attempt to verify a copy was completed successfully. :param target_path: The target directory to copy :type target_path: Path :param out_path: The location to copy the directory to :type out_path: Path :returns: The output path of the new directory. :rtype: Path .. py:function:: temporarily_move_into(subject: pathlib.Path, temporary_directory: pathlib.Path | None = None) -> Generator[pathlib.Path, None, None] Given a file or folder, temporarily copy it into the path specified by `temporary_directory` for the duration of the context manager. Upon exit the original copy, specified by `subject`, is removed and replaced by the copy within `temporary_directory`. `temporary_directory` will be created internally, and an error will be raised if it exists. If `temporary_directory` describes a nested path only the lowest directory is removed. If `temporary_directory` is None the `subject` path is returned and there is no copying and deleting performed. :param subject: The file or folder to temporarily move :type subject: Path :param temporary_directory: The temporary directory to work with. If none the subject path is returned. Defaults to None. :type temporary_directory: Optional[Path], optional :Yields: *Path* -- The path to the temporary object .. py:function:: timelimit_on_context(timelimit_seconds: int | float) -> Generator[None, None, None] Creates a context manager that will raise ``flint.exceptions.TimelimitException`` should the control not leave the ``with`` context within an specified amount of time. .. rubric:: Notes This function **can not** be used if the function calling it is not executing in the main thread, such as the case with ``dask``. The underlying ``signal`` module relies on being in the main thead, otherwise an ``Exception`` is raised. :param timelimit_seconds: The maximum time allowed for the with context to be escaped :type timelimit_seconds: Union[int,float] :raises TimeLimitException: Raised should the maximum timelimit be violated. :Yields: *Generator[None, None, None]* -- A generating function that returns nothing .. py:function:: zip_folder(in_path: pathlib.Path, out_zip: pathlib.Path | None = None, archive_format: str = 'tar') -> pathlib.Path Zip a directory and remove the original. :param in_path: The path that will be zipped up. :type in_path: Path :param out_zip: Name of the output file. A `archive_format` extension will be added by `shutil.make_archive`. Defaults to None. :type out_zip: Path, optional :param archive_format: The format of the archive. See `shutil.make_archive`. Defaults to "tar". :type archive_format: str, optional :returns: the path of the compressed zipped folder :rtype: Path