utils

Collection of utility functions for xai4mri.

Author: Simon M. Hofmann
Years: 2023

Bcolors

Use for color print-commands in console.

Usage

print(Bcolors.HEADER + "Warning: No active frommets remain. Continue?" + Bcolors.ENDC)
print(Bcolors.OKBLUE + "Warning: No active frommets remain. Continue?" + Bcolors.ENDC)

For more colors:

Name	Color Code
CSELECTED	\33[7m
CBLACK	\33[30m
CRED	\33[31m
CGREEN	\33[32m
CYELLOW	\33[33m
CBLUE	\33[34m
CVIOLET	\33[35m
CBEIGE	\33[36m
CWHITE	\33[37m
CBLACKBG	\33[40m
CREDBG	\33[41m
CGREENBG	\33[42m
CYELLOWBG	\33[43m
CBLUEBG	\33[44m
CVIOLETBG	\33[45m
CBEIGEBG	\33[46m
CWHITEBG	\33[47m
CGREY	\33[90m
CBEIGE2	\33[96m
CWHITE2	\33[97m
CGREYBG	\33[100m
CREDBG2	\33[101m
CGREENBG2	\33[102m
CYELLOWBG2	\33[103m
CBLUEBG2	\33[104m
CVIOLETBG2	\33[105m
CBEIGEBG2	\33[106m
CWHITEBG2	\33[107m

For preview use:

for i in (
    [1, 4, 7] + list(range(30, 38)) + list(range(40, 48)) + list(range(90, 98)) + list(range(100, 108))
):  # range(107+1)
    print(i, "\33[{}m".format(i) + "ABC & abc" + "\33[0m")

ask_true_false

ask_true_false(question: str, col: str = 'b') -> None

Ask user for input for a given True-or-False question.

Parameters:

Name	Type	Description	Default
question	str	Question to be asked to the user.	required
col	str	Print-color of question ['b'(lue), 'g'(reen), 'y'(ellow), 'r'(ed)]	'b'

Returns:

Type	Description
None	Answer to the question.

Source code in src/xai4mri/utils.py

@_true_false_request
def ask_true_false(question: str, col: str = "b") -> None:
    """
    Ask user for input for a given `True`-or-`False` question.

    :param question: Question to be asked to the user.
    :param col: Print-color of question ['b'(lue), 'g'(reen), 'y'(ellow), 'r'(ed)]
    :return: Answer to the question.
    """
    cprint(string=question, col=col)

browse_files

browse_files(
    initialdir: str | None = None,
    filetypes: str | None = None,
) -> str

Interactively browse and choose a file from the finder.

This function is a wrapper around the tkinter.filedialog.askopenfilename function and uses a GUI to select a file.

Note

ARGS MUST BE NAMED 'initialdir' and 'filetypes'.

Parameters:

Name	Type	Description	Default
initialdir	str | None	Directory, where the search should start	None
filetypes	str | None	What type of file-ending is searched for (suffix, e.g., *.jpg)	None

Returns:

Type	Description
str	Path to the chosen file.

Source code in src/xai4mri/utils.py

def browse_files(initialdir: str | None = None, filetypes: str | None = None) -> str:
    """
    Interactively browse and choose a file from the finder.

    This function is a wrapper around the `tkinter.filedialog.askopenfilename` function
    and uses a GUI to select a file.

    ??? note
        ARGS MUST BE NAMED 'initialdir' and 'filetypes'.

    :param initialdir: Directory, where the search should start
    :param filetypes: What type of file-ending is searched for (suffix, e.g., `*.jpg`)
    :return: Path to the chosen file.
    """
    import tkinter  # noqa: PLC0415, RUF100

    root = tkinter.Tk()
    root.withdraw()

    kwargs = {}
    if initialdir:
        kwargs.update({"initialdir": initialdir})
    if filetypes:
        kwargs.update({"filetypes": [(filetypes + " File", "*." + filetypes.lower())]})

    return tkinter.filedialog.askopenfilename(parent=root, title="Choose the file", **kwargs)

bytes_to_rep_string

bytes_to_rep_string(number_of_bytes: int) -> str

Convert the number of bytes into representative string.

The function is used to convert the number of bytes into a human-readable format.

The function rounds the number of bytes to two decimal places.

Example

print(bytes_to_rep_string(1_500_000))  # 1.5 MB
print(bytes_to_rep_string(1_005_500_000))  # 1.01 GB

Parameters:

Name	Type	Description	Default
number_of_bytes	int	Number of bytes.	required

Returns:

Type	Description
str	Representative string of the given bytes number.

Source code in src/xai4mri/utils.py

def bytes_to_rep_string(number_of_bytes: int) -> str:
    """
    Convert the number of bytes into representative string.

    The function is used to convert the number of bytes into a human-readable format.

    !!! note "The function rounds the number of bytes to two decimal places."

    !!! example
        ```python
        print(bytes_to_rep_string(1_500_000))  # 1.5 MB
        print(bytes_to_rep_string(1_005_500_000))  # 1.01 GB
        ```

    :param number_of_bytes: Number of bytes.
    :return: Representative string of the given bytes number.
    """
    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
    i = int(math.floor(math.log(number_of_bytes, 10**3)))
    p = math.pow(10**3, i)
    size_ = round(number_of_bytes / p, 2)

    return f"{size_} {size_name[i]}"

check_storage_size

check_storage_size(obj: Any, verbose: bool = True) -> int

Return the storage size of a given object in an appropriate unit.

Example

import numpy as np

a = np.random.rand(500, 500)
size_in_bytes = check_storage_size(obj=a, verbose=True)  # "Size of given object: 2.0 MB"

Parameters:

Name	Type	Description	Default
obj	Any	Any object in the workspace.	required
verbose	bool	Print human-readable size of the object and additional information.	True

Returns:

Type	Description
int	Object size in bytes.

Source code in src/xai4mri/utils.py

def check_storage_size(obj: Any, verbose: bool = True) -> int:
    """
    Return the storage size of a given object in an appropriate unit.

    !!! example
        ```python
        import numpy as np

        a = np.random.rand(500, 500)
        size_in_bytes = check_storage_size(obj=a, verbose=True)  # "Size of given object: 2.0 MB"
        ```

    :param obj: Any object in the workspace.
    :param verbose: Print human-readable size of the object and additional information.
    :return: Object size in bytes.
    """
    if isinstance(obj, np.ndarray):
        size_bytes = obj.nbytes
        message = ""
    else:
        size_bytes = sys.getsizeof(obj)
        message = "Only trustworthy for pure python objects, otherwise returns size of view object."

    if size_bytes == 0:
        if verbose:
            print("Size of given object equals 0 B")
        return 0

    if verbose:
        print(f"Size of given object: {bytes_to_rep_string(number_of_bytes=size_bytes)} {message}")

    return size_bytes

chop_microseconds

chop_microseconds(delta: timedelta) -> timedelta

Chop microseconds from given time delta.

Parameters:

Name	Type	Description	Default
delta	timedelta	time delta	required

Returns:

Type	Description
timedelta	time delta without microseconds

Source code in src/xai4mri/utils.py

def chop_microseconds(delta: timedelta) -> timedelta:
    """
    Chop microseconds from given time delta.

    :param delta: time delta
    :return: time delta without microseconds
    """
    return delta - timedelta(microseconds=delta.microseconds)

compute_array_size

compute_array_size(
    shape: tuple[int, ...] | list[int, ...],
    dtype: dtype | int | float,
    verbose: bool = False,
) -> int

Compute the theoretical size of a NumPy array with the given shape and data type.

The idea is to compute the size of the array before creating it to avoid potential memory issues.

Parameters:

Name	Type	Description	Default
shape	tuple[int, ...] | list[int, ...]	Shape of the array, e.g., (n_samples, x, y, z).	required
dtype	dtype | int | float	Data type of the array elements (e.g., np.float32, np.int64, np.uint8, int, float)	required
verbose	bool	Print the size of the array in readable format or not.	False

Returns:

Type	Description
int	Size of the array in bytes.

Source code in src/xai4mri/utils.py

def compute_array_size(
    shape: tuple[int, ...] | list[int, ...], dtype: np.dtype | int | float, verbose: bool = False
) -> int:
    """
    Compute the theoretical size of a NumPy array with the given shape and data type.

    The idea is to compute the size of the array before creating it to avoid potential memory issues.

    :param shape: Shape of the array, e.g., `(n_samples, x, y, z)`.
    :param dtype: Data type of the array elements (e.g., np.float32, np.int64, np.uint8, int, float)
    :param verbose: Print the size of the array in readable format or not.
    :return: Size of the array in bytes.
    """
    # Get the size of each element in bytes
    element_size = np.dtype(dtype).itemsize
    # Compute the total number of elements
    num_elements = np.prod(shape)
    # Compute the total size in bytes
    total_size_in_bytes = num_elements * element_size
    if verbose:
        print(f"Size of {dtype.__name__}-array of shape {shape}: {bytes_to_rep_string(total_size_in_bytes)}")
    return total_size_in_bytes

cprint

cprint(
    string: str,
    col: str | None = None,
    fm: str | None = None,
    ts: bool = False,
) -> None

Colorize and format print-out.

Add leading time-stamp (fs) if required.

Parameters:

Name	Type	Description	Default
string	str	Print message.	required
col	str | None	Color:'p'(ink), 'b'(lue), 'g'(reen), 'y'(ellow), OR 'r'(ed).	None
fm	str | None	Format: 'ul'(:underline) OR 'bo'(:bold).	None
ts	bool	Add leading time-stamp.	False

Source code in src/xai4mri/utils.py

def cprint(string: str, col: str | None = None, fm: str | None = None, ts: bool = False) -> None:
    """
    Colorize and format print-out.

    Add leading time-stamp (fs) if required.

    :param string: Print message.
    :param col: Color:'p'(ink), 'b'(lue), 'g'(reen), 'y'(ellow), OR 'r'(ed).
    :param fm: Format: 'ul'(:underline) OR 'bo'(:bold).
    :param ts: Add leading time-stamp.
    """
    if col:
        col = col[0].lower()
        if col not in {"p", "b", "g", "y", "r"}:
            msg = "col must be 'p'(ink), 'b'(lue), 'g'(reen), 'y'(ellow), 'r'(ed)"
            raise ValueError(msg)
        col = Bcolors.DICT[col]

    if fm:
        fm = fm[0:2].lower()
        if fm not in {"ul", "bo"}:
            msg = "fm must be 'ul'(:underline), 'bo'(:bold)"
            raise ValueError(msg)
        fm = Bcolors.DICT[fm]

    if ts:
        pfx = ""  # collecting leading indent or new line
        while string.startswith("\n") or string.startswith("\t"):
            pfx += string[:1]
            string = string[1:]
        string = f"{pfx}{datetime.now():%Y-%m-%d %H:%M:%S} | " + string

    print(f"{col if col else ''}{fm if fm else ''}{string}{Bcolors.ENDC}")

function_timed

function_timed(
    dry_funct: Callable[..., Any] | None = None,
    ms: bool | None = None,
) -> Callable[..., Any]

Time the processing duration of wrapped function.

How to use the function_timed

The following returns the duration of the function call without micro-seconds:

# Implement a function to be timed
@function_timed
def abc():
    return 2 + 2


# Call the function and get the processing time
abc()

The following returns micro-seconds as well:

@function_timed(ms=True)
def abcd():
    return 2 + 2

Parameters:

Name	Type	Description	Default
dry_funct	Callable[..., Any] | None	Parameter can be ignored. Results in output without micro-seconds.	None
ms	bool | None	If micro-seconds should be printed, set to True.	None

Returns:

Type	Description
Callable[..., Any]	Wrapped function with processing time.

Source code in src/xai4mri/utils.py

def function_timed(dry_funct: Callable[..., Any] | None = None, ms: bool | None = None) -> Callable[..., Any]:
    """
    Time the processing duration of wrapped function.

    !!! example "How to use the `function_timed`"

        The following returns the duration of the function call without micro-seconds:
        ```python
        # Implement a function to be timed
        @function_timed
        def abc():
            return 2 + 2


        # Call the function and get the processing time
        abc()
        ```

        The following returns micro-seconds as well:
        ```python
        @function_timed(ms=True)
        def abcd():
            return 2 + 2
        ```

    :param dry_funct: *Parameter can be ignored*. Results in output without micro-seconds.
    :param ms: If micro-seconds should be printed, set to `True`.
    :return: Wrapped function with processing time.
    """

    def _function_timed(funct):
        @wraps(funct)
        def wrapper(*args, **kwargs):
            """Wrap function to time the processing duration of wrapped function."""
            start_timer = datetime.now()

            # whether to suppress wrapper: use functimer=False in main funct
            w = kwargs.pop("functimer", True)

            output = funct(*args, **kwargs)

            duration = datetime.now() - start_timer

            if w:
                if ms:
                    print(f"\nProcessing time of {funct.__name__}: {duration} [h:m:s:ms]")

                else:
                    print(f"\nProcessing time of {funct.__name__}: {chop_microseconds(duration)} [h:m:s]")

            return output

        return wrapper

    if dry_funct:
        return _function_timed(dry_funct)

    return _function_timed

get_string_overlap

get_string_overlap(s1: str, s2: str) -> str

Find the longest overlap between two strings, starting from the left.

Example

get_string_overlap("Hello there Bob", "Hello there Alice")  # "Hello there "

Parameters:

Name	Type	Description	Default
s1	str	First string.	required
s2	str	Second string.	required

Returns:

Type	Description
str	Longest overlap between the two strings.

Source code in src/xai4mri/utils.py

def get_string_overlap(s1: str, s2: str) -> str:
    """
    Find the longest overlap between two strings, starting from the left.

    !!! example
        ```python
        get_string_overlap("Hello there Bob", "Hello there Alice")  # "Hello there "
        ```
    :param s1: First string.
    :param s2: Second string.
    :return: Longest overlap between the two strings.
    """
    s = difflib.SequenceMatcher(None, s1, s2)
    pos_a, _, size = s.find_longest_match(0, len(s1), 0, len(s2))  # _ = pos_b

    return s1[pos_a : pos_a + size]

normalize

normalize(
    array: ndarray,
    lower_bound: int | float,
    upper_bound: int | float,
    global_min: int | float | None = None,
    global_max: int | float | None = None,
) -> ndarray

Min-max-scaling: Normalize an input array to lower and upper bounds.

Parameters:

Name	Type	Description	Default
array	ndarray	Array to be transformed.	required
lower_bound	int | float	Lower bound a.	required
upper_bound	int | float	Upper bound b.	required
global_min	int | float | None	Global minimum. If the array is part of a larger tensor, normalize w.r.t. global min and ...	None
global_max	int | float | None	Global maximum. If the array is part of a larger tensor, normalize w.r.t. ... and global max (i.e., tensor min/max)	None

Returns:

Type	Description
ndarray	Normalized array.

Source code in src/xai4mri/utils.py

def normalize(
    array: np.ndarray,
    lower_bound: int | float,
    upper_bound: int | float,
    global_min: int | float | None = None,
    global_max: int | float | None = None,
) -> np.ndarray:
    """
    Min-max-scaling: Normalize an input array to lower and upper bounds.

    :param array: Array to be transformed.
    :param lower_bound: Lower bound `a`.
    :param upper_bound: Upper bound `b`.
    :param global_min: Global minimum.
                       If the array is part of a larger tensor, normalize w.r.t. global min and ...
    :param global_max: Global maximum.
                       If the array is part of a larger tensor, normalize w.r.t. ... and global max
                       (i.e., tensor min/max)
    :return: Normalized array.
    """
    if lower_bound >= upper_bound:
        msg = "lower_bound must be < upper_bound"
        raise ValueError(msg)

    array = np.array(array)
    a, b = lower_bound, upper_bound

    if global_min is not None:
        if not np.isclose(global_min, np.nanmin(array)) and global_min > np.nanmin(array):
            # Allow a small tolerance for global_min
            msg = "global_min must be <= np.nanmin(array)"
            raise ValueError(msg)
        mini = global_min
    else:
        mini = np.nanmin(array)

    if global_max is not None:
        if not np.isclose(global_max, np.nanmax(array)) and global_max < np.nanmax(array):
            # Allow a small tolerance for global_max
            msg = "global_max must be >= np.nanmax(array)"
            raise ValueError(msg)
        maxi = global_max
    else:
        maxi = np.nanmax(array)

    return (b - a) * ((array - mini) / (maxi - mini)) + a

run_gpu_test

run_gpu_test(log_device_placement: bool = False) -> bool

Test GPU implementation.

Parameters:

Name	Type	Description	Default
log_device_placement	bool	Log device placement.	False

Returns:

Type	Description
bool	GPU available or not.

Source code in src/xai4mri/utils.py

def run_gpu_test(log_device_placement: bool = False) -> bool:
    """
    Test GPU implementation.

    :param log_device_placement: Log device placement.
    :return: GPU available or not.
    """
    import tensorflow as tf  # noqa: PLC0415, RUF100

    n_gpus = len(tf.config.list_physical_devices("GPU"))
    gpu_available = n_gpus > 0
    cprint(string=f"\nNumber of GPU device(s) available: {n_gpus}", col="g" if gpu_available else "r", fm="bo")

    # Run some operations on the GPU/CPU
    device_name = ["/gpu:0", "/cpu:0"] if gpu_available else ["/cpu:0"]
    tf.debugging.set_log_device_placement(log_device_placement)
    for device in device_name:
        for shape in [6000, 12000]:
            cprint(string=f"\nRun operations on device: {device} using tensor with shape: {shape}", col="b", fm="ul")
            with tf.device(device):
                # Create some tensors and perform an operation
                start_time = datetime.now()
                a = tf.constant([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
                b = tf.constant([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]])
                c = tf.matmul(a, b)
                print(f"{c = }")

                # Create some more complex tensors and perform an operation
                random_matrix = tf.compat.v1.random_uniform(shape=(shape, shape), minval=0, maxval=1)
                dot_operation = tf.matmul(random_matrix, tf.transpose(random_matrix))
                sum_operation = tf.reduce_sum(dot_operation)
                print(f"{sum_operation = }")

            print("\n")
            cprint(string=f"Shape: {(shape, shape)} | Device: {device}", col="y")
            cprint(string=f"Time taken: {datetime.now() - start_time}", col="y")
            print("\n" + "*<o>" * 15)

    cprint(string=f"\nGPU available: {gpu_available}", col="g" if gpu_available else "y", fm="bo")
    return gpu_available

tree

tree(directory: str | Path) -> None

Print the directory tree starting at directory.

Use the same way as shell command tree.

This leads to output such as:

directory/
├── _static/
│   ├── embedded/
│   │   ├── deep_file
│   │   └── very/
│   │       └── deep/
│   │           └── folder/
│   │               └── very_deep_file
│   └── less_deep_file
├── about.rst
├── conf.py
└── index.rst

Source code in src/xai4mri/utils.py

def tree(directory: str | Path) -> None:
    """
    Print the directory tree starting at `directory`.

    Use the same way as `shell` command `tree`.

    !!! example "This leads to output such as:"
        ```plaintext
        directory/
        ├── _static/
        │   ├── embedded/
        │   │   ├── deep_file
        │   │   └── very/
        │   │       └── deep/
        │   │           └── folder/
        │   │               └── very_deep_file
        │   └── less_deep_file
        ├── about.rst
        ├── conf.py
        └── index.rst
        ```
    """
    paths = _DisplayablePath.make_tree(Path(directory))
    for path in paths:
        print(path.displayable())