karabo.util package

Submodules

karabo.util.config_util module

karabo.util.dask module

Module for dask-related functionality.

class DaskHandler

Bases: DaskHandlerBasic

Public & dev API for dask associated functionality.

This is the public dask-api for Karabo, where you don’t have to worry which dask-handler of this module to use. You can do almost everything through this class. The only exception is, if you want to adjust the default settings on a slurm-system (customization through DaskHandlerSlurm).

Attributes

dask_client:

The Dask client object. If None, a new client will be created.

memory_limit:

The memory_limit per worker in GB. If None, the memory limit will be set to the maximum available memory on the node (see documentation) in dask for memory_limit.

n_threads_per_worker:

The number of threads to use per worker. Standard is None, which means that the number of threads will be equal to the number of cores.

use_dask:

Whether to use Dask or not. If None, then Karabo will decide whether to use dask or not for certain tasks.

use_processes:

Use processes instead of threads? Threads:

Fast to initiate.

No need to transfer data to them.

Limited by the GIL, which allows one thread to read the code at once.

Processes:

Take time to set up.
Slow to transfer data to.
Each have their own GIL and so don’t need to take turns reading the code.

classmethod get_dask_client() → Client

Get (create if not exists) a dask-client.

Returns:: Dask-client.

classmethod parallelize_with_dask(iterate_function: Callable[[...], Any], iterable: Iterable[Any], *args: Any, **kwargs: Any) → Any | Tuple[Any, ...] | List[Any]

Run a function over an iterable in parallel using dask, and gather the results.

args & kwargs will get passed to Delayed.

Args:

iterate_function: The function to be applied to each element of iterable.: The function takes the current element of the iterable as its first argument, followed by any positional arguments, and then any keyword arguments.
iterable: The iterable over which the function will be applied. Each element of iterable will be passed to iterate_function.

Returns: A tuple containing the results of the iterate_function for each

element in the iterable. The results are gathered using dask’s compute function.

classmethod setup() → None: Calls get_dask_client.

classmethod should_dask_be_used(override: bool | None = None) → bool

Util function to decide whether dask should be used or not.

Args:: override: Override? Has highest priority.
Returns:: Decision whether dask should be used or not.

class DaskHandlerBasic

Bases: object

Base-class for dask-handler functionality.

Attributes

dask_client:

The Dask client object. If None, a new client will be created.

memory_limit:

The memory_limit per worker in GB. If None, the memory limit will be set to the maximum available memory on the node (see documentation) in dask for memory_limit.

n_threads_per_worker:

The number of threads to use per worker. Standard is None, which means that the number of threads will be equal to the number of cores.

use_dask:

Whether to use Dask or not. If None, then Karabo will decide whether to use dask or not for certain tasks.

use_processes:

Use processes instead of threads? Threads:

Fast to initiate.

No need to transfer data to them.

Limited by the GIL, which allows one thread to read the code at once.

Processes:

Take time to set up.
Slow to transfer data to.
Each have their own GIL and so don’t need to take turns reading the code.

dask_client: Client | None = None

classmethod get_dask_client() → Client

Get (create if not exists) a dask-client.

Returns:: Dask-client.

memory_limit: float | None = None

n_threads_per_worker: int | None = None

classmethod parallelize_with_dask(iterate_function: Callable[[...], Any], iterable: Iterable[Any], *args: Any, **kwargs: Any) → Any | Tuple[Any, ...] | List[Any]

Run a function over an iterable in parallel using dask, and gather the results.

args & kwargs will get passed to Delayed.

Args:

iterate_function: The function to be applied to each element of iterable.: The function takes the current element of the iterable as its first argument, followed by any positional arguments, and then any keyword arguments.
iterable: The iterable over which the function will be applied. Each element of iterable will be passed to iterate_function.

Returns: A tuple containing the results of the iterate_function for each

element in the iterable. The results are gathered using dask’s compute function.

classmethod setup() → None: Calls get_dask_client.

classmethod should_dask_be_used(override: bool | None = None) → bool

Util function to decide whether dask should be used or not.

Args:: override: Override? Has highest priority.
Returns:: Decision whether dask should be used or not.

use_dask: bool | None = None

use_processes: bool = False

class DaskHandlerSlurm

Bases: DaskHandlerBasic

Dask-handler for slurm-based jobs.

