blueetl.utils¶

Common utilities.

Functions

`all_equal`(iterable)	Return True if all the items of the given iterable are equal, False otherwise.
`checksum`(filepath[, chunk_size])	Calculate and return the checksum of the given file.
`checksum_json`(obj)	Calculate and return the checksum of the given object converted to json.
`checksum_str`(s)	Calculate and return the checksum of the given string.
`copy_config`(src, dst)	Copy the analysis configuration file to a different location.
`dict_product`(d)	Iterate over the product of the values of the given dict.
`dump_json`(filepath, data, *[, encoding, indent])	Dump to JSON file.
`dump_yaml`(filepath, data, **kwargs)	Dump to YAML file.
`ensure_dtypes`(df[, desired_dtypes])	Return a DataFrame with the columns and index cast to the desired types.
`ensure_list`(x)	Always return a list from the given argument.
`extract_items`(obj[, path])	Yield tuples (path, item) from the traversal of the given dict.
`get_shmdir`()	Return the shared memory directory, or return None if not set.
`import_by_string`(full_name)	Import and return a function by name.
`import_optional_dependency`(name)	Import an optional dependency.
`load_json`(filepath, *[, encoding])	Load from JSON file.
`load_yaml`(filepath)	Load from YAML file.
`relpath`(path, start)	Return a relative filepath to path from the start directory.
`resolve_path`(*paths[, symlinks])	Make the path absolute and return a new path object.
`setup_logging`(loglevel[, logformat])	Setup logging.
`timed`(log, msg)	Context manager to log the execution time using the specified logger function.

Classes

CachedPropertyMixIn()

MixIn to be used with classes using cached_property to be skipped when pickled.

class blueetl.utils.CachedPropertyMixIn¶

Bases: object

MixIn to be used with classes using cached_property to be skipped when pickled.

blueetl.utils.all_equal(iterable: Iterable) → bool¶

Return True if all the items of the given iterable are equal, False otherwise.

Notes: - The items doesn’t need to be hashable, because they are compared with equality. - The function returns True also when the iterable doesn’t contain any item.

blueetl.utils.checksum(filepath: str | PathLike, chunk_size: int = 65536) → str¶: Calculate and return the checksum of the given file.

blueetl.utils.checksum_json(obj: Any) → str¶: Calculate and return the checksum of the given object converted to json.

blueetl.utils.checksum_str(s: str) → str¶: Calculate and return the checksum of the given string.

blueetl.utils.copy_config(src: str | PathLike, dst: str | PathLike) → None¶

Copy the analysis configuration file to a different location.

If the simulation_campaign path is relative, then it’s resolved relatively to the directory containing the original configuration file.

The output path, instead, is not resolved even when it’s relative.

Note that any comment present in the original file is not preserved.

blueetl.utils.dict_product(d: dict) → Iterator[tuple]¶

Iterate over the product of the values of the given dict.

Yield tuples of tuples, where each item is composed by:

key
value
index of the key in the original list

Example

Passing the dictionary:

d = {"a": ["a1", "a2"], "b": ["b1", "b2"]}

Yields tuples from:

[
    (("a", "a1", 0), ("b", "b1", 0)),
    (("a", "a1", 0), ("b", "b2", 1)),
    (("a", "a2", 1), ("b", "b1", 0)),
    (("a", "a2", 1), ("b", "b2", 1)),
]

blueetl.utils.dump_json(filepath: str | PathLike, data: Any, *, encoding: str = 'utf-8', indent: int = 2, **kwargs) → None¶: Dump to JSON file.

blueetl.utils.dump_yaml(filepath: str | PathLike, data: Any, **kwargs) → None¶: Dump to YAML file.

blueetl.utils.ensure_dtypes(df: DataFrame, desired_dtypes: dict[str, Any] | None = None) → DataFrame¶

Return a DataFrame with the columns and index cast to the desired types.

Parameters:

df – original Pandas DataFrame.
desired_dtypes – dict of names and desired dtypes. If None, the predefined dtypes are used. If the dict contains names not present in the columns or in the index, they are ignored.

Returns:

The same DataFrame is modified and returned, if it’s possible to avoid making a copy.

blueetl.utils.ensure_list(x: Any) → list¶: Always return a list from the given argument.

blueetl.utils.extract_items(obj: dict[str, Any], path: str | None = None) → Iterator[tuple[str, Any]]¶

Yield tuples (path, item) from the traversal of the given dict.

Each yielded path is obtained from the concatenation of the nested keys, separated by “.”.

All the keys of the dicts are expected to be strings, not containing “.”.

For examples, iterating over the result of:

extract_items(
    {
        "latency": {"params": {"onset": False}},
        "decay": {"params": {"ratio": [0.25, 0.5, 0.75]}},
        "baseline_PSTH": {"params": {"bin_size": 0.5, "sigma": 0, "offset": -6}},
    }
)

would yield the following tuples:

("latency.params.onset", False)
("decay.params.ratio",  [0.25, 0.5, 0.75])
("baseline_PSTH.params.bin_size", 0.5)
("baseline_PSTH.params.sigma", 0)
("baseline_PSTH.params.offset", -6)

blueetl.utils.get_shmdir() → Path | None¶: Return the shared memory directory, or return None if not set.

blueetl.utils.import_by_string(full_name: str) → Callable¶

Import and return a function by name.

Parameters:: full_name – full name of the function, using dot as a separator if in a submodule.
Returns:: The imported function.

blueetl.utils.import_optional_dependency(name: str) → Any¶

Import an optional dependency.

If a dependency is missing, an ImportError with a custom message is raised. Based on: https://github.com/pandas-dev/pandas/blob/0d853e77/pandas/compat/_optional.py#L85

Parameters:: name – The module name.
Returns:: The imported module, if found.
Return type:: ModuleType

blueetl.utils.load_json(filepath: str | PathLike, *, encoding: str = 'utf-8', **kwargs) → Any¶: Load from JSON file.

blueetl.utils.load_yaml(filepath: str | PathLike) → Any¶: Load from YAML file.

blueetl.utils.relpath(path: str | PathLike, start: str | PathLike) → Path¶

Return a relative filepath to path from the start directory.

In Python>=3.12 it would be possible to use Path.relative_to with walk_up=True.

blueetl.utils.resolve_path(*paths: str | PathLike, symlinks: bool = False) → Path¶

Make the path absolute and return a new path object.

It may be different from calling Path.resolve(), because Path.resolve() always resolve symlinks. It may be different from calling Path.absolute(), because Path.absolute() doesn’t remove the relative paths. For example, Path(‘/tmp/..’).absolute() == PosixPath(‘/tmp/..’).

blueetl.utils.setup_logging(loglevel: int | str, logformat: str | None = None, **logparams) → None¶: Setup logging.

blueetl.utils.timed(log: Callable[[str], None], msg: str) → Iterator[list[str]]¶

Context manager to log the execution time using the specified logger function.

The yielded list of messages can be used to replace the existing message, or inject additional strings to the final log.