flint.utils¶

Collection of functions and tooling intended for general usage.

Classes¶

SlurmInfo

Functions¶

`_signal_timelimit_handler`(*args)
`copy_directory`(→ pathlib.Path)	Copy a directory into a new location.
`create_directory`(→ pathlib.Path)	Will attempt to safely create a directory. Should it
`estimate_image_centre`(→ astropy.coordinates.SkyCoord)
`estimate_skycoord_centre`(→ astropy.coordinates.SkyCoord)	Estimate the centre position of (RA, Dec) positions stored in a
`flatten_items`(→ list[Any])	Recursively flatten a collection (list or tuple) of items.
`generate_strict_stub_wcs_header`(→ astropy.wcs.WCS)	Create a WCS object using some strict quantities. There
`generate_stub_wcs_header`(→ astropy.wcs.WCS)	Create a basic WSC header object that can be used to calculate sky positions
`get_beam_shape`(→ flint.convol.BeamShape \| None)	Construct and return a beam shape from the fields in a FITS image
`get_job_info`(→ SlurmInfo)	Get the job information for the supplied mode
`get_packaged_resource_path`(→ pathlib.Path)	Load in the path of a package sources.
`get_pixels_per_beam`(→ float \| None)	Given a image with beam information, return the number of pixels
`get_slurm_info`(→ SlurmInfo)	Collect key slurm attributes of a job
`hold_then_move_into`(→ Generator[pathlib.Path, None, None])	Create a temporary directory such that anything within it on the
`log_job_environment`(→ SlurmInfo)	Log components of the slurm environment. Currently only support slurm
`parse_environment_variables`(→ str \| None)	Expand a $VARIABLE environment variable to obtain its value from
`remove_files_folders`(→ list[pathlib.Path])	Will remove a set of paths from the file system. If a Path points
`rsync_copy_directory`(→ pathlib.Path)	A small attempt to rsync a directory from one location to another.
`temporarily_move_into`(→ Generator[pathlib.Path, None, ...)	Given a file or folder, temporarily copy it into the path specified
`timelimit_on_context`(→ Generator[None, None, None])	Creates a context manager that will raise `flint.exceptions.TimelimitException`
`zip_folder`(→ pathlib.Path)	Zip a directory and remove the original.

Module Contents¶

class flint.utils.SlurmInfo[source]¶

Bases: NamedTuple

hostname: str[source]¶: The hostname of the slurm job

job_id: str | None = None[source]¶: The job ID of the slurm job

task_id: str | None = None[source]¶: The task ID of the slurm job

time: str | None = None[source]¶: The time time the job information was gathered

flint.utils._signal_timelimit_handler(*args)[source]¶

flint.utils.copy_directory(input_directory: pathlib.Path, output_directory: pathlib.Path, verify: bool = False, overwrite: bool = False) → pathlib.Path[source]¶

Copy a directory into a new location.

Parameters:

input_directory (Path) – The source directory to copy
output_directory (Path) – The location of the source directory to copy to
verify (bool, optional) – Attempt to run rsync to verify copy worked. Defaults to False.
overwrite (bool, optional) – Remove the target direcrtory if it exists. Defaults to False.

Returns:

Location of output directory

Return type:

Path

flint.utils.create_directory(directory: pathlib.Path, parents: bool = True) → pathlib.Path[source]¶

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.

Parameters:

directory (Path) – Path to directory to create
parents (bool, optional) – Create parent directories if necessary. Defaults to True.

Returns:

The directory created

Return type:

Path

flint.utils.estimate_image_centre(image_path: pathlib.Path) → astropy.coordinates.SkyCoord[source]¶

flint.utils.estimate_skycoord_centre(sky_positions: astropy.coordinates.SkyCoord, final_frame: str = 'fk5') → astropy.coordinates.SkyCoord[source]¶

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,.

Parameters:

sky_positions (SkyCoord) – Set of input positions to consider
final_frame (str, optional) – The frame of the returned SkyCoord objects set using .transform_to. Defaults to fk5.

Returns:

Centre position

Return type:

SkyCoord

flint.utils.flatten_items(items: list[Any]) → list[Any][source]¶

Recursively flatten a collection (list or tuple) of items.

Parameters:: items (list[Any]) – Items to flatten. Can be arbitrarily nested
Returns:: Flattened items
Return type:: list[Any]

flint.utils.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[source]¶

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.

Parameters:

position_at_image_center (SkyCoord) – The position that will be at the reference pixel
image_shape (Tuple[int, int]) – The size of the image
pixel_scale (Union[u.Quantity,str]) – Size of the square pixels. If str passed will be cast to Quantity.
image_shape_is_center (bool, optional) – 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.

Raises:

TypeError – Raised when pixel scale it not a str or astropy.units.Quantity

Returns:

A WCS header matching the input specs

Return type:

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.

Parameters:

ra (fUnion[loat,u.Quantity]) – The RA at the reference pixel. if a float is provided it is assumed to be in degrees.
dec (Union[float,u.Quantity]) – The Dec at the reference pizel. if a float is provided it is assumed to be in degrees.
image_shape (Tuple[int, int]) – Size of the representative image
pixel_scale (Union[u.Quantity, str, float]) – The size of the square pixels. if a float it is assumed to be arcseconds. If str, parsing is hangled by astropy.units.Quantity.
projection (str, optional) – Project scheme to encode in the header. Defaults to “SIN”.
base_wcs (Optional[Union[Path, WCS]], optional) – 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.

Returns:

The representative WCS objects

Return type:

WCS

flint.utils.get_beam_shape(fits_path: pathlib.Path) → flint.convol.BeamShape | None[source]¶

Construct and return a beam shape from the fields in a FITS image

Parameters:: fits_path (Path) – FITS image to extract the beam information from
Returns:: Shape of the beam stored in the FITS image. None is returned if the beam is not found.
Return type:: Optional[BeamShape]

flint.utils.get_job_info(mode: str = 'slurm') → SlurmInfo[source]¶

Get the job information for the supplied mode

Parameters:: mode (str, optional) – Which mode to poll information for. Defaults to “slurm”.
Raises:: ValueError – Raised if the mode is not supported
Returns:: The specified mode
Return type:: Union[SlurmInfo]

flint.utils.get_packaged_resource_path(package: str, filename: str) → pathlib.Path[source]¶

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.

Parameters:

package (str) – The module path to the resources
filename (str) – Filename of the datafile to load

Returns:

The absolute path to the packaged resource file

Return type:

Path

flint.utils.get_pixels_per_beam(fits_path: pathlib.Path) → float | None[source]¶

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.

Parameters:: fits_path (Path) – FITS image to consideer
Returns:: Number of pixels per beam. If beam is not in header then None is returned.
Return type:: Optional[float]

flint.utils.get_slurm_info() → SlurmInfo[source]¶

Collect key slurm attributes of a job

Returns:: Collection of slurm items from the job environment
Return type:: SlurmInfo

flint.utils.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][source]¶

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).

