anemoi.utils package

Anemoi Utils package.

Subpackages

Submodules

anemoi.utils.caching module

anemoi.utils.caching.clean_cache(collection: str = 'default') → None

Clean the cache for a collection.

This removes all cached data files and their associated lock files.

Parameters:: collection (str, optional) – The name of the collection, by default “default”

class anemoi.utils.caching.Cacher(collection: str, expires: int | None)

Bases: object

This class implements a simple caching mechanism. Private class, do not use directly.

cache(key: tuple, proc: Callable) → Any

Cache the result of a function.

Parameters:

key (tuple) – The cache key
proc (Callable) – The function to call if the result is not cached

Returns:

The cached result

Return type:

Any

class anemoi.utils.caching.JsonCacher(collection: str, expires: int | None)

Bases: Cacher

Cacher that uses JSON files.

ext = ''

save(path: str, data: dict) → str

Save data to a JSON file.

Parameters:

path (str) – The path to the JSON file
data (dict) – The data to save

Returns:

The temporary file path

Return type:

str

load(path: str) → dict

Load data from a JSON file.

Parameters:: path (str) – The path to the JSON file
Returns:: The loaded data
Return type:: dict

class anemoi.utils.caching.NpzCacher(collection: str, expires: int | None)

Bases: Cacher

Cacher that uses NPZ files.

ext = '.npz'

save(path: str, data: dict) → str

Save data to an NPZ file.

Parameters:

path (str) – The path to the NPZ file
data (dict) – The data to save

Returns:

The temporary file path

Return type:

str

load(path: str) → dict

Load data from an NPZ file.

Parameters:: path (str) – The path to the NPZ file
Returns:: The loaded data
Return type:: dict

anemoi.utils.caching.cached(collection: str = 'default', expires: int | None = None, encoding: str = 'json') → Callable

Decorator to cache the result of a function.

Default is to use a json file to store the cache, but you can also use npz files to cache dict of numpy arrays.

Parameters:

collection (str, optional) – The name of the collection, by default “default”
expires (int, optional) – The expiration time in seconds, or None for no expiration, by default None
encoding (str, optional) – The encoding type, either “json” or “npz”, by default “json”

Returns:

The decorated function

Return type:

Callable

anemoi.utils.checkpoints module

Read and write extra metadata in PyTorch checkpoints files. These files are zip archives containing the model weights.

anemoi.utils.checkpoints.has_metadata(path: str, *, name: str = 'anemoi.json') → bool

Check if a checkpoint file has a metadata file.

Parameters:

path (str) – The path to the checkpoint file
name (str, optional) – The name of the metadata file in the zip archive

Returns:

True if the metadata file is found

Return type:

bool

anemoi.utils.checkpoints.get_metadata_path(path: str, *, name: str = 'anemoi.json') → str

Get the full path of the metadata file in the checkpoint.

Parameters:

path (str) – The path to the checkpoint file
name (str, optional) – The name of the metadata file in the zip archive

Returns:

The full path of the metadata file in the zip archive

Return type:

str

Raises:

FileNotFoundError – If the metadata file is not found
ValueError – If multiple metadata files are found

anemoi.utils.checkpoints.load_metadata(path: str, *, supporting_arrays: Literal[False] = False, name: str = DEFAULT_NAME) → dict

anemoi.utils.checkpoints.load_metadata(path: str, *, supporting_arrays: Literal[True] = True, name: str = DEFAULT_NAME) → tuple[dict, dict]

Load metadata from a checkpoint file.

Parameters:

path (str) – The path to the checkpoint file
supporting_arrays (bool, optional) – If True, the function will return a dictionary with the supporting arrays
name (str, optional) – The name of the metadata file in the zip archive

Returns:

The content of the metadata file from JSON

Return type:

dict

Raises:

ValueError – If the metadata file is not found

anemoi.utils.checkpoints.load_supporting_arrays(zipf: ZipFile, entries: dict) → dict

Load supporting arrays from a zip file.

Parameters:

zipf (zipfile.ZipFile) – The zip file
entries (dict) – A dictionary of entries with paths, shapes, and dtypes

Returns:

A dictionary of supporting arrays

Return type:

dict

anemoi.utils.checkpoints.save_metadata(path: str, metadata: dict, *, supporting_arrays: dict | None = None, name: str = 'anemoi.json', folder: str = 'anemoi-metadata') → None

Save metadata to a checkpoint file.

Parameters:

path (str) – The path to the checkpoint file
metadata (dict) – A JSON serializable object
supporting_arrays (dict | None, optional) – A dictionary of supporting NumPy arrays
name (str, optional) – The name of the metadata file in the zip archive
folder (str, optional) – The folder where the metadata file will be saved

anemoi.utils.checkpoints.replace_metadata(path: str, metadata: dict, supporting_arrays: dict | None = None, *, name: str = 'anemoi.json') → None

Replace metadata in a checkpoint file.

Parameters:

path (str) – The path to the checkpoint file
metadata (dict) – A JSON serializable object
supporting_arrays (dict, optional) – A dictionary of supporting NumPy arrays
name (str, optional) – The name of the metadata file in the zip archive

anemoi.utils.checkpoints.remove_metadata(path: str, *, name: str = 'anemoi.json') → None

Remove metadata from a checkpoint file.

Parameters:

path (str) – The path to the checkpoint file
name (str, optional) – The name of the metadata file in the zip archive

anemoi.utils.cli module

class anemoi.utils.cli.Command

Bases: object

Base class for commands.

accept_unknown_args = False

check(parser: ArgumentParser, args: Namespace) → None: Check the command arguments.

run(args: Namespace) → None

Run the command.

Parameters:: args (argparse.Namespace) – The arguments for the command