Attributes

use_workers_or_nannies:: Whether to use workers or nannies (default). This could lead to more processing (see documentation for dask usage in Karabo).
n_workers_scheduler_nodeint: The number of workers to start on the scheduler node.
timeout: int: Timeout in seconds for the dask-scheduler to wait for all the workers to connect.

classmethod get_dask_client() → Client

Get (create if not exists) a dask-client for a SLURM environment.

Returns:: Dask-client.

classmethod get_node_id() → int

Gets the current node-id.

Returns:: Node-id.

classmethod get_node_name() → str

Gets the current node-name.

Returns:: Node-name.

classmethod get_number_of_nodes() → int

Gets the number of nodes of the slurm-job.

Returns:: Number of nodes.

classmethod is_first_node() → bool

Util function to check if current-node is fist-node.

Returns:: Check-result.

classmethod is_on_slurm_cluster() → bool

Util function to check if code is running in a slurm-job.

Returns:: Check-result.

n_workers_scheduler_node: int = 1

classmethod should_dask_be_used(override: bool | None = None) → bool

Util function to decide whether dask should be used or not.

This implementation differs a bit from the basic-class, where: on SLURM-systems, additional checks are taken into consideration.
Args:: override: Override? Has highest priority.
Returns:: Decision whether dask should be used or not.

timeout: int = 60

use_workers_or_nannies: Literal['workers', 'nannies'] = 'nannies'

karabo.util.data_util module

Gauss(x: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], x0: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], y0: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], a: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], sigma: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]]) → int64 | float64 | ndarray[Any, dtype[int64 | float64]]

Voigt(x: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], x0: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], y0: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], a: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], sigma: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]], gamma: bool | int | integer | float | floating | ndarray[Any, dtype[int64 | float64]]) → int64 | float64 | ndarray[Any, dtype[int64 | float64]]

calculate_chunk_size_from_max_chunk_size_in_memory(max_chunk_memory_size: str, data_array: DataArray | List[DataArray]) → int

calculate_required_number_of_chunks(max_chunk_size_in_memory: str, data_array: List[DataArray]) → int

extract_chars_from_string(string: str) → str

extract_digit_from_string(string: str) → int

full_getter(self: object) → Dict[str, Any]

full_setter(self: object, state: Dict[str, Any]) → None

get_module_absolute_path() → str

get_module_path_of_module(module: ModuleType) → str

get_spectral_sky_data(ra: ndarray[Any, dtype[float64]], dec: ndarray[Any, dtype[float64]], freq0: ndarray[Any, dtype[float64]], nfreq: int) → ndarray[Any, dtype[float64]]

input_wrapper(msg: str, ret: str = 'y') → str

Wrapper of standard input to define what return ret it will get during Unit-tests, since the test just stops otherwise. The environment variable ‘SKIP_INPUT’ or ‘UNIT_TEST’ must be set with an arbitrary value to return ret.

Parameters:

msg – input message
ret – return value if ‘SKIP_INPUT’ or ‘UNIT_TEST’ is set, default=’y’

parse_size(size_str: str) → int

read_CSV_to_ndarray(file: str) → ndarray[Any, dtype[float64]]

resample_spectral_lines(npoints: int, dfreq: ndarray[Any, dtype[float64]], spec_line: ndarray[Any, dtype[float64]]) → Tuple[ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]

karabo.util.file_handler module

class FileHandler

Bases: object

Utility file-handler for unspecified directories.

Provides cache-management functionality. FileHandler.root_stm (short-term-memory-dir) and FileHandler.root_ltm (long-term-memory-dir) are static root-directories where each according cache-dir is located. In case someone wants to extract something specific from the cache, the path is usually printed blue & bold in stdout.

Honors ‘TMPDIR’ & ‘TMP’ and ‘SCRATCH’ env-var(s) for STM-disk-cache where ‘TMPDIR’ = ‘TMP’ > ‘SCRATCH’ > /tmp Honors ‘XDG_CACHE_HOME’ env-var(s) for LTM-disk-cache where ‘XDG_CACHE_HOME’ > $HOME/.cache > /tmp Note: Setting env-vars has only an effect if they’re set before importing Karabo. Run-time adjustments must be done directly on root_stm and root_ltm!

The root STM and LTM must be unique per user (seeded rnd chars+digits) to avoid conflicting dir-names on any computer with any root-directory.

