unravel.core.utils module#

Utility functions and decorators for handling configurations, processing files and directories, and enhancing command-line scripts with progress bars and detailed function execution info.

Classes:

CustomMofNCompleteColumn: Progress bar column for completed/total items.
CustomTimeElapsedColumn: Progress bar column for elapsed time in green.
CustomTimeRemainingColumn: Progress bar column for remaining time in dark orange.
AverageTimePerIterationColumn: Progress bar column for average time per iteration.

Functions:

load_config: Load settings from a config file.
get_samples: Get a list of sample directories based on provided parameters.
initialize_progress_bar: Initialize a Rich progress bar.
verbose_start_msg: Print the start command and time if verbose mode is enabled.
verbose_end_msg: Print the end time if verbose mode is enabled.
log_command: Decorator to log the command and execution times to a hidden file.
print_func_name_args_times: Decorator to print function execution details.
load_text_from_file: Load text content from a file.
copy_files: Copy specified files from source to target directory.
process_files_with_glob: Process files matching a glob pattern using a processing function.

Usage:

Import the functions and decorators to enhance your scripts.

Examples

>>> # Import the functions and decorators
>>> from unravel.core.utils import load_config, get_samples, initialize_progress_bar, print_func_name_args_times, load_text_from_file, copy_files

>>> # Load the configuration from a file
>>> config = load_config("path/to/config.ini")

>>> # Get a list of sample directories
>>> samples = get_samples(["path/to/dir1", "path/to/dir2"], dir_pattern="sample??", verbose=True)

>>> # Initialize a progress bar
>>> progress, task_id = initialize_progress_bar(len(samples), task_message="[red]Processing samples...")

unravel.core.utils.load_config(config_path)[source]#: Load settings from the config file and return a Config object.

unravel.core.utils.get_samples(dir_list=None, dir_pattern='sample??', verbose=False)[source]#

Finds and returns paths to directories matching a specified pattern within given directories or, if none are provided, the current working directory.

Parameters:

dir_list (list of Path or str, or Path or str, optional) – A list of paths (as Path objects or strings) to sample?? directories or directories that may contain subdirectories matching the dir_pattern.
dir_pattern (str, optional) – A pattern to match directory names, default is “sample??”, where “?” is a wildcard matching a single character. This pattern is used to identify directories of interest.
dir_pattern – A Unix shell-style wildcard pattern used by fnmatch to match directory names. Default is “sample??”, where each “?” matches a single character.
verbose (bool, optional) – If True, prints the found directories, grouped by their parent directories. Default is False.

Returns:

samples – A list of resolved Path objects pointing to directories that match the dir_pattern.

Return type:

list of Path

Notes

If no directories are provided via dir_list, the function searches the current working directory.
If a directory (e.g., the current dir) matches the dir_pattern, it is included in the results and not searched for subdirectories.

Examples

>>> sample_paths = get_samples()  # Search the current working directory for sample?? directories
>>> sample_paths = get_samples([path1, path2], dir_pattern="sample???")  # Search path1 and path2 for sample??? directories

class unravel.core.utils.CustomMofNCompleteColumn(separator='/', table_column=None)[source]#

Bases: MofNCompleteColumn

Parameters:

separator (str) –
table_column (Column | None) –

render(task)[source]#

Show completed/total.

Return type:: Text

class unravel.core.utils.CustomTimeElapsedColumn(table_column=None)[source]#

Bases: TimeElapsedColumn

Parameters:: table_column (Column | None) –

render(task)[source]#

Show time elapsed.

Return type:: Text

class unravel.core.utils.CustomTimeRemainingColumn(compact=False, elapsed_when_finished=False, table_column=None)[source]#

Bases: TimeRemainingColumn

Parameters:

compact (bool) –
elapsed_when_finished (bool) –
table_column (Column | None) –

render(task)[source]#

Show time remaining.

Return type:: Text

class unravel.core.utils.AverageTimePerIterationColumn(table_column=None)[source]#

Bases: ProgressColumn

Parameters:: table_column (Column | None) –

render(task)[source]#

Render the average time per iteration.

Parameters:: task – An object representing a task, which should have a speed attribute.
Returns:: A Text object displaying the average time per iteration.
Return type:: Text

unravel.core.utils.initialize_progress_bar(num_of_items_to_iterate, task_message='[red]Processing...')[source]#

unravel.core.utils.verbose_start_msg()[source]#: Print the start command and time if verbose mode is enabled.

unravel.core.utils.verbose_end_msg()[source]#: Print the end time if verbose mode is enabled.

unravel.core.utils.log_command(func)[source]#: A decorator for main() to log the command and execution times to a hidden file (.command_log.txt).

unravel.core.utils.print_func_name_args_times(print_dir=True)[source]#: A decorator that prints the function name, arguments, duration, and memory usage of the function it decorates.

unravel.core.utils.load_text_from_file(file_path)[source]#