anemoi.utils.cli.make_parser(description: str, commands: dict[str, Command]) → ArgumentParser

Create an argument parser for the CLI.

Parameters:

description (str) – The description of the CLI
commands (dict[str, Command]) – A dictionary of command names to Command instances

Returns:

The argument parser

Return type:

argparse.ArgumentParser

class anemoi.utils.cli.Failed(name: str, error: ImportError)

Bases: Command

Command not available.

add_arguments(command_parser: ArgumentParser) → None

Add arguments to the command parser.

Parameters:: command_parser (argparse.ArgumentParser) – The command parser

run(args: Namespace) → None

Run the command.

Parameters:: args (argparse.Namespace) – The arguments for the command

anemoi.utils.cli.register_commands(here: str, package: str, select: Callable, fail: Callable = None) → dict[str, Command]

Register commands from a package.

Parameters:

here (str) – The directory containing the commands
package (str) – The package name
select (Callable) – A function to select the command object from the module
fail (Callable, optional) – A function to create a Failed command if a command cannot be imported

Returns:

A dictionary of command names to Command instances

Return type:

dict[str, Command]

anemoi.utils.cli.cli_main(version: str, description: str, commands: dict[str, Command], test_arguments: list[str] | None = None) → None

Main entry point for the CLI.

Parameters:

version (str) – The version of the CLI
description (str) – The description of the CLI
commands (dict[str, Command]) – A dictionary of command names to Command instances
test_arguments (list[str], optional) – The command line arguments to parse, used for testing purposes, by default None

anemoi.utils.compatibility module

anemoi.utils.compatibility.aliases(aliases: dict[str, str | list[str]] | None = None, **kwargs: Any) → Callable[[Callable], Callable]

Alias keyword arguments in a function call.

Allows for dynamically renaming keyword arguments in a function call.

Parameters:

aliases (dict[str, Union[str, list[str]]], optional) – Key, value pair of aliases, with keys being the true name, and value being a str or list of aliases, by default None
**kwargs (Any) – Kwargs form of aliases

Returns:

Decorator function that renames keyword arguments in a function call.

Return type:

Callable

Raises:

ValueError – If the aliasing would result in duplicate keys.

Examples

```python @aliases(a=”b”, c=[“d”, “e”]) def func(a, c):

return a, c

func(a=1, c=2) # (1, 2) func(b=1, d=2) # (1, 2) ```

anemoi.utils.config module

class anemoi.utils.config.DotDict(*args, **kwargs)

Bases: dict

A dictionary that allows access to its keys as attributes.

>>> d = DotDict({"a": 1, "b": {"c": 2}})
>>> d.a
1
>>> d.b.c
2
>>> d.b = 3
>>> d.b
3

The class is recursive, so nested dictionaries are also DotDicts.

The DotDict class has the same constructor as the dict class.

>>> d = DotDict(a=1, b=2)

static convert_to_nested_dot_dict(value: Any) → Any

Convert nested dicts to DotDict recursively.

Parameters:: value (Any) – The value to convert
Returns:: Converted value with nested dicts as DotDict
Return type:: Any

classmethod from_file(path: str) → DotDict

Create a DotDict from a file.

Parameters:: path (str) – The path to the file.
Returns:: The created DotDict.
Return type:: DotDict

classmethod from_yaml_file(path: str) → DotDict

Create a DotDict from a YAML file.

Parameters:: path (str) – The path to the YAML file.
Returns:: The created DotDict.
Return type:: DotDict

classmethod from_json_file(path: str) → DotDict

Create a DotDict from a JSON file.

Parameters:: path (str) – The path to the JSON file.
Returns:: The created DotDict.
Return type:: DotDict

classmethod from_toml_file(path: str) → DotDict

Create a DotDict from a TOML file.

Parameters:: path (str) – The path to the TOML file.
Returns:: The created DotDict.
Return type:: DotDict

anemoi.utils.config.is_omegaconf_dict(value: Any) → bool

Check if a value is an OmegaConf DictConfig.

Parameters:: value (Any) – The value to check.
Returns:: True if the value is a DictConfig, False otherwise.
Return type:: bool

anemoi.utils.config.is_omegaconf_list(value: Any) → bool

Check if a value is an OmegaConf ListConfig.

Parameters:: value (Any) – The value to check.
Returns:: True if the value is a ListConfig, False otherwise.
Return type:: bool

anemoi.utils.config.config_path(name: str = 'settings.toml') → str

Get the path to a configuration file.

Parameters:: name (str, optional) – The name of the configuration file, by default “settings.toml”.
Returns:: The path to the configuration file.
Return type:: str

anemoi.utils.config.load_any_dict_format(path: str) → dict

Load a configuration file in any supported format: JSON, YAML and TOML.

Parameters:: path (str) – The path to the configuration file.
Returns:: The decoded configuration file.
Return type:: dict

anemoi.utils.config.save_config(name: str, data: Any) → None

Save a configuration file.

Parameters:

name (str) – The name of the configuration file to save.
data (Any) – The data to save.

Read a configuration file.

Parameters:

name (str, optional) – The name of the config file to read, by default “settings.toml”
secrets (str or list, optional) – The name of the secrets file, by default None
defaults (str or dict, optional) – The name of the defaults file, by default None

Returns:

Return DotDict if it is a dictionary, otherwise the raw data

Return type:

DotDict or str

anemoi.utils.config.load_raw_config(name: str, default: Any = None) → DotDict | str

Load a raw configuration file.

Parameters:

name (str) – The name of the configuration file.
default (Any, optional) – The default value if the file does not exist, by default None.

Returns:

The loaded configuration or the default value.

Return type:

DotDict or str

anemoi.utils.config.check_config_mode(name: str = 'settings.toml', secrets_name: str = None, secrets: list[str] = None) → None

