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.

ext: str

abstractmethod load(path: Path) → dict

abstractmethod save(path: Path, data: dict) → Path

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: str = '.json'

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

Save data to a JSON file.

Parameters:

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

Returns:

The temporary file path

Return type:

Path

load(path: Path) → dict

Load data from a JSON file.

Parameters:: path (Path) – 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: str = '.npz'

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

Save data to an NPZ file.

Parameters:

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

Returns:

The temporary file path

Return type:

str

load(path: Path) → dict

Load data from an NPZ file.

Parameters:: path (Path) – 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.checkpoints.unpickle_model(path, **kwargs) → dict

anemoi.utils.cli module

class anemoi.utils.cli.Command

Bases: object

Base class for commands.

accept_unknown_args = False

aliases: list[str] = []

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

@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.

anemoi.utils.grib.SETTINGS = AnemoiSettings(object_storage=ObjectStorageConfig(endpoint_url=None, skip_signature=False, access_key_id=None, secret_access_key=None, region=None, account_name=None, account_key=None, sas_token=None, type=None), datasets=DatasetsConfig(path=[], use_search_path_not_found=False, ignore_naming_conventions=False, named=DatasetsNamedConfig()), paramdb=ParamDBConfig(default_origin='ecmf', cache_length=30, local_cache=None), registry=RegistryConfig(api_url=None, api_token=None, allow_delete=False, plots_uri_pattern=None, datasets_uri_pattern=None, weights_uri_pattern=None, weights_platform=None), utils=UtilsConfig(grids_path=None, cache_directory=PosixPath('/home/docs/.cache/anemoi'), debug_imports_in_cli=False)): Anemoi settings, loaded on module import.

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.settings module

anemoi.utils.settings.copy_default_settings(dest: Path | None = None, *, overwrite: bool = False) → Path

Copy the bundled settings.defaults.toml to the user default settings location.

Parameters:

dest (Path, optional) – Custom destination path for the copied defaults file. If None (default), uses the standard user config location (~/.config/anemoi/settings.defaults.toml).
overwrite (bool) – If False (default), skip the copy when the destination file already exists.

Returns:

The path the file was (or would have been) written to.

Return type:

Path

anemoi.utils.settings.convert_to_secret(val: dict[str, str | Any]) → dict[str, SecretStr | Any]
anemoi.utils.settings.convert_to_secret(val: str) → SecretStr

class anemoi.utils.settings.AnemoiConfigFileSource(settings_cls: type[BaseSettings], toml_file: Path, yaml_file: Path)

Bases: PydanticBaseSettingsSource

Loads settings from a TOML/YAML config file pair.

Secret (SecretStr) values may live in any settings file, provided that file has mode 0600. A file that contains no secrets has no permission requirement. There is therefore no need to partition secret and non-secret keys into separate files: keys may go anywhere as long as any file holding a secret is mode 0600.

get_field_value(field: Any, field_name: str) → tuple[Any, str, bool]

Gets the value, the key for model creation, and a flag to determine whether value is complex.

This is an abstract method that should be overridden in every settings source classes.

Parameters:

field – The field.
field_name – The field name.

Returns:

A tuple that contains the value, key and a flag to determine whether value is complex.

class anemoi.utils.settings.AnemoiSettings(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_prefix_target: EnvPrefixTarget | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_nested_max_split: int | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None, _cli_ignore_unknown_args: bool | None = None, _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None, _cli_shortcuts: Mapping[str, str | list[str]] | None = None, _secrets_dir: PathType | None = None, _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None, *, object_storage: ~anemoi.utils.settings_schema.object_storage.ObjectStorageConfig = <factory>, datasets: ~anemoi.utils.settings_schema.datasets.DatasetsConfig = <factory>, paramdb: ~anemoi.utils.settings_schema.paramdb.ParamDBConfig = <factory>, registry: ~anemoi.utils.settings_schema.registry.RegistryConfig = <factory>, utils: ~anemoi.utils.settings_schema.utils.UtilsConfig = <factory>)

Bases: BaseSettings

Settings for Anemoi.

Use the ANEMOI_SETTINGS_FILE environment variable to specify a custom location for the main settings file (default: ~/.config/anemoi/settings.toml).

The main settings file can be in either TOML or YAML format; the extension is ignored. Keys may be placed in any settings file, secret or not: there is no requirement to partition secret and non-secret keys into separate files. The only rule is that any file containing a ``SecretStr`` value must have permissions ``0600``; a secret found in a file that is readable by group or others is a fatal error (PermissionError).

An optional secrets file with the same name but suffixed with .secrets (e.g. settings.secrets.toml) is also read. It is the natural place to keep secrets when the main settings file must stay readable by others.

Note: Init kwargs are intentionally disabled as a source of settings.

Settings are loaded with the following priority (highest to lowest):

Environment variables (with prefix ANEMOI_SETTINGS_ and keys in upper case with underscores)
Values from the .secrets.(toml|yaml) file
Values from the .(toml|yaml) file
Default values defined in the AnemoiSettings class and its nested models

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'ANEMOI_SETTINGS_', 'env_prefix_target': 'variable', 'extra': 'ignore', 'hide_input_in_errors': True, 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'populate_by_name': True, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'serialize_by_alias': True, 'toml_file': None, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