LTM-root └── karabo-LTM-<user>-<10 rnd chars+digits>

├── <prefix><10 rnd chars+digits> | ├── <subdir> | └── <file> └── <prefix><10 rnd chars+digits>

├── <subdir> └── <file>

STM-root └── karabo-STM-<user>-<10 rnd chars+digits>

├── <prefix><10 rnd chars+digits> | ├── <subdir> | └── <file> └── <prefix><10 rnd chars+digits>

├── <subdir> └── <file>

FileHandler can be used the same way as tempfile.TemporaryDirectory using with.

classmethod clean(term: Literal['long', 'short'] = 'short') → None

Removes the entire directory specified by term.

We strongly suggest to NOT use this function in a workflow. This function removed the entire karabo-disk-cache. So if there’s another karabo-process running in parallel, you could mess with their disk-cache as well.

Args:: term: “long” or “short” term memory

clean_instance() → None: Cleans instance-bound tmp-dirs of self.tmps from disk.

classmethod empty_dir(dir_path: Path | str) → None

Deletes all contents of dir_path, but not the directory itself.

This function assumes that all files and directories are owned by the function-user.

Args:: dir_path: Directory to empty.

Gets a tmp-dir path.

This is the to-go function to get a tmp-dir in the according directory.

Args:

prefix: Dir-name prefix for STM (optional) and dir-name for LTM (required). term: “short” for STM or “long” for LTM. purpose: Creates a verbose print-msg with it’s purpose if set. unique: If an object which has attributes is provided, then you get

the same tmp-dir for the unique instance.

mkdir: Make-dir directly? seed: Seed rnd chars+digits of a STM sub-dir for relocation

purpose of different processes? Shouldn’t be used for LTM sub-dirs, unless you know what you’re doing. LTM sub-dirs are already seeded with prefix. However, if they are seeded for some reason, the seed is then something like prefix + seed, which leads to different LTM sub-dirs.

Returns:

tmp-dir path

classmethod is_dir_empty(dirname: Path | str) → bool

Checks if dirname is empty assuming dirname exists.

Args:: dirname: Directory to check.
Raises:: NotADirectoryError: If dirname is not an existing directory.
Returns:: True if dir is empty, else False

classmethod ltm() → str: LTM (long-term-memory) path.

classmethod remove_empty_dirs(term: Literal['long', 'short'] = 'short') → None

Removes empty directories in the chosen cache-dir.

Args:: term: “long” or “short” term memory

root_ltm: str = '/home/runner/.cache'

root_stm: str = '/tmp'

classmethod stm() → str: STM (short-term-memory) path.

assert_valid_ending(path: str | Path, ending: str) → None

Utility function to check if the ending of path is ending.

Args:: path: Path to check. ending: Ending match.
Raises:: ValueError: When the ending of path doesn’t match ending.

getsize(inode: str | Path) → int

Gets the total size of a file or directory in number of bytes.

Args:: inode: Directory or file to get size from. Can take a while for a large dir.
Returns:: Number of bytes of inode.

write_dir(dir: TDirPathType, *, overwrite: bool = False) → Generator[TDirPathType, None, None]

Enables transactional creating and writing into dir.

Assumes that dir is empty, meaning NOT partially filled with anything.

This function is NOT thread-safe!

Args:: dir: Directory to create & write. overwrite: Allow overwrite? Be aware that it will replace the entire

dir if it already exists. So be careful to provide the correct dir.
Yields:: Directory to fill safely.

karabo.util.gpu_util module

get_gpu_memory() → int

Retrieves the available GPU memory in MiB by invoking nvidia-smi.

Returns:: int: Available GPU memory in MiB.
Raises:: RuntimeError: If unexpected output is encountered when running nvidia-smi.

is_cuda_available() → bool

Checks if CUDA-compatible GPU is available on the system by invoking nvidia-smi.

Returns:: bool: True if a CUDA-compatible GPU is found, otherwise False.

karabo.util.hdf5_util module

convert_healpix_2_radec(arr: ndarray[Any, dtype[Any]]) → Tuple[float64, float64, int]: Convert array from healpix to 2-D array of RADEC :param arr: :return: RADEC in degrees

get_healpix_image(hdffile: Any) → Any: Get index maps, maps and frequency from HDF5 file