Check that a configuration file is secure.

Parameters:

name (str, optional) – The name of the configuration file, by default “settings.toml”
secrets_name (str, optional) – The name of the secrets file, by default None
secrets (list, optional) – The list of secrets to check, by default None

Raises:

SystemError – If the configuration file is not secure.

anemoi.utils.config.find(metadata: dict | list, what: str, result: list = None, *, select: callable = None) → list

Find all occurrences of a key in a nested dictionary or list with an optional selector.

Parameters:

metadata (dict or list) – The metadata to search.
what (str) – The key to search for.
result (list, optional) – The list to store results, by default None.
select (callable, optional) – A function to filter the results, by default None.

Returns:

The list of found values.

Return type:

list

anemoi.utils.config.merge_configs(*configs: dict) → dict

Merge multiple configuration dictionaries.

Parameters:: *configs (dict) – The configuration dictionaries to merge.
Returns:: The merged configuration dictionary.
Return type:: dict

anemoi.utils.config.temporary_config(tmp: dict) → Generator[None, None, None]

anemoi.utils.dates module

anemoi.utils.dates.normalise_frequency(frequency: int | str) → int

Normalise frequency to hours.

Parameters:: frequency (int or str) – The frequency to normalise.
Returns:: The normalised frequency in hours.
Return type:: int

anemoi.utils.dates.as_datetime(date: date | datetime | str, keep_time_zone: bool = False) → datetime

Convert a date to a datetime object, removing any time zone information.

Parameters:

date (datetime.date or datetime.datetime or str) – The date to convert.
keep_time_zone (bool, optional) – If True, the time zone information is kept, by default False.

Returns:

The datetime object.

Return type:

datetime.datetime

anemoi.utils.dates.as_datetime_list(date: date | datetime | str, default_increment: int = 1) → list[datetime]

Convert a date to a list of datetime objects.

Parameters:

date (datetime.date or datetime.datetime or str) – The date to convert.
default_increment (int, optional) – The default increment in hours, by default 1.

Returns:

A list of datetime objects.

Return type:

list of datetime.datetime

anemoi.utils.dates.as_timedelta(frequency: int | str | timedelta) → timedelta

Convert anything to a timedelta object.

Parameters:

frequency (int or str or datetime.timedelta) –

The frequency to convert. If an integer, it is assumed to be in hours. If a string, it can be in the format:

”1h” for 1 hour
”1d” for 1 day
”1m” for 1 minute
”1s” for 1 second
”1:30” for 1 hour and 30 minutes
”1:30:10” for 1 hour, 30 minutes and 10 seconds
”PT10M” for 10 minutes (ISO8601)

If a timedelta object is provided, it is returned as is.

Returns:

The timedelta object.

Return type:

datetime.timedelta

Raises:

ValueError – Exception raised if the frequency cannot be converted to a timedelta.

anemoi.utils.dates.frequency_to_timedelta(frequency: int | str | timedelta) → timedelta

Convert a frequency to a timedelta object.

Parameters:: frequency (int or str or datetime.timedelta) – The frequency to convert.
Returns:: The timedelta object.
Return type:: datetime.timedelta

anemoi.utils.dates.frequency_to_string(frequency: timedelta) → str

Convert a frequency (i.e. a datetime.timedelta) to a string.

Parameters:: frequency (datetime.timedelta) – The frequency to convert.
Returns:: A string representation of the frequency.
Return type:: str

anemoi.utils.dates.frequency_to_seconds(frequency: int | str | timedelta) → int

Convert a frequency to seconds.

Parameters:: frequency (int or str or datetime.timedelta) – The frequency to convert.
Returns:: Number of seconds.
Return type:: int

Bases: object

The DateTimes class is an iterator that generates datetime objects within a given range.

class anemoi.utils.dates.Year(year: int, **kwargs)

Bases: DateTimes

Year is defined as the months of January to December.

class anemoi.utils.dates.Winter(year: int, **kwargs)

Bases: DateTimes

Winter is defined as the months of December, January and February.

class anemoi.utils.dates.Spring(year: int, **kwargs)

Bases: DateTimes

Spring is defined as the months of March, April and May.

class anemoi.utils.dates.Summer(year: int, **kwargs)

Bases: DateTimes

Summer is defined as the months of June, July and August.

class anemoi.utils.dates.Autumn(year: int, **kwargs)

Bases: DateTimes

Autumn is defined as the months of September, October and November.

class anemoi.utils.dates.ConcatDateTimes(*dates: DateTimes)

Bases: object

ConcatDateTimes is an iterator that generates datetime objects from a list of dates.

class anemoi.utils.dates.EnumDateTimes(dates: list[date | datetime | str])

Bases: object

EnumDateTimes is an iterator that generates datetime objects from a list of dates.

anemoi.utils.dates.datetimes_factory(*args: Any, **kwargs: Any) → DateTimes | ConcatDateTimes | EnumDateTimes

Create a DateTimes, ConcatDateTimes, or EnumDateTimes object.

Parameters:

*args (Any) – Positional arguments.
**kwargs (Any) – Keyword arguments.

Returns:

The created object.

Return type:

DateTimes or ConcatDateTimes or EnumDateTimes

anemoi.utils.devtools module

anemoi.utils.grib module

Utilities for working with GRIB parameters.

See https://codes.ecmwf.int/grib/param-db/ for more information.

class anemoi.utils.grib.LazyConfig

Bases: object

property default_origin: str: Default origin to use when disambiguating parameters with the same shortname.

property cache_length: int: Cache length in seconds for GRIB parameter lookups.

property local_cache: str | None: Path to a local cache file for GRIB parameters. If set, this will be used instead of making API calls.

