config

couple

class anemoi.inference.config.couple.CoupleConfiguration(*, date: datetime | None = None, description: str | None = None, lead_time: str | int | timedelta | None = None, name: str | None = None, transport: str, couplings: list[dict[str, list[str]]], tasks: dict[str, Any], env: dict[str, Any] = {})

Bases: Configuration

Configuration class for the couple runner.

description: str | None
lead_time: str | int | datetime.timedelta | None

The lead time for the forecast. This can be a string, an integer or a timedelta object. If an integer, it represents a number of hours. Otherwise, it is parsed by anemoi.utils.dates.as_timedelta().

name: str | None

Used by prepml.

transport: str
couplings: list[dict[str, list[str]]]
model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tasks: dict[str, Any]
env: dict[str, Any]

Environment variables to set before running the model. This may be useful to control some packages such as eccodes. In certain cases, the variables mey be set too late, if the package for which they are intended is already loaded when the runner is configured.

run

class anemoi.inference.config.run.RunConfiguration(*, date: ~datetime.datetime | None = None, description: str | None = None, checkpoint: str | dict[~typing.Literal['huggingface'], dict[str, ~typing.Any] | str], runner: str | dict[str, ~typing.Any] = 'default', lead_time: str | int | ~datetime.timedelta = '10d', name: str | None = None, verbosity: int = 0, use_profiler: bool = False, input: str | dict[str, ~typing.Any] = 'test', output: str | dict[str, ~typing.Any] = 'printer', pre_processors: list[str | dict[str, ~typing.Any]] = [], post_processors: list[str | dict[str, ~typing.Any]] = [], dynamic_forcings: str | dict[str, ~typing.Any] | None = None, constant_forcings: str | dict[str, ~typing.Any] | None = None, device: str | None = None, precision: str | None = None, allow_nans: bool | None = None, use_grib_paramid: bool = False, write_initial_state: bool = True, initial_state_categories: list[str] = ['prognostics', 'constant_forcings'], predict_kwargs: dict[str, ~typing.Any] = <factory>, typed_variables: dict[str, dict] = <factory>, output_frequency: str | None = None, env: dict[str, str | int] = {}, patch_metadata: dict[str, ~typing.Any] | ~pathlib.Annotated[~pathlib.Path, ~pydantic.types.PathType(path_type=file)] = {}, development_hacks: dict[str, ~typing.Any] = {}, trace_path: dict | str | None = None, debugging_info: dict[str, ~typing.Any] = {}, preload_checkpoint: bool = False, preload_buffer_size: int = 33554432)

Bases: Configuration

Configuration class for a default runner.

description: str | None
checkpoint: str | dict[Literal['huggingface'], dict[str, Any] | str]

A path to an Anemoi checkpoint file.

runner: str | dict[str, Any]

The runner to use. For runners that take extra options, they can be passed here in dictionary format with the runner name as the top level dictionary key. For example, if using parallel, the cluster option can be set here.

lead_time: str | int | datetime.timedelta

The lead time for the forecast. This can be a string, an integer or a timedelta object. If an integer, it represents a number of hours. Otherwise, it is parsed by anemoi.utils.dates.as_timedelta().

name: str | None

Used by prepml.

verbosity: int

The verbosity level of the runner. This can be 0 (default), 1, 2 or 3.

use_profiler: bool

If True, the inference will be profiled, producing time and memory report.

input: str | dict[str, Any]
output: str | dict[str, Any]
pre_processors: list[ProcessorConfig]
post_processors: list[ProcessorConfig]
dynamic_forcings: str | dict[str, Any] | None

Where to find the dinamic forcings. (default is input)

constant_forcings: str | dict[str, Any] | None

Where to find the constant forcings (default to input).

device: str | None

The device on which the model should run. This can be “cpu”, “cuda” or any other value supported by PyTorch. If None, the device will be automatically detected using anemoi.inference.device.get_available_device().

precision: str | None

The precision in which the model should run. If not provided, the model will use the precision used during training.

allow_nans: bool | None
  • If None (default), the model will check for NaNs in the input. If NaNs are found, the model issue a warning and allow_nans to True.

  • If False, the model will raise an exception if NaNs are found in the input and output.

  • If True, the model will allow NaNs in the input and output.

use_grib_paramid: bool

If True, the runner will use the grib parameter ID when generating MARS requests.

model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

write_initial_state: bool

Wether to write the initial state to the output file. If the model is multi-step, only fields at the forecast reference date are written.

initial_state_categories: list[str]

A list of categories to use when writing step zero.

predict_kwargs: dict[str, Any]

Extra keyword arguments to pass to the model’s predict_step method. Will ignore kwargs that are already passed by the runner.

typed_variables: dict[str, dict]

A list of typed variables to support the encoding of outputs.

output_frequency: str | None

The frequency at which to write the output. This can be a string or an integer. If a string, it is parsed by anemoi.utils.dates.as_timedelta().

env: dict[str, str | int]

Environment variables to set before running the model. This may be useful to control some packages such as eccodes. In certain cases, the variables mey be set too late, if the package for which they are intended is already loaded when the runner is configured.

patch_metadata: dict[str, Any] | FilePath

A dictionary of metadata to patch the checkpoint metadata with, or a path to a YAML file containing the metadata. This is used to test new features or to work around issues with the checkpoint metadata.

development_hacks: dict[str, Any]

A dictionary of development hacks to apply to the runner. This is used to test new features or to work around.

trace_path: dict | str | None

A path to a file where to store the trace of the runner. This is useful to debug the runner.

debugging_info: dict[str, Any]

A dictionary to store debug information. This is ignored.

preload_checkpoint: bool

Whether to read the checkpoint file from disk before loading the model.

preload_buffer_size: int

Size of the buffer to use when preloading the checkpoint file, in bytes. Default is 32 MB.

classmethod as_dict(patch_metadata: dict | Annotated[Path, PathType(path_type=file)]) dict

utils

anemoi.inference.config.utils.input_types_config(config: DotDict, *names: str) DotDict

Get the input type configuration from the config using a list of possible names, returning the first one found. For example, names could be [“constant_forcings”, “input”] to get forcings from the constant_forcings input, or the main input if not specified.

anemoi.inference.config.utils.multi_datasets_config(config: Any, dataset_name: str, datasets: list[str], strict: bool = True)

Extract the configuration for a specific dataset name from a multi-dataset config entry. If the config only has a single entry, the entry corresponding to the dataset name is returned if it exists, otherwise the config is returned as is. For example, used with a config like:

``` output:

era5:

grib: out-era5.grib

cerra:

netcdf: out-cerra.nc

```

If strict is True and config is a dictionary with multiple keys but dataset_name is not one of them, an error is raised. If strict is False, the config is returned as is if dataset_name is not one of the keys. Used for config-entries that can take multi-key dicts (like patch_metadata).