unravel.core.utils.copy_files(source_dir, target_dir, filename, sample_path=None, verbose=False)[source]#

Copy the specified slices to the target directory.

Parameters:

source_dir (-) – Path to the source directory containing the .tif files.
target_dir (-) – Path to the target directory where the selected slices will be copied.
filename (-) – Name of the file to copy.
sample_path (-) – Path to the sample directory (provide to prepend to the filename).
verbose (-) – Increase verbosity.

unravel.core.utils.match_files(patterns, base_path=None)[source]#

Expand one or more glob patterns to match file paths.

Parameters:

patterns (str or list of str) – Glob pattern(s) or explicit file paths. Wildcards like *.nii.gz or *.tif are supported. Both relative and absolute paths are accepted. When multiple patterns are provided, explicit paths preserve their order, while glob expansions are sorted within that position.
base_path (str or Path, optional) – Base directory where relative patterns are applied. Defaults to the current working directory.

Returns:

A list of Path objects matching the provided patterns, preserving explicit input order and sorting only glob expansions.

Return type:

list of Path

Raises:

TypeError – If patterns is not a string, Path, or list of such types.
ValueError – If no files match the given patterns.

unravel.core.utils.get_stem(file_path)[source]#

Get the stem of a file path by removing known compound extensions (e.g., ‘.nii.gz’, ‘.ome.tif’, ‘.tar.gz’) and falling back to single-extension logic.

Parameters:: file_path (str or Path) – Path to a file.
Returns:: Stem of the file with all recognized extensions removed. E.g., path/to/file.nii.gz -> file
Return type:: str

unravel.core.utils.get_extension(file_path)[source]#

Get the extension of a file path, including compound extensions.

Parameters:: file_path (str or Path) – Path to a file.
Returns:: The extension of the file, including compound extensions. E.g., path/to/file.nii.gz -> .nii.gz
Return type:: str

unravel.core.utils.resolve_output_paths(file_paths, output_paths=None, ext=None, stem_suffix=None, base_dir=None, *, skip_existing=False, return_inputs=False)[source]#

Resolve and prepare output file path(s) based on input file(s) and an optional output argument.

The output_paths argument can include an optional suffix using colon notation:

“results/:_filtered” → output dir = “results/”, suffix = “_filtered”
“:_processed” → no dir, just apply suffix “_processed” next to inputs
“results/” → output dir only (adds ‘_out’ if input/output names would match)
“results/file.csv” → explicit file output

Suffix logic#

If input and output filenames (including extension) would match, ‘_out’ is added automatically to prevent overwriting.
If filenames differ (e.g., due to directory, stem, or extension), no default suffix is used.
User-specified suffixes (via colon notation or stem_suffix) always take precedence.

General rules#

One input, output is a file → use it directly.
One input, output is a directory → save inside it.
Multiple inputs, output must be a directory (auto-created).
No output → save next to each input file, adding a suffix if needed to avoid overwriting.
All parent directories for outputs are created automatically.

For multiple input files in nested directories, their relative structure under base_dir is preserved automatically. If base_dir is not provided, it is inferred as: The common parent directory of all inputs, if shared, or the cwd

Parallel processing#

This function is safe for parallel processing. It ensures all parent directories exist so workers can write immediately.

Example usage#

>>> inputs = ["data/a.nrrd", "data/b.nrrd"]
>>> outputs = resolve_output_paths(inputs, "results/:_aligned", ext=".nii.gz")
>>> for i, o in zip(inputs, outputs):
...     print(f"{i} → {o}")
data/a.nrrd → results/a.nii.gz
data/b.nrrd → results/b.nii.gz

>>> # Same format and name → '_out' added
>>> outputs = resolve_output_paths(inputs, "results/")
data/a.nrrd → results/a_out.nrrd

param file_paths:: One or more input file paths.
type file_paths:: list[str | Path]
param output_paths:: Output path with optional suffix using colon notation (e.g., ‘outdir/:_filtered’). If None, outputs are saved next to input files.
type output_paths:: str | Path | None, optional
param ext:: Optional override for file extension (e.g., ‘.csv’, ‘.nii.gz’). If None, keeps the input file’s extension.
type ext:: str | None, optional
param stem_suffix:: Manual suffix to append before the extension. Overrides any suffix in output_paths.
type stem_suffix:: str | None, optional
param base_dir:: Base directory to preserve relative paths under. If None, inferred automatically (common parent or CWD).
type base_dir:: str | Path | None, optional
param skip_existing:: If True, existing outputs are skipped.
type skip_existing:: bool, optional
param return_inputs:: If True, returns (filtered_inputs, outputs).
type return_inputs:: bool, optional
returns:: List of resolved output paths (always Path objects). If return_inputs=True, returns (filtered_inputs, outputs).
rtype:: list[Path] or (list[Path], list[Path])

unravel.core.utils.get_pad_percent(reg_outputs_path, pad_percent)[source]#