get_vis_from_hdf5(hdffile: Any) → Any: Get index maps, maps and frequency from HDF5 file

h5_diter(g: Dict[str, Dataset | Group], prefix: str = '') → Generator[Tuple[str, Dataset], Any, Any]: Get the data elements from the hdf5 datasets and groups Input: HDF5 file Output: Items and its path of data elements

print_hd5_object_and_keys(hdffile: Any) → Tuple[File, KeysViewHDF5]: Read HDF5 file Returns: HDF Object, relavent keys

karabo.util.helpers module

Module for helper utils which doesn’t belong to any other modules.

get_rnd_str(k: int, seed: str | int | float | bytes | None = None) → str

Creates a random ascii+digits string with length=`k`.

Most tmp-file tools are using a string-length of 10.

Args:: k: Length of random string. seed: Seed.
Returns:: Random generated string.

karabo.util.jupyter module

isNotebook() → bool

karabo.util.math_util module

cartesian_to_ll(x: bool | int | float, y: bool | int | float, z: int = 0) → Tuple[float, float]

get_poisson_disk_sky(min_size: Tuple[bool | int | float, bool | int | float], max_size: Tuple[bool | int | float, bool | int | float], flux_min: bool | int | float, flux_max: bool | int | float, r: int = 10) → ndarray[Any, dtype[float64]]

long_lat_to_cartesian(lat: bool | int | integer | float | floating, lon: bool | int | integer | float | floating) → ndarray[Any, dtype[float64]]

poisson_disc_samples(width: bool | int | float, height: bool | int | float, r: int, k: int = 5, ord: None | float | Literal['fro', 'nuc'] = None) → List[Tuple[float, float]]

karabo.util.plotting_util module

class Font

Bases: object

BLUE = '\x1b[94m'

BOLD = '\x1b[1m'

CYAN = '\x1b[96m'

DARKCYAN = '\x1b[36m'

END = '\x1b[0m'

GREEN = '\x1b[92m'

PURPLE = '\x1b[95m'

RED = '\x1b[91m'

UNDERLINE = '\x1b[4m'

YELLOW = '\x1b[93m'

get_slices(wcs: WCS) → List[str]

karabo.util.setup_pkg module

This .py file is ONLY for setup-specific util-functions. Thus, ONLY deps at building-time are allowed here. If you don’t know what that means, don’t touch anything here.

set_rascil_data_directory_env() → None

Sets specific environment variables that the jupyter kernel is not loading by default.

This function is idempotent (running it more than once brings no side effects).

karabo.util.survey module

This module is to create according survey-files for Karabo.

create_MALS_survey_as_fits(directory: Path | str, version: int = 3, check_for_updates: bool = False, verbose: bool = True) → str

Creates MALS (https://mals.iucaa.in/) survey as a .fits.gz file.

It takes care of downloading the file from ‘https://mals.iucaa.in/catalogue/catalogue_malsdr1v{0}_all.csv’, and convert it into a .fits.gz file. Downloading the .csv catalogue may take a while, because it is about 3.6GB large and downloading-speed depends on some uncontrollable factors. However, if you already have the .csv catalogue, just put it into directory to take it from the disk-cache. All file-products (.csv & .fits.gz) are saved in and loaded from directory.

This is just a utility function and is not meant to be embedded in any library-code.

In case the .fits.gz file already exists, it just returns the the file-path without doing anything like downloading or creating any file.

Args:: directory: Directory to save and load the according catalogue files. version: Survey version. check_for_updates: Also check for new updates? verbose: Verbose?
Returns:: .fits.gz file-path.

karabo.util.testing module

class ChangeWorkingDir

Bases: object

Changes temporarily working dir for test-discovery.

run_tests(pytest_args: str | None = None) → None

Launches pytest.

Args:: args: pytest cli-args, e.g. “-k test_my_favorite”

karabo.util package

Submodules

karabo.util.config_util module

karabo.util.dask module

Attributes

Attributes

Attributes

karabo.util.data_util module

karabo.util.file_handler module

karabo.util.gpu_util module

karabo.util.hdf5_util module

karabo.util.helpers module

karabo.util.jupyter module

karabo.util.math_util module

karabo.util.plotting_util module

karabo.util.setup_pkg module

karabo.util.survey module

karabo.util.testing module

Module contents