anemoi.utils.grib.shortname_to_paramid(shortname: str, **filters) → int

Return the GRIB parameter id given its shortname.

Parameters:

shortname (str) – Parameter shortname.
filters (Any) – Additional filters to disambiguate parameters with the same shortname (e.g. origin, encoding, table, discipline, category).

Returns:

int – Parameter id.
>>> shortname_to_paramid(“2t”)
167

anemoi.utils.grib.paramid_to_shortname(paramid: int, **filters) → str

Return the shortname of a GRIB parameter given its id.

Parameters:

paramid (int) – Parameter id.
filters (Any) – Additional filters to disambiguate parameters with the same shortname (e.g. origin, encoding, table, discipline, category).

Returns:

str – Parameter shortname.
>>> paramid_to_shortname(167)
’2t’

anemoi.utils.grib.units(param: int | str) → str

Return the units of a GRIB parameter given its name or id.

Parameters:

param (int or str) – Parameter id or name.

Returns:

str – Parameter unit.
>>> unit(167)
’K’

anemoi.utils.grib.must_be_positive(param: int | str) → bool

Check if a parameter must be positive.

Parameters:

param (int or str) – Parameter id or shortname.

Returns:

bool – True if the parameter must be positive.
>>> must_be_positive(“tp”)
True

anemoi.utils.grids module

Utilities for working with grids.

anemoi.utils.grids.xyz_to_latlon(x: ndarray, y: ndarray, z: ndarray) → tuple[ndarray, ndarray]

Convert Cartesian coordinates to latitude and longitude.

Parameters:

x (np.ndarray) – The x coordinates
y (np.ndarray) – The y coordinates
z (np.ndarray) – The z coordinates

Returns:

The latitude and longitude

Return type:

tuple[np.ndarray, np.ndarray]

Deprecated since version 0.4.25: This will be removed in 0.5.0. Use anemoi.transform.spatial.xyz_to_latlon instead.

anemoi.utils.grids.latlon_to_xyz(lat: ndarray, lon: ndarray, radius: float = 1.0) → tuple[ndarray, ndarray, ndarray]

Convert latitude and longitude to Cartesian coordinates.

Parameters:

lat (np.ndarray) – The latitudes
lon (np.ndarray) – The longitudes
radius (float, optional) – The radius of the sphere, by default 1.0

Returns:

The x, y, and z coordinates

Return type:

tuple[np.ndarray, np.ndarray, np.ndarray]

Deprecated since version 0.4.25: This will be removed in 0.5.0. Use anemoi.transform.spatial.xyz_to_latlon instead.

anemoi.utils.grids.nearest_grid_points(source_latitudes: ndarray, source_longitudes: ndarray, target_latitudes: ndarray, target_longitudes: ndarray) → ndarray

Find the nearest grid points.

Parameters:

source_latitudes (np.ndarray) – The source latitudes
source_longitudes (np.ndarray) – The source longitudes
target_latitudes (np.ndarray) – The target latitudes
target_longitudes (np.ndarray) – The target longitudes

Returns:

The indices of the nearest grid points

Return type:

np.ndarray

Deprecated since version 0.4.25: This will be removed in 0.5.0. Use anemoi.transform.spatial.nearest_grid_points instead.

anemoi.utils.grids.grids(name: str | list[float] | tuple[float, ...]) → dict

Load grid data by name.

Parameters:: name (str) – The name of the grid
Returns:: The grid data
Return type:: dict

Deprecated since version 0.4.25: This will be removed in 0.5.0. Use anemoi.transform.grids.named.lookup instead.

anemoi.utils.hindcasts module

class anemoi.utils.hindcasts.HindcastDatesTimes(reference_dates: list[datetime], years: int = 20)

Bases: object

The HindcastDatesTimes class is an iterator that generates datetime objects within a given range.

reference_dates

List of reference dates.

Type:: List[datetime.datetime]

years

Number of years to go back from each reference date.

Type:: int

anemoi.utils.humanize module

Generate human readable strings.

anemoi.utils.humanize.bytes_to_human(n: float) → str

Convert a number of bytes to a human readable string.

>>> bytes_to_human(4096)
'4 KiB'

>>> bytes_to_human(4000)
'3.9 KiB'

Parameters:: n (float) – The number of bytes
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.bytes(n: float) → str

Deprecated function to convert bytes to a human readable string.

Parameters:: n (float) – The number of bytes
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.base2_to_human(n: float) → str

Convert a number to a human readable string using base 2 units.

Parameters:: n (float) – The number to convert
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.base2(n: float) → str

Deprecated function to convert a number to a human readable string using base 2 units.

Parameters:: n (float) – The number to convert
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.seconds_to_human(seconds: float | timedelta) → str

Convert a number of seconds to a human readable string.

>>> seconds_to_human(4000)
'1 hour 6 minutes 40 seconds'

Parameters:: seconds (float) – The number of seconds
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.seconds(seconds: float) → str

Deprecated function to convert seconds to a human readable string.

Parameters:: seconds (float) – The number of seconds
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.plural(value: int, what: str) → str

Return a string with the value and the pluralized form of what.

Parameters:

value (int) – The value
what (str) – The string to pluralize

Returns:

The value and the pluralized form of what

Return type:

str

anemoi.utils.humanize.when(then: datetime, now: datetime | None = None, short: bool = True, use_utc: bool = False) → str

Generate a human readable string for a date, relative to now.

>>> when(datetime.datetime.now() - datetime.timedelta(hours=2))
'2 hours ago'

>>> when(datetime.datetime.now() - datetime.timedelta(days=1))
'yesterday at 08:46'

>>> when(datetime.datetime.now() - datetime.timedelta(days=5))
'last Sunday'

