utilities#
Various utilities that do not have a specific category
- class CancelableProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(), *, max_tasks_per_child=None)[source]#
Bases:
ProcessPoolExecutor
- class FilePath(base, prefix=None, postfix=None, extension=None, asset_sub_type=None)[source]#
Bases:
object
- bytes_to_human(num)[source]#
Convert bytes to human-readable format.
- Parameters:
num (int) – Number of bytes
Returns
- str
Human-readable format of the number of bytes
- check_stopped(func)[source]#
Decorator to check if the object is stopped (self.stopped) before running the function. If the object is stopped, the function returns immediately. The force argument can be used to force the function to run even if the object is stopped and reset the stopped flag.
- deep_freeze(obj) object | MappingProxyType[source]#
Recursively make a data structure immutable (freeze it). Only works for dict, list, set, tuple and scalars (str/int/float/None/…).
Note
This is faster than deepcopy and more explicit because if we try to modify a frozen object, we get an error immediately.
- Parameters:
obj (object) – Typically a nested dict like a config
Returns
- object | MappingProxyType
An immutable version of the input object
- del_item_recursive(d: dict, keys: list) None[source]#
Delete a nested key from a dict. No-op if path does not exist.
- get_ok_n_ok_symbols()[source]#
Detect whether we can print✓/✗ in this terminal
otherwise, use [OK]/[FAIL] instead
Returns
- tuple
ok_symbol, fail_symbol
- handle_deprecated_args(deprecated_args_map)[source]#
Decorator to handle deprecated arguments by renaming them. It takes a dictionary and renames old arguments to new ones.
- Parameters:
deprecated_args_map (dict) – Dictionary mapping old argument names to new ones.
Returns
- infer_origin_from_caller() str[source]#
Infer an origin string from the caller’s module, class and function name. This is typically the method that is mutating the config.
Returns
- str
The inferred origin string in the format “module.class.function” or “module.function” if
- requires_assets(asset_specs)[source]#
Decorator to check if the required files exist before running the function. For the case of channel, it can be extracted (in this order) from - the kwargs - the FilePath object - the instance to which the wrapped function belongs
- Parameters:
asset_specs (List[FilePath]) – List of FilePath objects
- title_to_snake(string)[source]#
Convert a human/TitleCase/mixed string into canonical snake_case.
Rules: - Inserts underscores at CamelCase boundaries (e.g., “TubeMap” -> “tube_map”). - Keeps acronyms together (e.g., “HTTPServer” -> “http_server”, “ROI3D” -> “roi3d”). - Normalizes spaces, hyphens, slashes, and other punctuation to single underscores. - Collapses multiple underscores and trims leading/trailing underscores. - Lowercases the final result.
- Parameters:
s (str) – Input string (e.g., tab title, pipeline name, or file-ish label).
Returns
- str
Snake_case version of the input. Empty string if input is None/empty.
Examples
>>> title_to_snake("TubeMap") 'tube_map' >>> title_to_snake("Sample Info") 'sample_info' >>> title_to_snake("Tract-Map") 'tract_map' >>> title_to_snake("ROI3D") 'roi3_d' # FIXME: ideally 'roi3d' but hard to do robustly >>> title_to_snake("HTTPServerError") 'http_server_error'
- trim_or_pad(lst: List, target_len: int, pad_value=0)[source]#
Strip or pad a list to a target length.
- Parameters:
lst (list) – The list to strip or pad
target_len (int) – The target length
pad_value (any) – The value to use for padding
Returns
- list
The stripped or padded list