ablator.utils package#

Submodules#

ablator.utils.base module#

class ablator.utils.base.Dummy(*args, **kwargs)[source]#: Bases: object

class ablator.utils.base.Lock(timeout: float | None = None)[source]#

Bases: object

acquire()[source]#

release()[source]#

ablator.utils.base.apply_lambda_to_iter(iterable: Iterable, fn: Callable) → Any[source]#

Applies a given function fn to each element of an iterable data structure.

This function recursively applies fn to elements within nested dictionaries or lists. It can be used for converting torch.Tensor elements to NumPy arrays or moving tensors to a specified device.

Parameters:

iterableIterable: The input iterable.
fnCallable: The function to apply to each element.

Returns:

ty.Any: The resulting data structure after applying fn to each element of the input iterable. The type of the returned object matches the type of the input iterable.

ablator.utils.base.debugger_is_active() → bool[source]#

Check if the debugger is currently active.

Returns:

bool: True if the debugger is active, False otherwise.

Notes

Return if the debugger is currently active

ablator.utils.base.get_latest_chkpts(checkpoint_dir: Path) → list[pathlib.Path][source]#

Get a list of all checkpoint files in a directory, sorted from the latest to the earliest.

Parameters:

checkpoint_dirPath: The directory containing checkpoint files.

Returns:

list[Path]: A list of the checkpoint files sorted by filename.

ablator.utils.base.get_lr(optimizer: Optimizer | dict) → float[source]#

Get the learning rate from an optimizer.

Parameters:

optimizertorch.optim.Optimizer | dict: The optimizer.

Returns:

float: The learning rate.

ablator.utils.base.is_oom_exception(err: RuntimeError) → bool[source]#

is_oom_exception checks whether the exception is caused by CUDA out of memory errors.

Parameters:

errRuntimeError: the exception to parse

Returns:

bool: whether the exception indicates out of memory error.

ablator.utils.base.iter_to_device(data_dict: Iterable, device: str) → Sequence[Tensor] | dict[str, torch.Tensor][source]#

Moving torch.Tensor elements to the specified device.

Parameters:

data_dictIterable: The input dictionary or list containing torch.Tensor elements.
devicestr: The target device for the tensors.

Returns:

ty.Union[Sequence[torch.Tensor], dict[str, torch.Tensor]]: The input data with tensors moved to the target device.

ablator.utils.base.iter_to_numpy(iterable: Iterable) → Any[source]#

Convert elements of the input iterable to NumPy arrays if they are torch.Tensor objects.

Parameters:

iterableIterable: The input iterable.

Returns:

ty.Any: The iterable with torch.Tensor elements replaced with their NumPy array equivalents.

ablator.utils.base.num_format(value: str | int | float | integer | floating, width: int = 8) → str[source]#

Format number to be no larger than width by converting to scientific notation when the value exceeds width either by informative decimal places or size.

Parameters:

valuestr | int | float | np.integer | np.floating: the value to format
widthint: the width of the decimal places, by default 8.

Returns:

str: The formatted string representation of the value

Examples

>>> num_format(123456, width=8)
123456
>>> num_format(123456789, width=8)
1.23e+08
>>> num_format(1234.5678, width=8)
1.23e+03
>>> num_format(0.000012345, width=8)
1.23e-05
>>> num_format(np.float64(12345678.12345678), width=8)
1.23e+07

ablator.utils.base.parse_device(device: str | list[str] | int) → Any[source]#

Parse a device string, an integer, or a list of device strings or integers.

Parameters:

devicety.Union[str, list[str], int]: The target device for the tensors.

Returns:

ty.Any: The parsed device string, integer, or list of device strings or integers.

Raises:

ValueError: If the device string is not one of {‘cpu’, ‘cuda’} or doesn’t start with ‘cuda:’.
AssertionError: If cuda is not found on system or gpu number of device is not available.

Examples

>>> parse_device("cpu")
'cpu'
>>> parse_device("cuda")
'cuda'
>>> parse_device("cuda:0")
'cuda:0'
>>> parse_device(["cpu", "cuda"])
['cpu', 'cuda']
>>> parse_device(["cpu", "cuda:0", "cuda:1", "cuda:2"])
['cpu', 'cuda:0', 'cuda:1', 'cuda:2']

ablator.utils.base.set_seed(seed: int) → int[source]#

Set the random seed.

Parameters:

seedint: The random seed to set.

Returns:

int: The set random seed.

ablator.utils.file module#

ablator.utils.file.clean_checkpoints(checkpoint_folder: Path, n_checkpoints: int)[source]#

Remove all but the n latest checkpoints from the given directory.

Parameters:

checkpoint_folderPath: Directory containing the checkpoint files.
n_checkpointsint: Number of checkpoints to keep.

ablator.utils.file.default_val_parser(val: Any) → Any[source]#

Converts the input value to a JSON compatible format.