>>> when(datetime.datetime.now() - datetime.timedelta(days=365))
'last year'

>>> when(datetime.datetime.now() + datetime.timedelta(days=365))
'next year'

Parameters:

then (datetime.datetime) – A datetime
now (datetime.datetime, optional) – The reference date, by default NOW
short (bool, optional) – Generate shorter strings, by default True
use_utc (bool, optional) – Use UTC time, by default False

Returns:

A human readable string

Return type:

str

anemoi.utils.humanize.string_distance(s: str, t: str) → int

Calculate the Levenshtein distance between two strings.

Parameters:

s (str) – The first string
t (str) – The second string

Returns:

The Levenshtein distance

Return type:

int

anemoi.utils.humanize.did_you_mean(word: str, vocabulary: list[str]) → str

Pick the closest word in a vocabulary.

>>> did_you_mean("aple", ["banana", "lemon", "apple", "orange"])
'apple'

Parameters:

word (str) – The word to look for
vocabulary (list of str) – The list of known words

Returns:

The closest word in the vocabulary

Return type:

str

anemoi.utils.humanize.dict_to_human(query: dict[str, Any]) → str

Convert a dictionary to a human readable string.

Parameters:: query (dict) – The dictionary to convert
Returns:: A human readable string
Return type:: str

anemoi.utils.humanize.list_to_human(lst: list[str], conjunction: str = 'and') → str

Convert a list of strings to a human readable string.

>>> list_to_human(["banana", "lemon", "apple", "orange"])
'banana, lemon, apple and orange'

Parameters:

lst (list of str) – The list of strings to concatenate
conjunction (str, optional) – The word to connect the last word in the list (like “or” or “and”), by default “and”

Returns:

Human readable string of list

Return type:

str

anemoi.utils.humanize.human_to_number(value: str | int, name: str, units: dict[str, int], none_ok: bool) → int | None

Convert a human readable string to a number.

Parameters:

value (str or int) – The value to convert
name (str) – The name of the value
units (dict) – The units to use for conversion
none_ok (bool) – Whether None is an acceptable value

Returns:

The converted value

Return type:

int or None

anemoi.utils.humanize.as_number(value: str | int, name: str | None = None, units: dict[str, int] | None = None, none_ok: bool = False) → int | None

Deprecated function to convert a human readable string to a number.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
units (dict, optional) – The units to use for conversion
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value

Return type:

int or None

