Utils
coola.utils ¶
Provide utility functions for the coola library.
This package contains utility modules for various common operations:
array: Array utility functionsenv_vars: Environment variable handlingfallback: Fallback implementations for optional dependenciesformat: String formatting and representation utilitiesimports: Import checking and lazy loading for optional dependenciesintrospection: Code introspection utilitiesmapping: Mapping/dict utility functionspath: Path manipulation utilitiesstats: Statistical utility functionstensor: Tensor utility functionsversion: Version comparison utilities
coola.utils.array ¶
Implement some utility functions for numpy.ndarrays.
coola.utils.array.to_array ¶
to_array(
data: Sequence[int | float] | Tensor | ndarray,
) -> ndarray
Convert the input to a numpy.ndarray.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
Sequence[int | float] | Tensor | ndarray
|
The data to convert to a NumPy array. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
A NumPy array. |
Example
>>> from coola.utils.array import to_array
>>> x = to_array([1, 2, 3, 4, 5])
>>> x
array([1, 2, 3, 4, 5])
>>> import torch
>>> x = to_array(torch.tensor([1, 2, 3, 4, 5]))
>>> x
array([1, 2, 3, 4, 5])
coola.utils.env_vars ¶
Implement some utility functions to manage environment variables.
coola.utils.env_vars.get_required_env_var ¶
get_required_env_var(name: str) -> str
Retrieve a required environment variable with validation.
This function fetches an environment variable and ensures it exists and contains a non-empty value after stripping whitespace. If the variable is missing or empty, a ValueError is raised with a descriptive message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The name of the environment variable to retrieve. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The value of the environment variable with leading/trailing whitespace |
str
|
removed. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the environment variable is not set or contains only whitespace. |
Example
from coola.utils.env_vars import get_required_env_var os.environ['API_KEY'] = 'my-secret-key' get_required_env_var('API_KEY') 'my-secret-key' get_required_env_var('MISSING_VAR') # doctest: +SKIP ValueError: Environment variable 'MISSING_VAR' is required but not set
coola.utils.env_vars.temp_env_vars ¶
temp_env_vars(
env_vars: dict[str, Any],
) -> Generator[None, None, None]
Context manager to temporarily set or modify environment variables.
This context manager allows you to temporarily change environment variables within a specific scope. All changes are automatically reverted when exiting the context, even if an exception occurs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env_vars
|
dict[str, Any]
|
Environment variables to set as keyword arguments. Keys are variable names, values are the values to set. Values are automatically converted to strings. |
required |
Behavior
- If a variable already exists, its original value is saved and restored
- If a variable doesn't exist, it's created temporarily and removed on exit
- All operations are guaranteed to execute via try/finally
- Thread-safe for the current process (but note that os.environ affects the entire process, not just the current thread)
Example
>>> from coola.utils.env_vars import temp_env_vars
>>> # Temporarily override an existing variable
>>> os.environ["HOME"] = "/original/home"
>>> with temp_env_vars({"HOME": "/tmp/home"}):
... print(os.environ["HOME"]) # '/tmp/home'
...
>>> print(os.environ["HOME"]) # '/original/home'
>>> # Temporarily create new variables
>>> with temp_env_vars({"API_KEY": "secret123", "DEBUG": "true"}):
... print(os.environ["API_KEY"]) # 'secret123'
... print(os.environ["DEBUG"]) # 'true'
...
>>> print(os.environ.get("API_KEY")) # None (removed)
Notes
Changes to os.environ affect the entire Python process, not just the current thread. Use with caution in multi-threaded applications.
coola.utils.format ¶
Implement some utility functions to compute string representations of objects.
coola.utils.format.find_best_byte_unit ¶
find_best_byte_unit(size: int) -> str
Return the best byte unit given the byte size.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size
|
int
|
The size in bytes. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The best unit. The supported units are: |
Raises:
| Type | Description |
|---|---|
ValueError
|
if size is negative. |
Example
>>> from coola.utils.format import find_best_byte_unit
>>> find_best_byte_unit(2)
'B'
>>> find_best_byte_unit(2048)
'KB'
>>> find_best_byte_unit(2097152)
'MB'
coola.utils.format.make_bar ¶
make_bar(value: float, length: int = 10) -> str
Create a progress bar string for a given value between 0 and 1.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
float
|
A float between 0 and 1 representing the metric value. |
required |
length
|
int
|
The length of the progress bar. Must be a positive integer. |
10
|
Returns:
| Type | Description |
|---|---|
str
|
A string representing the progress bar. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Example
>>> from coola.utils.format import make_bar
>>> make_bar(0.6)
'[██████░░░░]'
>>> make_bar(0.1)
'[█░░░░░░░░░]'
>>> make_bar(1.0)
'[██████████]'
>>> make_bar(0.0)
'[â–‘â–‘â–‘â–‘â–‘â–‘â–‘â–‘â–‘â–‘]'
coola.utils.format.repr_indent ¶
repr_indent(original: object, num_spaces: int = 2) -> str
Add indentations if the original string is a multi-lines string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
original
|
object
|
The original string. If the inputis not a
string, it will be converted to a string with the function
|
required |
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The indented string. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if num_spaces is not a positive integer. |
Example
>>> from coola.utils.format import repr_indent
>>> print(repr_indent("string1\nstring2\n string3", 4))
string1
string2
string3
coola.utils.format.repr_mapping ¶
repr_mapping(
mapping: Mapping[Any, Any],
sorted_keys: bool = False,
num_spaces: int = 2,
) -> str
Compute a string representation of a mapping.
This function was designed for flat dictionary. If you have a nested dictionary, you may consider other functions. Note that this function works for nested dict but the output may not be nice.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The mapping. |
required |
sorted_keys
|
bool
|
If |
False
|
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the mapping. |
Example
>>> from coola.utils.format import repr_mapping
>>> print(repr_mapping({"key1": "abc", "key2": "something\nelse"}))
(key1): abc
(key2): something
else
coola.utils.format.repr_mapping_line ¶
repr_mapping_line(
mapping: Mapping[Any, Any],
sorted_keys: bool = False,
separator: str = ", ",
) -> str
Compute a single line string representation of the given mapping.
This function is designed for flat dictionary. If you have a nested dictionary, you may consider other functions. Note that this function works for nested dict but the output may not be nice.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The mapping. |
required |
sorted_keys
|
bool
|
If |
False
|
separator
|
str
|
The separator to use between each key-value pair. |
', '
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the mapping. |
Example
>>> from coola.utils.format import repr_mapping_line
>>> repr_mapping_line({"key1": "abc", "key2": "meow", "key3": 42})
key1='abc', key2='meow', key3=42
coola.utils.format.repr_sequence ¶
repr_sequence(
sequence: Sequence[Any], num_spaces: int = 2
) -> str
Compute a string representation of a sequence.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sequence
|
Sequence[Any]
|
The sequence. |
required |
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the sequence. |
Example
>>> from coola.utils.format import repr_indent
>>> print(repr_sequence(["abc", "something\nelse"]))
(0): abc
(1): something
else
coola.utils.format.repr_sequence_line ¶
repr_sequence_line(
sequence: Sequence[Any], separator: str = ", "
) -> str
Compute a single line string representation of a sequence.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sequence
|
Sequence[Any]
|
The sequence. |
required |
separator
|
str
|
The separator to use between each item. |
', '
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the sequence. |
Example
>>> from coola.utils.format import repr_sequence_line
>>> repr_sequence_line(["abc", "meow", 42])
'abc', 'meow', 42
coola.utils.format.str_human_byte_size ¶
str_human_byte_size(
size: int, unit: str | None = None
) -> str
Get a human-readable representation of the byte size.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size
|
int
|
The size in bytes. |
required |
unit
|
str | None
|
The unit to use to show the byte size. If |
None
|
Returns:
| Type | Description |
|---|---|
str
|
The byte size in a human-readable format. |
Raises:
| Type | Description |
|---|---|
ValueError
|
if size is negative or unit is not supported. |
Example
>>> from coola.utils.format import str_human_byte_size
>>> str_human_byte_size(2)
'2.00 B'
>>> str_human_byte_size(2048)
'2.00 KB'
>>> str_human_byte_size(2097152)
'2.00 MB'
>>> str_human_byte_size(2048, unit="B")
'2,048.00 B'
coola.utils.format.str_indent ¶
str_indent(original: object, num_spaces: int = 2) -> str
Add indentations if the original string is a multi-lines string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
original
|
object
|
The original string. If the inputis not a
string, it will be converted to a string with the function
|
required |
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The indented string. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if num_spaces is not a positive integer. |
Example
>>> from coola.utils.format import str_indent
>>> print(str_indent("string1\nstring2\n string3", 4))
string1
string2
string3
coola.utils.format.str_mapping ¶
str_mapping(
mapping: Mapping[Any, Any],
sorted_keys: bool = False,
num_spaces: int = 2,
) -> str
Compute a string representation of a mapping.
This function was designed for flat dictionary. If you have a nested dictionary, you may consider other functions. Note that this function works for nested dict but the output may not be nice.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The mapping. |
required |
sorted_keys
|
bool
|
If |
False
|
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the mapping. |
Example
>>> from coola.utils.format import str_mapping
>>> print(str_mapping({"key1": "abc", "key2": "something\nelse"}))
(key1): abc
(key2): something
else
coola.utils.format.str_mapping_line ¶
str_mapping_line(
mapping: Mapping[Any, Any],
sorted_keys: bool = False,
separator: str = ", ",
) -> str
Compute a single line string representation of the given mapping.
This function is designed for flat dictionary. If you have a nested dictionary, you may consider other functions. Note that this function works for nested dict but the output may not be nice.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The mapping. |
required |
sorted_keys
|
bool
|
If |
False
|
separator
|
str
|
The separator to use between each key-value pair. |
', '
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the mapping. |
Example
>>> from coola.utils.format import str_mapping_line
>>> str_mapping_line({"key1": "abc", "key2": "meow", "key3": 42})
key1=abc, key2=meow, key3=42
coola.utils.format.str_sequence ¶
str_sequence(
sequence: Sequence[Any], num_spaces: int = 2
) -> str
Compute a string representation of a sequence.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sequence
|
Sequence[Any]
|
The sequence. |
required |
num_spaces
|
int
|
The number of spaces used for the indentation. |
2
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the sequence. |
Example
>>> from coola.utils.format import str_sequence
>>> print(str_sequence(["abc", "something\nelse"]))
(0): abc
(1): something
else
coola.utils.format.str_sequence_line ¶
str_sequence_line(
sequence: Sequence[Any], separator: str = ", "
) -> str
Compute a single line string representation of a sequence.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sequence
|
Sequence[Any]
|
The sequence. |
required |
separator
|
str
|
The separator to use between each item. |
', '
|
Returns:
| Type | Description |
|---|---|
str
|
The string representation of the sequence. |
Example
>>> from coola.utils.format import str_sequence_line
>>> str_sequence_line(["abc", "meow", 42])
abc, meow, 42
coola.utils.format.str_time_human ¶
str_time_human(seconds: float) -> str
Return a number of seconds in an easier format to read
hh:mm:ss.
If the number of seconds is bigger than 1 day, this representation also encodes the number of days.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
seconds
|
float
|
The number of seconds. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The number of seconds in a string format ( |
Example
>>> from coola.utils.format import str_time_human
>>> str_time_human(1.2)
'0:00:01.200000'
>>> str_time_human(61.2)
'0:01:01.200000'
>>> str_time_human(3661.2)
'1:01:01.200000'
coola.utils.imports ¶
Implement some utility functions to manage optional dependencies.
coola.utils.imports.LazyModule ¶
Bases: ModuleType
Define a proxy module that lazily imports a module.
The module is imported the first time it is actually used.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The fully-qualified module name to import. |
required |
Example
>>> from coola.utils.imports import LazyModule
>>> # Lazy version of import numpy as np
>>> np = LazyModule("numpy")
>>> # The module is imported the first time it is actually used.
>>> np.ones((2, 3))
array([[1., 1., 1.],
[1., 1., 1.]])
coola.utils.imports.check_jax ¶
check_jax() -> None
Check if the jax package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_jax
>>> check_jax()
coola.utils.imports.check_numpy ¶
check_numpy() -> None
Check if the numpy package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_numpy
>>> check_numpy()
coola.utils.imports.check_package ¶
check_package(
package: str, command: str | None = None
) -> None
Check if the given package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
package
|
str
|
The package name. |
required |
command
|
str | None
|
The command to install the package. |
None
|
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the package is not installed. |
Example
>>> from coola.utils.imports import check_package
>>> check_package("numpy")
coola.utils.imports.check_packaging ¶
check_packaging() -> None
Check if the packaging package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_packaging
>>> check_packaging()
coola.utils.imports.check_pandas ¶
check_pandas() -> None
Check if the pandas package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_pandas
>>> check_pandas()
coola.utils.imports.check_polars ¶
check_polars() -> None
Check if the polars package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_polars
>>> check_polars()
coola.utils.imports.check_pyarrow ¶
check_pyarrow() -> None
Check if the pyarrow package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_pyarrow
>>> check_pyarrow()
coola.utils.imports.check_torch ¶
check_torch() -> None
Check if the torch package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_torch
>>> check_torch()
coola.utils.imports.check_torch_numpy ¶
check_torch_numpy() -> None
Check if the torch and numpy packages are installed and
are compatible.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if one of the packages is not installed or if they are not compatible. |
Example
>>> from coola.utils.imports import check_torch_numpy
>>> check_torch_numpy()
coola.utils.imports.check_xarray ¶
check_xarray() -> None
Check if the xarray package is installed.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if the |
Example
>>> from coola.utils.imports import check_xarray
>>> check_xarray()
coola.utils.imports.decorator_package_available ¶
decorator_package_available(
fn: F, condition: Callable[[], bool]
) -> F
Implement a decorator to execute a function only if a package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
condition
|
Callable[[], bool]
|
The condition to check if a package is installed or not. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from functools import partial
>>> from coola.utils.imports import decorator_package_available, is_numpy_available
>>> decorator = partial(decorator_package_available, condition=is_numpy_available)
>>> @decorator
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function(2)
44
coola.utils.imports.is_jax_available
cached
¶
is_jax_available() -> bool
Indicate if the jax package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_jax_available
>>> is_jax_available()
coola.utils.imports.is_numpy_available
cached
¶
is_numpy_available() -> bool
Indicate if the numpy package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_numpy_available
>>> is_numpy_available()
coola.utils.imports.is_packaging_available
cached
¶
is_packaging_available() -> bool
Indicate if the packaging package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_packaging_available
>>> is_packaging_available()
coola.utils.imports.is_pandas_available
cached
¶
is_pandas_available() -> bool
Indicate if the pandas package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_pandas_available
>>> is_pandas_available()
coola.utils.imports.is_polars_available
cached
¶
is_polars_available() -> bool
Indicate if the polars package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_polars_available
>>> is_polars_available()
coola.utils.imports.is_pyarrow_available
cached
¶
is_pyarrow_available() -> bool
Indicate if the pyarrow package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_pyarrow_available
>>> is_pyarrow_available()
coola.utils.imports.is_torch_available
cached
¶
is_torch_available() -> bool
Indicate if the torch package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_torch_available
>>> is_torch_available()
coola.utils.imports.is_torch_numpy_available
cached
¶
is_torch_numpy_available() -> bool
Indicate if the torch and numpy packages are installed
and are compatible.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_torch_numpy_available
>>> is_torch_numpy_available()
coola.utils.imports.is_xarray_available
cached
¶
is_xarray_available() -> bool
Indicate if the xarray package is installed or not.
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import is_xarray_available
>>> is_xarray_available()
coola.utils.imports.jax_available ¶
jax_available(fn: F) -> F
Implement a decorator to execute a function only if jax
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import jax_available
>>> @jax_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.lazy_import ¶
lazy_import(name: str) -> LazyModule
Return a proxy of the module/package to lazily import.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The fully-qualified module name to import. |
required |
Returns:
| Type | Description |
|---|---|
LazyModule
|
A proxy module that lazily imports a module the first time it is actually used. |
Example
>>> from coola.utils.imports import lazy_import
>>> # Lazy version of import numpy as np
>>> np = lazy_import("numpy")
>>> # The module is imported the first time it is actually used.
>>> np.ones((2, 3))
array([[1., 1., 1.],
[1., 1., 1.]])
coola.utils.imports.module_available
cached
¶
module_available(name: str) -> bool
Indicate if a module is available or not.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The module name to check. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import module_available
>>> module_available("os")
True
>>> module_available("os.missing")
False
>>> module_available("missing.module")
False
coola.utils.imports.numpy_available ¶
numpy_available(fn: F) -> F
Implement a decorator to execute a function only if numpy
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import numpy_available
>>> @numpy_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.package_available
cached
¶
package_available(name: str) -> bool
Indicate if a package is available or not.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The package name to check. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.utils.imports import package_available
>>> package_available("os")
True
>>> package_available("missing_package")
False
coola.utils.imports.packaging_available ¶
packaging_available(fn: F) -> F
Implement a decorator to execute a function only if packaging
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import packaging_available
>>> @packaging_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.pandas_available ¶
pandas_available(fn: F) -> F
Implement a decorator to execute a function only if pandas
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import pandas_available
>>> @pandas_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.polars_available ¶
polars_available(fn: F) -> F
Implement a decorator to execute a function only if polars
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import polars_available
>>> @polars_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.pyarrow_available ¶
pyarrow_available(fn: F) -> F
Implement a decorator to execute a function only if pyarrow
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import pyarrow_available
>>> @pyarrow_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.raise_jax_missing_error ¶
raise_jax_missing_error() -> NoReturn
Raise a RuntimeError to indicate the jax package is
missing.
coola.utils.imports.raise_numpy_missing_error ¶
raise_numpy_missing_error() -> NoReturn
Raise a RuntimeError to indicate the numpy package is
missing.
coola.utils.imports.raise_package_missing_error ¶
raise_package_missing_error(
package_name: str, install_cmd: str
) -> NoReturn
Raise a RuntimeError for a missing package.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
package_name
|
str
|
The name of the missing package. |
required |
install_cmd
|
str
|
The pip install command for the package. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always raised to indicate the package is missing. |
coola.utils.imports.raise_packaging_missing_error ¶
raise_packaging_missing_error() -> NoReturn
Raise a RuntimeError to indicate the packaging package is
missing.
coola.utils.imports.raise_pandas_missing_error ¶
raise_pandas_missing_error() -> NoReturn
Raise a RuntimeError to indicate the pandas package is
missing.
coola.utils.imports.raise_polars_missing_error ¶
raise_polars_missing_error() -> NoReturn
Raise a RuntimeError to indicate the polars package is
missing.
coola.utils.imports.raise_pyarrow_missing_error ¶
raise_pyarrow_missing_error() -> NoReturn
Raise a RuntimeError to indicate the pyarrow package is
missing.
coola.utils.imports.raise_torch_missing_error ¶
raise_torch_missing_error() -> NoReturn
Raise a RuntimeError to indicate the torch package is
missing.
coola.utils.imports.raise_xarray_missing_error ¶
raise_xarray_missing_error() -> NoReturn
Raise a RuntimeError to indicate the xarray package is
missing.
coola.utils.imports.torch_available ¶
torch_available(fn: F) -> F
Implement a decorator to execute a function only if torch
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import torch_available
>>> @torch_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.torch_numpy_available ¶
torch_numpy_available(fn: F) -> F
Implement a decorator to execute a function only if torch and
numpy packages are installed and are compatible.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import torch_numpy_available
>>> @torch_numpy_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.imports.xarray_available ¶
xarray_available(fn: F) -> F
Implement a decorator to execute a function only if xarray
package is installed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fn
|
F
|
The function to execute. |
required |
Returns:
| Type | Description |
|---|---|
F
|
A wrapper around |
Example
>>> from coola.utils.imports import xarray_available
>>> @xarray_available
... def my_function(n: int = 0) -> int:
... return 42 + n
...
>>> my_function()
coola.utils.introspection ¶
Utilities for introspecting Python objects and retrieving their fully qualified names.
coola.utils.introspection.get_fully_qualified_name ¶
get_fully_qualified_name(obj: object) -> str
Return the fully qualified name of a Python object.
Supports functions, classes, methods, and instances. For instances, returns the fully qualified class name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
obj
|
object
|
The object whose name is to be computed. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The fully qualified name. |
Example
>>> from coola.utils.introspection import get_fully_qualified_name
>>> import collections
>>> get_fully_qualified_name(collections.Counter)
'collections.Counter'
>>> class MyClass:
... pass
...
>>> get_fully_qualified_name(MyClass)
'coola.utils.introspection.MyClass'
>>> get_fully_qualified_name(map)
'builtins.map'
coola.utils.mapping ¶
Contain utility functions for mappings.
coola.utils.mapping.sort_by_keys ¶
sort_by_keys(mapping: Mapping[Any, Any]) -> dict[Any, Any]
Sort a dictionary by keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The dictionary to sort. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[Any, Any]
|
The sorted dictionary. |
Example
>>> from coola.utils.mapping import sort_by_keys
>>> sort_by_keys({"dog": 1, "cat": 5, "fish": 2})
{'cat': 5, 'dog': 1, 'fish': 2}
coola.utils.mapping.sort_by_values ¶
sort_by_values(
mapping: Mapping[Any, Any],
) -> dict[Any, Any]
Sort a dictionary by keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[Any, Any]
|
The dictionary to sort. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[Any, Any]
|
The sorted dictionary. |
Example
>>> from coola.utils.mapping import sort_by_values
>>> sort_by_values({"dog": 1, "cat": 5, "fish": 2})
{'dog': 1, 'fish': 2, 'cat': 5}
coola.utils.password ¶
Implement utility functions to securely get passwords from users.
coola.utils.password.get_password ¶
get_password(
username: str | None = None, *, confirm: bool = False
) -> str
Securely prompt for and retrieve a password from the user.
This function uses getpass to securely prompt for password input without echoing characters to the terminal. The password undergoes validation for length constraints before being returned.
Security features
- No password echoing to terminal
- Length validation (non-empty, max bound)
- Optional confirmation to prevent typos
- Prevents use in non-interactive contexts
- No logging or storage of password values
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
username
|
str | None
|
Optional username for prompt clarity. |
None
|
confirm
|
bool
|
Require password confirmation if True. |
False
|
Returns:
| Type | Description |
|---|---|
str
|
The password entered by the user (whitespace preserved). |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If not running in an interactive terminal. |
(KeyboardInterrupt, EOFError)
|
If input is interrupted. |
ValueError
|
If password validation fails. |
coola.utils.path ¶
Contain path utility functions.
coola.utils.path.sanitize_path ¶
sanitize_path(path: Path | str) -> Path
Sanitize the given path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path | str
|
The path to sanitize. The path can be a string or a
|
required |
Returns:
| Type | Description |
|---|---|
Path
|
The sanitized path as a |
Example
>>> from pathlib import Path
>>> from coola.utils.path import sanitize_path
>>> sanitize_path("something")
PosixPath('.../something')
>>> sanitize_path("")
PosixPath('...')
>>> sanitize_path(Path("something"))
PosixPath('.../something')
>>> sanitize_path(Path("something/./../"))
PosixPath('...')
coola.utils.path.working_directory ¶
working_directory(path: Path) -> Generator[None]
Context manager to change the working directory to the given path, and then changes it back to its previous value on exit.
source: https://gist.github.com/nottrobin/3d675653244f8814838a
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path
|
The path to the temporary working directory. |
required |
Example
>>> from coola.utils.path import working_directory
>>> with working_directory(Path("src")):
... x = 1
...
coola.utils.secret ¶
Define utility functions for secrets.
coola.utils.secret.mask_secret ¶
mask_secret(
secret: str, show_first: int = 3, show_last: int = 4
) -> str
Mask the content of the secret with *.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
secret
|
str
|
The secret to mask. |
required |
show_first
|
int
|
The number of first values to show. |
3
|
show_last
|
int
|
The number of last values to show. |
4
|
Returns:
| Type | Description |
|---|---|
str
|
The masked secret. |
Example
>>> from coola.utils.secret import mask_secret
>>> secret = "abcdefghijklmnopqrstuvwxyz"
>>> mask_secret(secret)
'abc*******************wxyz'
>>> mask_secret(secret, show_first=0)
'**********************wxyz'
>>> mask_secret(secret, show_last=0)
'abc***********************'
>>> mask_secret(secret, show_first=0, show_last=0)
'**************************'
coola.utils.stats ¶
Implement some utility functions to compute statistics.
coola.utils.stats.quantile ¶
quantile(
values: Sequence[float | int],
quantiles: Sequence[float],
) -> list[float]
Compute the quantiles with the linear method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
values
|
Sequence[float | int]
|
The values. |
required |
quantiles
|
Sequence[float]
|
The quantile values in the
range |
required |
Returns:
| Type | Description |
|---|---|
list[float]
|
The quantiles. |
Example
>>> from coola.utils.stats import quantile
>>> quantile([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10], (0.2, 0.5, 0.9))
[2.0, 5.0, 9.0]
coola.utils.string ¶
Contain utility functions to process strings.
coola.utils.string.remove_empty_lines ¶
remove_empty_lines(text: str) -> str
Remove empty lines from a string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The input string from which empty lines will be removed. |
required |
Returns:
| Type | Description |
|---|---|
str
|
A new string with all empty or whitespace-only lines removed. |
Example
>>> from coola.utils.string import remove_empty_lines
>>> remove_empty_lines("Hello\n\nWorld\n\n\nFoo")
'Hello\nWorld\nFoo'
>>> remove_empty_lines("\n\nOnly empty lines\n\n")
'Only empty lines'
coola.utils.tensor ¶
Implement some utility functions for torch.Tensors.
coola.utils.tensor.get_available_devices
cached
¶
get_available_devices() -> tuple[str, ...]
Get the available PyTorch devices on the machine.
Returns:
| Type | Description |
|---|---|
tuple[str, ...]
|
The available devices. |
Example
>>> from coola.utils.tensor import get_available_devices
>>> get_available_devices()
('cpu'...)
coola.utils.tensor.is_cuda_available
cached
¶
is_cuda_available() -> bool
Indicate if CUDA is currently available.
Returns:
| Type | Description |
|---|---|
bool
|
A boolean indicating if CUDA is currently available. |
Example
>>> from coola.utils.tensor import is_cuda_available
>>> is_cuda_available()
coola.utils.tensor.is_mps_available
cached
¶
is_mps_available() -> bool
Indicate if MPS is currently available.
Returns:
| Type | Description |
|---|---|
bool
|
A boolean indicating if MPS is currently available. |
Example
>>> from coola.utils.tensor import is_mps_available
>>> is_mps_available()
coola.utils.tensor.to_tensor ¶
to_tensor(
data: Sequence[int | float] | Tensor | ndarray,
) -> Tensor
Convert the input to a torch.Tensor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
Sequence[int | float] | Tensor | ndarray
|
The data to convert to a tensor. |
required |
Returns:
| Type | Description |
|---|---|
Tensor
|
A tensor. |
Example
>>> from coola.utils.tensor import to_tensor
>>> x = to_tensor([1, 2, 3, 4, 5])
>>> x
tensor([1, 2, 3, 4, 5])
>>> import numpy as np
>>> x = to_tensor(np.array([1, 2, 3, 4, 5]))
>>> x
tensor([1, 2, 3, 4, 5])
coola.utils.today ¶
Contain utility functions for today.
coola.utils.today.get_today_date ¶
get_today_date(
timezone: str = "UTC", date_format: str = "%Y-%m-%d"
) -> str
Return the current date for the specified timezone.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timezone
|
str
|
The IANA timezone name to use when determining
the current date. Defaults to |
'UTC'
|
date_format
|
str
|
A :func: |
'%Y-%m-%d'
|
Returns:
| Type | Description |
|---|---|
str
|
A string representing the current date in the given timezone,
formatted according to |
Example
>>> from coola.utils.today import get_today_date
>>> date = get_today_date()
>>> isinstance(date, str)
True
>>> date = get_today_date(timezone="America/New_York", date_format="%d/%m/%Y")
>>> isinstance(date, str)
True
coola.utils.version ¶
Contain functions to manage package versions.
coola.utils.version.compare_version ¶
compare_version(
package: str, op: Callable, version: str
) -> bool
Compare a package version to a given version.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
package
|
str
|
The package to check. |
required |
op
|
Callable
|
The comparison operator. |
required |
version
|
str
|
The version to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
The comparison status. |
Example
>>> import operator
>>> from coola.utils.version import compare_version
>>> compare_version("pytest", op=operator.ge, version="7.3.0")
True
coola.utils.version.get_package_version
cached
¶
get_package_version(package: str) -> Version | None
Get the package version.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
package
|
str
|
The package name. |
required |
Returns:
| Type | Description |
|---|---|
Version | None
|
The package version. |
Example
>>> from coola.utils.version import get_package_version
>>> get_package_version("pytest")
<Version('...')>