Parameters:

valty.Any: The value to be converted.

Returns:

ty.Any: The converted value.

ablator.utils.file.dict_to_json(dict_: dict) → str[source]#

Convert a dictionary into a JSON string.

Parameters:

dict_dict: The dictionary to be converted.

Returns:

str: The JSON string representation of the dictionary.

ablator.utils.file.json_to_dict(json_: str) → dict[source]#

Convert a JSON string into a dictionary.

Parameters:

json_str: JSON string to be converted.

Returns:

dict: A dictionary representation of the JSON string.

ablator.utils.file.make_sub_dirs(parent: str | Path, *dir_names: str) → list[pathlib.Path][source]#

Create subdirectories under the given parent directory.

Parameters:

parentstr | Path: Parent directory where subdirectories should be created.
*dir_namesstr: Names of the subdirectories to create.

Returns:

list[Path]: A list of created subdirectory paths.

Examples

>>> parent_directory = "C:/example_parent_directory"
>>> subdirectory_names = ["subdir1", "subdir2", "subdir3"]
>>> created_subdirectories = make_sub_dirs(parent_directory, *subdirectory_names)
>>> created_subdirectories
[Path('C:/example_parent_directory/subdir1'),
 Path('C:/example_parent_directory/subdir2'),
 Path('C:/example_parent_directory/subdir3')]

ablator.utils.file.nested_set(dict_: dict, keys: list[str], value: Any) → dict[source]#

Set a value in a nested dictionary.

Parameters:

dict_dict: The dictionary to update.
keyslist[str]: List of keys representing the nested path.
valuety.Any: The value need to set at the specified path.

Returns:

dict: The updated dictionary with the new value set.

Examples

>>> dict_ = {'a': {'b': {'c': 1}}}
>>> nested_set(dict_, ['a', 'b', 'c'], 2)
>>> dict_
{'a': {'b': {'c': 2}}}

ablator.utils.file.save_checkpoint(state: object, filename: str = 'checkpoint.pt')[source]#

Save a checkpoint of the given state.

Parameters:

stateobject: Model State dictionary to save.
filenamestr: The name of the checkpoint file, by default “checkpoint.pt”.

ablator.utils.file.truncate_utf8_chars(filename: Path, last_char: str)[source]#

ablator.utils.progress_bar module#

class ablator.utils.progress_bar.Display[source]#

Bases: object

Class for handling display for terminal and notebook.

Attributes:

_cursescurses: curses object
stdscrcurses.initscr(): To initialize the curses library and create a window object stdscr.
nrowsint: height of stdscr window.
nrowsint: width of stdscr window.
html_widgetwidget.HTML: html_widget with empty value
html_valuestr: html value for widget

close()[source]#

print_texts(texts: list[str])[source]#

class ablator.utils.progress_bar.ProgressBar(total_steps: int, epoch_len: int | None = None, logfile: ~pathlib.Path | None = None, update_interval: int = 1, remote_display: <ablator.utils.progress_bar.ActorClass(RemoteProgressBar) object at 0x7f0d3ce62290> | None = None, uid: str | None = None)[source]#

Bases: object

Class for using progress bar. [config.verbose = “progress”]

Parameters:

total_stepsint: The total steps the progress bar is expected to iterate
epoch_lenint | None: The number of iterations for a single epoch that is used to calculate the time it takes per epoch.
logfilePath | None: Path of logfile to read from to display on console.
update_intervalint: The time interval by which the progress bar will update the displayed metrics.
remote_displayty.Optional[RemoteProgressBar]: A Remote display that can be used to report the progress to instead of printing it directly on console
uidstr | None: The trial uid that is used to report the metrics.

close()[source]#

classmethod make_bar(current_iteration: int, start_time: float, epoch_len: int | None, total_steps: int, ncols: int | None = None)[source]#

classmethod make_metrics_message(metrics: dict[str, Any], nrows: int | None = None, ncols: int | None = None) → list[source]#

make_print_message() → list[source]#

property ncols#

property nrows: int | None#

reset() → None[source]#

update_metrics(metrics: dict[str, Any], current_iteration: int)[source]#

class ablator.utils.progress_bar.RemoteDisplay(remote_progress_bar: <ablator.utils.progress_bar.ActorClass(RemoteProgressBar) object at 0x7f0d3ce62290>, update_interval: int = 1)[source]#

Bases: Display

refresh(force: bool = False)[source]#

ablator.utils.progress_bar.get_last_line(filename: Path | None) → str | None[source]#

This functions gets the last line from the file.

Parameters:

filenamePath | None: The path of the filename.

Returns:

str | None: None if file doesn’t exists or the last line of the file as a string.

ablator.utils.progress_bar.in_notebook() → bool[source]#

ablator.utils package#

Submodules#

ablator.utils.base module#

ablator.utils.file module#

ablator.utils.progress_bar module#

Module contents#