anemoi.utils.humanize.human_seconds(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Convert a human readable string to seconds.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in seconds

Return type:

int or None

anemoi.utils.humanize.as_seconds(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Deprecated function to convert a human readable string to seconds.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in seconds

Return type:

int or None

anemoi.utils.humanize.human_to_percent(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Convert a human readable string to a percentage.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in percentage

Return type:

int or None

anemoi.utils.humanize.as_percent(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Deprecated function to convert a human readable string to a percentage.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in percentage

Return type:

int or None

anemoi.utils.humanize.human_to_bytes(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Convert a human readable string to bytes.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in bytes

Return type:

int or None

anemoi.utils.humanize.as_bytes(value: str | int, name: str | None = None, none_ok: bool = False) → int | None

Deprecated function to convert a human readable string to bytes.

Parameters:

value (str or int) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value in bytes

Return type:

int or None

anemoi.utils.humanize.human_to_timedelta(value: str, name: str | None = None, none_ok: bool = False) → timedelta

Convert a human readable string to a timedelta.

Parameters:

value (str) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value as a timedelta

Return type:

datetime.timedelta

anemoi.utils.humanize.as_timedelta(value: str, name: str | None = None, none_ok: bool = False) → timedelta

Deprecated function to convert a human readable string to a timedelta.

Parameters:

value (str) – The value to convert
name (str, optional) – The name of the value
none_ok (bool, optional) – Whether None is an acceptable value

Returns:

The converted value as a timedelta

Return type:

datetime.timedelta

anemoi.utils.humanize.rounded_datetime(d: datetime) → datetime

Round a datetime to the nearest second.

Parameters:: d (datetime.datetime) – The datetime to round
Returns:: The rounded datetime
Return type:: datetime.datetime

anemoi.utils.humanize.json_pretty_dump(obj: Any, max_line_length: int = 120, default: Callable = <class 'str'>) → str

Custom JSON dump function that keeps dicts and lists on one line if they are short enough.

Parameters:

obj (Any) – The object to be dumped as JSON.
max_line_length (int, optional) – Maximum allowed line length for pretty-printing. Default is 120.
default (function, optional) – Default function to convert non-serializable objects. Default is str.

Returns:

JSON string.

Return type:

str

anemoi.utils.humanize.shorten_list(lst: list[Any] | tuple[Any], max_length: int = 5) → list[Any] | tuple[Any]

Shorten a list to a maximum length.

Parameters:

lst (list or tuple) – The list to be shortened.
max_length (int, optional) – Maximum length of the shortened list. Default is 5.

Returns:

Shortened list.

Return type:

list or tuple

anemoi.utils.humanize.compress_dates(dates: list[datetime | str]) → str

Compress a list of dates into a human-readable format.

Parameters:: dates (list) – A list of dates, as datetime objects or strings.
Returns:: A human-readable string representing the compressed dates.
Return type:: str

anemoi.utils.humanize.print_dates(dates: list[datetime | str]) → None

Print a list of dates in a human-readable format.

Parameters:: dates (list) – A list of dates, as datetime objects or strings.

anemoi.utils.humanize.make_list_int(value: str | list[int] | tuple[int] | int) → list[int]

Convert a value to a list of integers.

Handles slash-separated strings including MARS-style range notation: "1/2/3", "1/to/3", and "1/to/10/by/2".

Parameters:: value (str, list, tuple, or int) – The value to convert to a list of integers.
Returns:: A list of integers.
Return type:: list[int]
Raises:: ValueError – If the value cannot be converted to a list of integers.

Examples

>>> make_list_int("1/2/3")
[1, 2, 3]
>>> make_list_int("0/to/6")
[0, 1, 2, 3, 4, 5, 6]
>>> make_list_int("0/to/12/by/6")
[0, 6, 12]

anemoi.utils.logs module

Logging utilities.

anemoi.utils.logs.set_logging_name(name: str) → None

Set the logging name for the current thread.

Parameters:: name (str) – The name to set for logging.

class anemoi.utils.logs.ThreadCustomFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

Custom logging formatter that includes thread-specific logging names.

format(record: LogRecord) → str

Format the log record to include the thread-specific logging name.

Parameters:: record (logging.LogRecord) – The log record to format.
Returns:: The formatted log record.
Return type:: str

anemoi.utils.logs.enable_logging_name(name: str = 'main') → None

Enable logging with a thread-specific logging name.

Parameters:: name (str, optional) – The default logging name to set, by default “main”.

anemoi.utils.logs.get_rich_handler() → Handler: Return a RichHandler with custom formatting for logging.

anemoi.utils.provenance module

Collect information about the current environment, like:

The Python version
The versions of the modules which are currently loaded
The git information for the modules which are currently loaded from a git repository
…

anemoi.utils.provenance.editable_installs() → dict[str, Path]

Return a dictionary of editable installs.

The check relies on how editable installs are handled based on PEP610. A <path-to-venv>/lib/site-packages/<package>.dist-info/direct_url.json file should be present.

Returns:: A dictionary mapping package names to their source directories for editable installs.
Return type:: dict[str, Path]

anemoi.utils.provenance.is_editable_install(init_path: str | Path) → bool: Determine if the given path corresponds to an editable install.

anemoi.utils.provenance.lookup_git_repo(path: str) → Any | None

Lookup the git repository for a given path.

Parameters:: path (str) – The path to lookup.
Returns:: The git repository if found, otherwise None.
Return type:: Repo, optional

anemoi.utils.provenance.package_distributions() → dict[str, list[str]]

Get the package distributions.

Returns:: The package distributions.
Return type:: dict

anemoi.utils.provenance.import_name_to_distribution_name(packages: list[str]) → dict[str, str]

Convert import names to distribution names.

Parameters:: packages (list of str) – The list of import names.
Returns:: The dictionary mapping import names to distribution names.
Return type:: dict

anemoi.utils.provenance.module_versions(full: bool) → tuple[dict[str, Any], dict[str, Any]]

Collect version information for all loaded modules and their git information.

Parameters:: full (bool) – Whether to collect full information.
Returns:: The version information and the git information.
Return type:: tuple of dict and dict

anemoi.utils.provenance.git_check(*args: Any) → dict[str, Any]

Return the git information for the given arguments.

Arguments can be:

an empty list, in that case all loaded modules are checked
a module name
a module object
an object or a class
a path to a directory

Parameters:

args (list) – The list of arguments to check

Returns:

An object with the git information for the given arguments.

>>> {
        "anemoi.utils": {
            "sha1": "c999d83ae283bcbb99f68d92c42d24315922129f",
            "remotes": [
                "git@github.com:ecmwf/anemoi-utils.git"
            ],
            "modified_files": [
                "anemoi/utils/checkpoints.py"
            ],
            "untracked_files": []
        }
    }

Return type:

dict

anemoi.utils.provenance.platform_info() → dict[str, Any]

Get the platform information.

Returns:: The platform information.
Return type:: dict

anemoi.utils.provenance.gpu_info() → str | list[dict[str, Any]]

Get the GPU information.

Returns:: The GPU information or an error message.
Return type:: str or list of dict

anemoi.utils.provenance.path_md5(path: str) → str

Calculate the MD5 hash of a file.

Parameters:: path (str) – The path to the file.
Returns:: The MD5 hash of the file.
Return type:: str

anemoi.utils.provenance.assets_info(paths: list[str]) → dict[str, Any]

Get information about the given assets.

Parameters:: paths (list of str) – The list of paths to the assets.
Returns:: The information about the assets.
Return type:: dict

anemoi.utils.provenance.gather_provenance_info(assets: list[str] = [], full: bool = False) → dict[str, Any]

Gather information about the current environment.

Parameters:

assets (list, optional) – A list of file paths for which to collect the MD5 sum, the size and time attributes, by default []
full (bool, optional) – If true, will also collect various paths, by default False

Returns:

A dictionary with the collected information

Return type:

dict

anemoi.utils.registry module

class anemoi.utils.registry.Wrapper(name: str, registry: Registry)

Bases: Generic[T]

A wrapper for the registry.

Parameters:

name (str) – The name of the wrapper.
registry (Registry) – The registry to wrap.

class anemoi.utils.registry.Error(error: Exception)

Bases: object

An error class. Used in place of a plugin that failed to load.

Parameters:: error (Exception) – The error.

class anemoi.utils.registry.Registry(package: str, key: str = '_type', api_version: str = '1.0.0')

Bases: Generic[T]

A registry of factories.

Parameters:

package (str) – The package name.
key (str, optional) – The key to use for the registry, by default “_type”.
api_version (str, optional) – The API version, by default ‘1.0.0’.

classmethod lookup_kind(kind: str) → Registry | None

Lookup a registry by kind.

Parameters:: kind (str) – The kind of the registry.
Returns:: The registry if found, otherwise None.
Return type:: Registry, optional

register(name: str, factory: Callable[[...], T], source: Any | None = None, aliases: list[str] | None = None) → None

register(name: str, factory: None = None, source: Any | None = None, aliases: list[str] | None = None) → Wrapper

Register a factory with the registry.

Parameters:

name (str) – The name of the factory.
factory (Callable, optional) – The factory to register, by default None.
source (Any, optional) – The source of the factory, by default None.
aliases (list of str, optional) – Aliases for the factory, by default None.

Returns:

A wrapper if the factory is None, otherwise None.

Return type:

Wrapper, optional

is_registered(name: str) → bool

Check if a factory is registered.

Parameters:: name (str) – The name of the factory.
Returns:: Whether the factory is registered.
Return type:: bool

lookup(name: str, *, return_none: bool = False) → type[T] | None

Lookup a factory by name.

Parameters:

name (str) – The name of the factory.
return_none (bool, optional) – Whether to return None if the factory is not found, by default False.

Returns:

The factory if found, otherwise None.

Return type:

Callable, optional

property factories: dict[str, type[T]]

property registered: list[str]: Get the registered factories.

create(name: str, *args: Any, **kwargs: Any) → T

Create an instance using a factory.

Parameters:

name (str) – The name of the factory.
*args (Any) – Positional arguments for the factory.
**kwargs (Any) – Keyword arguments for the factory.

Returns:

The created instance.

Return type:

Any

from_config(config: str | dict[str, Any], *args: Any, **kwargs: Any) → T

Create an instance from a configuration.

Parameters:

config (str or dict) – The configuration.
*args (Any) – Positional arguments for the factory.
**kwargs (Any) – Keyword arguments for the factory.

Returns:

The created instance.

Return type:

Any

aliases(): Get the aliases.

anemoi.utils.rules module

class anemoi.utils.rules.Rule(match: dict[str, Any], result: Any)

Bases: object

match(obj: Mapping[str, Any]) → bool

Check if the rule matches the given object.

Parameters:: obj (Mapping[str, Any]) – The object to check against the rule’s conditions.
Returns:: True if the rule matches, False otherwise.
Return type:: bool

property result: Any: The result associated with the rule.

property condition: dict[str, Any]: The conditions that define the rule.

class anemoi.utils.rules.RuleSet(rules: list[Rule | dict[str, Any] | list[Any]])

Bases: object

classmethod from_list(rules: list[Any]) → RuleSet

Create a RuleSet from a list of rules.

Parameters:: rules (List[Any]) – A list of rules to initialize the RuleSet.
Returns:: A new RuleSet object.
Return type:: RuleSet

classmethod from_files(path: str) → RuleSet

Create a RuleSet from a file.

Parameters:: path (str) – The path to the file containing the rules. Supported formats are .json and .yaml/.yml.
Returns:: A new RuleSet object.
Return type:: RuleSet
Raises:: ValueError – If the file format is unsupported.

classmethod from_any(rules: str | list[Any]) → RuleSet

Create a RuleSet from a list or a file path.

Parameters:: rules (Union[str, List[Any]]) – The rules to initialize the RuleSet, either as a list or a file path.
Returns:: A new RuleSet object.
Return type:: RuleSet
Raises:: ValueError – If the rules format is unsupported.

match(obj: Mapping[str, Any], strategy: str = 'first-match') → Rule | None

Match an object against the rules in the RuleSet.

Parameters:

obj (Mapping[str, Any]) – The object to match against the rules.
strategy (str, optional) – The matching strategy to use. Currently, only ‘first-match’ is supported.

Returns:

The first matching rule, or None if no match is found.

Return type:

Optional[Rule]

Raises:

AssertionError – If an unsupported strategy is provided.

anemoi.utils.s3 module

anemoi.utils.s3.s3_client(*args: Any, **kwargs: Any) → Any

Create an S3 client.

Parameters:

*args (Any) – Positional arguments for the S3 client.
**kwargs (Any) – Keyword arguments for the S3 client.

Returns:

The S3 client.

Return type:

Any

anemoi.utils.s3.upload(source: str, target: str, *, overwrite: bool = False, resume: bool = False, verbosity: int = 1, progress: Callable | None = None, threads: int = 1) → None

Upload a file to S3.

Parameters:

source (str) – The source file path.
target (str) – The target S3 path.
overwrite (bool, optional) – Whether to overwrite the target file, by default False.
resume (bool, optional) – Whether to resume a previous upload, by default False.
verbosity (int, optional) – The verbosity level, by default 1.
progress (Callable, optional) – A callback function for progress updates, by default None.
threads (int, optional) – The number of threads to use, by default 1.

anemoi.utils.s3.download(*args: Any, **kwargs: Any) → Any

Download a file from S3.

Parameters:

*args (Any) – Positional arguments for the download.
**kwargs (Any) – Keyword arguments for the download.

Returns:

The result of the download.

Return type:

Any

anemoi.utils.s3.delete(*args: Any, **kwargs: Any) → Any

Delete a file from S3.

Parameters:

*args (Any) – Positional arguments for the delete.
**kwargs (Any) – Keyword arguments for the delete.

Returns:

The result of the delete.

Return type:

Any

anemoi.utils.sanitise module

anemoi.utils.sanitise.sanitise(obj: Any, level=1) → Any

Sanitise an object by replacing all full paths with shortened versions and URL credentials with ‘***’.

Parameters:

obj (Any) – The object to sanitise.
level (int, optional) – The level of sanitation. The higher levels will also apply the levels below it. - 1: Shorten file paths to file name and hide credentials in URLs (default). - 2: Hide hostnames in URLs. - 3: Hide full file paths and URLs.

Returns:

The sanitised object.

Return type:

Any

anemoi.utils.sanitize module

anemoi.utils.sanitize.sanitize(obj: Any, level=1) → Any

Sanitise an object by replacing all full paths with shortened versions and URL credentials with ‘***’.

Parameters:

obj (Any) – The object to sanitise.
level (int, optional) – The level of sanitation. The higher levels will also apply the levels below it. - 1: Shorten file paths to file name and hide credentials in URLs (default). - 2: Hide hostnames in URLs. - 3: Hide full file paths and URLs.

Returns:

The sanitised object.

Return type:

Any

anemoi.utils.testing module

anemoi.utils.text module

Text utilities.

anemoi.utils.text.dotted_line(width: int = 84) → str

Return a dotted line using ‘┈’.

>>> dotted_line(40)
┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈

Parameters:: width (int, optional) – Number of characters, by default 84
Returns:: The dotted line
Return type:: str

anemoi.utils.text.visual_len(s: str | list[tuple[str, int]]) → int

Compute the length of a string as it appears on the terminal.

Parameters:: s (str or list of tuple) – The string or list of visual characters with their lengths.
Returns:: The visual length of the string.
Return type:: int

anemoi.utils.text.boxed(text: str, min_width: int = 80, max_width: int = 80) → str

Put a box around a text.

>>> boxed("Hello,\nWorld!", max_width=40)
┌──────────────────────────────────────────┐
│ Hello,                                   │
│ World!                                   │
└──────────────────────────────────────────┘

Parameters:

text (str) – The text to box
min_width (int, optional) – The minimum width of the box, by default 80
max_width (int, optional) – The maximum width of the box, by default 80

Returns:

A boxed version of the input text

Return type:

str

anemoi.utils.text.bold(text: str) → str

Make the text bold.

Parameters:: text (str) – The text to make bold.
Returns:: The bold text.
Return type:: str

anemoi.utils.text.red(text: str) → str

Make the text red.

Parameters:: text (str) – The text to make red.
Returns:: The red text.
Return type:: str

anemoi.utils.text.green(text: str) → str

Make the text green.

Parameters:: text (str) – The text to make green.
Returns:: The green text.
Return type:: str

anemoi.utils.text.blue(text: str) → str

Make the text blue.

Parameters:: text (str) – The text to make blue.
Returns:: The blue text.
Return type:: str

class anemoi.utils.text.Tree(actor: Any, parent: Tree | None = None)

Bases: object

Tree data structure.

Parameters:

actor (Any) – The actor associated with the tree node.
parent (Tree, optional) – The parent tree node, by default None.

adopt(kid: Tree) → None

Adopt a child tree node.

Parameters:: kid (Tree) – The child tree node to adopt.

forget() → None: Forget the current tree node.

property is_leaf: bool

True if the tree node is a leaf, False otherwise.

Type:: Bool

property key: tuple

The key of the tree node.

Type:: Tuple

property summary: str

The summary of the tree node.

Type:: Str

as_dict() → dict

Convert the tree node to a dictionary.

Returns:: The dictionary representation of the tree node.
Return type:: dict

node(actor: Any, insert: bool = False) → Tree

Create a new tree node.

Parameters:

actor (Any) – The actor associated with the new tree node.
insert (bool, optional) – Whether to insert the new tree node at the beginning, by default False.

Returns:

The new tree node.

Return type:

Tree

print() → None: Print the tree.

to_json(depth: int = 0) → dict

Convert the tree to a JSON serializable dictionary.

Parameters:: depth (int, optional) – The depth of the tree, by default 0.
Returns:: The JSON serializable dictionary representation of the tree.
Return type:: dict

anemoi.utils.text.table(rows: list[list[Any]], header: list[str], align: list[str], margin: int = 0) → str

Format a table.

>>> table([['Aa', 12, 5],
        ['B', 120, 1],
        ['C', 9, 123]],
        ['C1', 'C2', 'C3'],
        ['<', '>', '>'])
    C1 │  C2 │  C3
    ───┼─────┼────
    Aa │  12 │   5
    B  │ 120 │   1
    C  │   9 │ 123
    ───┴─────┴────

Parameters:

rows (list of lists (or tuples)) – The rows of the table
header (A list or tuple of strings) – The header of the table
align (A list of '<', '>', or '^') – To align the columns to the left, right, or center
margin (int, optional) – Extra spaces on the left side of the table, by default 0

Returns:

A table as a string

Return type:

str

anemoi.utils.text.progress(done: int, todo: int, width: int = 80) → str

Generates a progress bar string.

Parameters:

done (int) – The number of tasks completed.
todo (int) – The total number of tasks.
width (int, optional) – The width of the progress bar, by default 80.

Returns:

A string representing the progress bar.

Return type:

str

Example

>>> print(progress(10, 100,width=50))
█████▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒

anemoi.utils.timer module

Logging utilities.

class anemoi.utils.timer.Timer(title: str, logger: Logger = <Logger anemoi.utils.timer (WARNING)>)

Bases: object

Context manager to measure elapsed time.

Parameters:

title (str) – The title of the timer.
logger (logging.Logger, optional) – The logger to use for logging the elapsed time, by default LOGGER.

property elapsed: float

The elapsed time in seconds.

Type:: Float

class anemoi.utils.timer.Timers(logger: Logger = <Logger anemoi.utils.timer (WARNING)>)

Bases: object

A collection of timers.

Parameters:: logger (logging.Logger, optional) – The logger to use for logging the elapsed time, by default LOGGER.

report() → None: Log the elapsed time for all timers.

anemoi.utils.window module

class anemoi.utils.window.Window(window: str)

Bases: object

Represents a time window for selecting data, with before/after offsets and inclusivity.

Parses a window string to determine the time offsets before and after a central point, and whether the window is inclusive or exclusive at each end. Used by WindowView to select data slices.

property width: float: Calculate the total width of the window in seconds.

property closed: tuple[bool, bool]: Determine if the window is closed (inclusive at both ends).