Parameters:

move_directory (Path) – Final directory location to move items into
hold_directory (Optional[Path], optional) – 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.
delete_hold_on_exit (bool, optional) – Whether hold_directory is deleted on exit of the context. Defaults to True.
overwrite_if_exists (bool, optional) – If a file already exists in the move directory overwrite it with a new copy. Defaults to False.
append_uuid (bool, optional) – add a folder whose name is a uuid to the returned Path. Defaults to False.

Returns:

Path to the temporary folder

Return type:

Path

Yields:

Iterator[Path] – Path to the temporary folder

flint.utils.log_job_environment() → SlurmInfo[source]¶

Log components of the slurm environment. Currently only support slurm

Returns:: Collection of slurm items from the job environment
Return type:: SlurmInfo

flint.utils.parse_environment_variables(variable: str | None, default: str | None = None) → str | None[source]¶

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:

Strips any leading “$”.
If the remaining name matches an env-var, substitutes its value.
If it started with “$” but the env-var is unset, returns default.
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

Parameters:

variable – e.g. “TEST1/$SLURM_TMPDIR” or “$HOME/subdir” or “literal/path”
default – returned if any “$VAR” lookup fails

Returns:

Expanded path string, default on lookup failure, or None if variable is None.

flint.utils.remove_files_folders(*paths_to_remove: pathlib.Path) → list[pathlib.Path][source]¶

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.

Parameters:: paths_to_remove (Path) – Set of Paths that will be removed
Returns:: Set of Paths that were removed
Return type:: List[Path]

flint.utils.rsync_copy_directory(target_path: pathlib.Path, out_path: pathlib.Path) → pathlib.Path[source]¶

A small attempt to rsync a directory from one location to another. This is an attempt to verify a copy was completed successfully.

Parameters:

target_path (Path) – The target directory to copy
out_path (Path) – The location to copy the directory to

Returns:

The output path of the new directory.

Return type:

Path

flint.utils.temporarily_move_into(subject: pathlib.Path, temporary_directory: pathlib.Path | None = None) → Generator[pathlib.Path, None, None][source]¶

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.

Parameters:

subject (Path) – The file or folder to temporarily move
temporary_directory (Optional[Path], optional) – The temporary directory to work with. If none the subject path is returned. Defaults to None.

Yields:

Path – The path to the temporary object

flint.utils.timelimit_on_context(timelimit_seconds: int | float) → Generator[None, None, None][source]¶

Creates a context manager that will raise flint.exceptions.TimelimitException should the control not leave the with context within an specified amount of time.

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.

Parameters:: timelimit_seconds (Union[int,float]) – The maximum time allowed for the with context to be escaped
Raises:: TimeLimitException – Raised should the maximum timelimit be violated.
Yields:: Generator[None, None, None] – A generating function that returns nothing

flint.utils.zip_folder(in_path: pathlib.Path, out_zip: pathlib.Path | None = None, archive_format: str = 'tar') → pathlib.Path[source]¶

Zip a directory and remove the original.

Parameters:

in_path (Path) – The path that will be zipped up.
out_zip (Path, optional) – Name of the output file. A archive_format extension will be added by shutil.make_archive. Defaults to None.
archive_format (str, optional) – The format of the archive. See shutil.make_archive. Defaults to “tar”.

Returns:

the path of the compressed zipped folder

Return type:

Path