object_storage: ObjectStorageConfig: Configuration for cloud (S3-compatible and Azure Blob) object storage.

datasets: DatasetsConfig: Dataset discovery and validation settings.

paramdb: ParamDBConfig: GRIB parameter database lookup settings.

registry: RegistryConfig: Configuration for access to the Anemoi registry.

utils: UtilsConfig: Miscellaneous anemoi-utils settings.

classmethod settings_customise_sources(settings_cls: type[BaseSettings], init_settings: PydanticBaseSettingsSource, env_settings: PydanticBaseSettingsSource, dotenv_settings: PydanticBaseSettingsSource, file_secret_settings: PydanticBaseSettingsSource) → tuple[PydanticBaseSettingsSource, ...]

Define the sources and their order for loading the settings values.

Parameters:

settings_cls – The Settings class.
init_settings – The InitSettingsSource instance.
env_settings – The EnvSettingsSource instance.
dotenv_settings – The DotEnvSettingsSource instance.
file_secret_settings – The SecretsSettingsSource instance.

Returns:

A tuple containing the sources and their order for loading the settings values.

anemoi.utils.settings.SETTINGS = AnemoiSettings(object_storage=ObjectStorageConfig(endpoint_url=None, skip_signature=False, access_key_id=None, secret_access_key=None, region=None, account_name=None, account_key=None, sas_token=None, type=None), datasets=DatasetsConfig(path=[], use_search_path_not_found=False, ignore_naming_conventions=False, named=DatasetsNamedConfig()), paramdb=ParamDBConfig(default_origin='ecmf', cache_length=30, local_cache=None), registry=RegistryConfig(api_url=None, api_token=None, allow_delete=False, plots_uri_pattern=None, datasets_uri_pattern=None, weights_uri_pattern=None, weights_platform=None), utils=UtilsConfig(grids_path=None, cache_directory=PosixPath('/home/docs/.cache/anemoi'), debug_imports_in_cli=False))

Global instance of the AnemoiSettings. This will be created on first import of the anemoi.utils.settings module, and can be reloaded with reload_settings().

Use AnemoiSettings() to create separate instances if needed, but these will be runtime specific.

anemoi.utils.settings.reload_settings()

Reload the Anemoi settings.

Is run in-place on the global SETTINGS instance.

anemoi.utils.testing module

class anemoi.utils.testing.TemporaryDirectoryForTestData(base_dir: Path): Bases: object

anemoi.utils.testing.temporary_directory_for_test_data(tmp_path_factory: TempPathFactory) → TemporaryDirectoryForTestData

anemoi.utils.testing.url_for_test_data(path: str) → str

Generate the URL for the test data based on the given path.

Parameters:: path (str) – The relative path to the test data.
Returns:: The full URL to the test data.
Return type:: str

class anemoi.utils.testing.GetTestData(temporary_directory_for_test_data: TemporaryDirectoryForTestData): Bases: object

anemoi.utils.testing.get_test_data(temporary_directory_for_test_data: TemporaryDirectoryForTestData) → GetTestData

class anemoi.utils.testing.GetTestArchive(temporary_directory_for_test_data: TemporaryDirectoryForTestData, get_test_data: GetTestData): Bases: object

anemoi.utils.testing.get_test_archive(temporary_directory_for_test_data: TemporaryDirectoryForTestData, get_test_data: GetTestData) → GetTestArchive

anemoi.utils.testing.packages_installed(*names: str) → bool

Check if all the given packages are installed.

Use this function to check if the required packages are installed before running tests.

>>> @pytest.mark.skipif(not packages_installed("foo", "bar"), reason="Packages 'foo' and 'bar' are not installed")
>>> def test_foo_bar() -> None:
>>>    ...

Parameters:: names (str) – The names of the packages to check.
Returns:: Flag indicating if all the packages are installed.”
Return type:: bool

anemoi.utils.testing.skip_missing_packages(*names: str) → MarkDecorator

Skip a test if any of the specified packages are missing.

Parameters:: names (str) – The names of the packages to check.
Returns:: A decorator that skips the test if any of the specified packages are missing.
Return type:: Callable

anemoi.utils.testing.skip_if_missing_command(cmd: str) → MarkDecorator

Skip a test if the specified command is not available.

Parameters:: cmd (str) – The name of the command to check.
Returns:: A decorator that skips the test if the specified command is not available.
Return type:: Callable

anemoi.utils.testing.cli_testing(package: str, cmd: str, *args: str) → None

Run a CLI command for testing purposes.

Parameters:

package (str) – The name of the package containing the CLI commands. Can be ‘anemoi-datasets’ or ‘anemoi.datasets’.
cmd (str) – The command to run.
*args (str) – Additional arguments to pass to the command.

anemoi.utils.testing.run_tests(globals: dict[str, Callable[[], None]]) → None

Run all test functions that start with test_.

Parameters:: globals (dict[str, Callable[[], None]]) – The global namespace containing the test functions.

Example

Call from a test file to run all tests in that file:

if __name__ == "__main__":
    from anemoi.utils.testing import run_tests
    run_tests(globals())

Useful for debugging or running tests in an interactive environment.

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).