inputs

cds

anemoi.inference.inputs.cds.retrieve(requests: list[dict[str, Any]], grid: str | list[float] | None, area: list[float] | str | None, dataset: str | dict[str, Any], **kwargs: Any) → FieldList

Retrieve data from CDS.

Parameters:

• requests (List[Dict[str, Any]]) – List of request dictionaries.

• grid (Optional[Union[str, List[float]]]) – Grid specification.

• area (Optional[Union[List[float], str]]) – Area specification.

• dataset (Union[str, Dict[str, Any]]) – Dataset to use.

• **kwargs (Any) – Additional keyword arguments.

Returns:

Retrieved data.

Return type:

Any

class anemoi.inference.inputs.cds.CDSInput(context: Context, metadata: Metadata, *, variables: list[str] | None = None, pre_processors: list[str | dict[str, Any]] | None = None, dataset: str | dict[str, Any], namer: Any | None = None, purpose: str | None = None, **kwargs: Any)

Bases: GribInput

Get input fields from CDS.

Initialize the CDSInput.

Parameters:

• context (Context) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• variables (list[str] | None) – List of variables to be handled by the input, or None for a sensible default variables.

• pre_processors (Optional[List[ProcessorConfig]], default None) – Pre-processors to apply to the input

• dataset (Union[str, Dict[str, Any]]) – The dataset to use.

• namer (Optional[Any]) – Optional namer for the input.

• **kwargs (Any) – Additional keyword arguments.

trace_name = 'cds'

create_input_state(*, dates: list[str | datetime | int], **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• dates (list[Date]) – The dates for which to create the input state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

State

retrieve(variables: list[str], dates: list[str | datetime | int]) → Any

Retrieve data for the given variables and dates.

Parameters:

• variables (List[str]) – List of variables to retrieve.

• dates (List[Date]) – List of dates for which to retrieve data.

Returns:

Retrieved data.

Return type:

Any

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for the given variables and dates.

Parameters:

• dates (List[Date]) – The list of dates for which to load the forcings state.

• current_state (State) – The current state to be updated with the loaded forcings state.

Returns:

The loaded forcings state.

Return type:

Any

default_initial_date() → datetime

Return the initial date to use when none is given in the config (assuming a forecasting context).

The runner calls this when date isn’t set in the config or on the CLI. Each input decides how to pick the date: some read it from the data (e.g. the latest valid time in a file), others use a fixed rule (e.g. yesterday at 00:00), and some don’t implement this method at all.

cutout

class anemoi.inference.inputs.cutout.Cutout(context: Context, metadata: Metadata, *args: dict[str, dict], sources: list[dict[str, dict]] | None = None, **kwargs)

Bases: Input

Combines one or more LAMs into a global source using cutouts.

Create a cutout input from a list of sources.

Parameters:

• context (Context) – The context runner.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• sources (list[dict[str, dict]]) – List of sources / inputs to combine, the order defines the order in which they are combined.

create_input_state(*, dates: list[str | datetime | int], **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• dates (list[Date]) – The dates for which to create the input state.

• **kwargs (dict) – Additional keyword arguments for the source input state creation.

Returns:

The created input state.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for the given variables and dates.

Parameters:

• dates (List[Date]) – List of dates for which to load the forcings.

• current_state (State) – The current state of the input.

Returns:

The loaded forcings state.

Return type:

State

dataset

class anemoi.inference.inputs.dataset.DatasetInput(context: Context, metadata: Metadata, *, open_dataset_args: tuple[Any, ...], open_dataset_kwargs: dict[str, Any], grid_indices: Any = None, use_trajectories: bool = False, **kwargs: Any)

Bases: Input

Handles anemoi-datasets dataset as input.

Initialize the DatasetInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• open_dataset_args (Tuple[Any, ...]) – Arguments for the dataset.

• open_dataset_kwargs (Dict[str, Any]) – Keyword arguments for the dataset.

• grid_indices (Optional[Any]) – Indices to reduce the input grid. If None, the full grid is used.

• use_trajectories (bool) – Whether to expect dataset as a trajectory (i.e. with a step / forecast dimension), and to get multiple dates from within a single trajectory.

• **kwargs (Any) – Additional keyword arguments.

property ds: Any

Return the dataset.

property ds_dates: ndarray

Return the dates of the dataset.

property latitudes: ndarray[tuple[Any, ...], dtype[Any]]

Return the latitudes.

property longitudes: ndarray[tuple[Any, ...], dtype[Any]]

Return the longitudes.

create_input_state(*, dates: list[str | datetime | int], ref_date_index: int = -1, **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• dates (list[Date]) – The dates for which to create the input state.

• ref_date_index (int = -1) – The index in dates to use as reference date for the state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

Dict[str, Any]

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for the given variables and dates.

Parameters:

• dates (List[Any]) – List of dates for which to load the forcings.

• current_state (State) – The current state of the input.

Returns:

The loaded forcings state.

Return type:

State

class anemoi.inference.inputs.dataset.DatasetInputArgsKwargs(context: Context, metadata: Metadata, *args: Any, use_original_paths: bool = False, variables: list[str] | None, pre_processors: list[str | dict[str, Any]] | None = None, grid_indices=None, purpose: str | None = None, use_trajectories: bool = False, **kwargs: Any)

Bases: DatasetInput

Handles anemoi-datasets dataset as input.

Initialize the DatasetInputArgsKwargs.

Parameters:

• context (Context) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• use_original_paths (bool) – Whether to use original paths.

• use_trajectories (bool) – Whether to expect dataset as a trajectory (i.e. with a step / forecast dimension), and to get multiple dates from within a single trajectory.

trace_name = 'dataset/provided'

class anemoi.inference.inputs.dataset.DataloaderInput(context: Context, metadata: Metadata, *, use_original_paths: bool = False, **kwargs: Any)

Bases: DatasetInput

Handles anemoi-datasets dataset as input.

Initialize the DataloaderInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – The metadata corresponding to the dataset this input is handling.

• name (str) – The name of the dataloader.

• use_original_paths (bool) – Whether to use original paths.

class anemoi.inference.inputs.dataset.TestInput(context: Context, metadata: Metadata, *, use_original_paths: bool = False, **kwargs: Any)

Bases: DataloaderInput

Handles anemoi-datasets dataset as input.

Initialize the DataloaderInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – The metadata corresponding to the dataset this input is handling.

• name (str) – The name of the dataloader.

• use_original_paths (bool) – Whether to use original paths.

trace_name = 'dataset/test'

name = 'test'

class anemoi.inference.inputs.dataset.ValidationInput(context: Context, metadata: Metadata, *, use_original_paths: bool = False, **kwargs: Any)

Bases: DataloaderInput

Handles anemoi-datasets dataset as input.

Initialize the DataloaderInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – The metadata corresponding to the dataset this input is handling.

• name (str) – The name of the dataloader.

• use_original_paths (bool) – Whether to use original paths.

trace_name = 'dataset/validation'

name = 'validation'

class anemoi.inference.inputs.dataset.TrainingInput(context: Context, metadata: Metadata, *, use_original_paths: bool = False, **kwargs: Any)

Bases: DataloaderInput

Handles anemoi-datasets dataset as input.

Initialize the DataloaderInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – The metadata corresponding to the dataset this input is handling.

• name (str) – The name of the dataloader.

• use_original_paths (bool) – Whether to use original paths.

trace_name = 'dataset/training'

name = 'training'

dummy

Dummy input used for testing.

It will generate fields with constant values for each variable and date. These values are then tested in the mock model.

class anemoi.inference.inputs.dummy.DummyInput(context: Context, metadata: Metadata, **kwargs)

Bases: EkdInput

Dummy input used for testing.

Initialize the EkdInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• namer (Optional[Union[Callable[[Any, Dict[str, Any]], str], Dict[str, Any]]]) – Optional namer for the input.

trace_name = 'dummy'

create_input_state(*, dates: list[str | datetime | int], **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• dates (list[Date]) – The dates for which to create the input state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for the given variables and dates.

Parameters:

• dates (list[Date]) – List of dates for which to load the forcings.

• current_state (State) – The current state of the input.

Returns:

The loaded forcings state.

Return type:

Any

template_lookup(name: str) → dict

Lookup a template by name.

Parameters:

name (str) – The name of the template to lookup.

Returns:

The template dictionary.

Return type:

dict

default_initial_date() → datetime

Return the initial date to use when none is given in the config (assuming a forecasting context).

The runner calls this when date isn’t set in the config or on the CLI. Each input decides how to pick the date: some read it from the data (e.g. the latest valid time in a file), others use a fixed rule (e.g. yesterday at 00:00), and some don’t implement this method at all.

ekd

anemoi.inference.inputs.ekd.find_variable(data: FieldList, name: str, namer: callable, **kwargs: Any) → FieldList

Find a variable in an earthkit FieldList.

Parameters:

• data (ekd.FieldList) – The data to search.

• name (str) – The name of the variable to find.

• namer (callable) – The namer function to use for naming fields.

• **kwargs (Any) – Additional arguments for selecting the variable.

Returns:

The selected variable (FieldList subset).

Return type:

ekd.FieldList

class anemoi.inference.inputs.ekd.RulesNamer(rules: Any, default_namer: Callable[[Any, dict[str, Any]], str])

Bases: object

A namer that uses rules to generate names.

Initialize the RulesNamer.

Parameters:

• rules (List[List[Dict[str, Any], Dict[str, Any]]]) – The rules for naming.

• default_namer (Callable[[Any, Dict[str, Any]], str]) – The default namer to use if no rules match.

substitute(template: str, field: Field, original_metadata: dict[str, Any]) → str

Substitute placeholders in the template with metadata values.

Parameters:

• template (str) – The template string with placeholders.

• field (ekd.Field) – The field for which to generate a name.

• original_metadata (Dict[str, Any]) – The original metadata of the field.

Returns:

The generated name with placeholders substituted.

Return type:

str

class anemoi.inference.inputs.ekd.EkdInput(context: Context, metadata: Metadata, *, namer: Any | None = None, **kwargs)

Bases: Input

Handles earthkit-data FieldList as input.

Initialize the EkdInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• namer (Optional[Union[Callable[[Any, Dict[str, Any]], str], Dict[str, Any]]]) – Optional namer for the input.

set_private_attributes(state: dict[str, Any], fields: FieldList) → None

Set private attributes to the state.

Provides geography information if available retrieved from the fields.

empty

Dummy input used for testing.

It will generate fields with constant values for each variable and date. These values are then tested in the mock model.

class anemoi.inference.inputs.empty.EmptyInput(context: Context, metadata: Metadata, **kwargs: Any)

Bases: Input

An Input that is always empty.

Initialise the EmptyInput.

Parameters:

• context (Context) – The context object for the input.

• **kwargs (object) – Additional keyword arguments.

trace_name = 'empty'

create_input_state(*, dates: list[str | datetime | int], **kwargs) → dict[str, Any]

Create an empty input state.

Parameters:

• dates (list[Date]) – The dates for the input state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created empty input state.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load an empty forcings state.

Parameters:

• dates (list[Date]) – The list of dates for the forcings state.

• current_state (State) – The current state (unused).

Returns:

The loaded empty forcings state.

Return type:

State

fdb

class anemoi.inference.inputs.fdb.FDBInput(context: Context, metadata: Metadata, *, fdb_config: dict | None = None, fdb_userconfig: dict | None = None, variables: list[str] | None = None, pre_processors: list[str | dict[str, Any]] | None = None, namer: Any | None = None, purpose: str | None = None, **kwargs: Any)

Bases: GribInput

Get input fields from FDB.

Initialise the FDB input.

Parameters:

• context (Context) – The context for the input.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• fdb_config (dict, optional) – The FDB config to use.

• fdb_userconfig (dict, optional) – The FDB userconfig to use.

• variables (list[str] | None) – List of variables to be handled by the input, or None for a sensible default variables.

• pre_processors (list[ProcessorConfig], optional) – Pre-processors to apply to the retrieved data.

• namer (Optional[Any]) – Optional namer for the input.

• purpose (str, optional) – The purpose of the input.

• kwargs (dict, optional) – Additional keyword arguments for the request to FDB.

trace_name = 'fdb'

create_input_state(*, dates: list[str | datetime | int], **kwargs) → dict[str, Any]

Create the input state dictionary.

Parameters:

• dates (list[Date]) – The list of dates for which to create the input state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The input state dictionary.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load forcings (constant and dynamic).

Parameters:

• dates (List[Date]) – The list of dates for which to load the forcings.

• current_state (State) – The current state of the model.

Returns:

The updated state with the loaded forcings.

Return type:

State

retrieve(variables: list[str], dates: list[str | datetime | int]) → Any

grib

class anemoi.inference.inputs.grib.GribInput(context: Context, metadata: Metadata, *, namer: Any | None = None, **kwargs)

Bases: EkdInput

Handles GRIB input fields.

Initialize the EkdInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• namer (Optional[Union[Callable[[Any, Dict[str, Any]], str], Dict[str, Any]]]) – Optional namer for the input.

set_private_attributes(state: Any, fields: FieldList) → None

Set private attributes for the state.

Parameters:

• state (Any) – The state to set private attributes for.

• fields (ekd.FieldList) – The input fields.

gribfile

class anemoi.inference.inputs.gribfile.GribFileInput(*args, **kwargs)

Bases: FieldlistInput, GribInput

Handles grib files.

Initialise the FieldlistInput.

Parameters:

• context (Any) – The context in which the input is used.

• path (str) –

  Path, directory or glob pattern to file(s). Examples:

  • ”/path/to/file.grib”

  • ”/path/to/*.grib”

  • ”/path/to/**/*.grib2”

  • ”/path/to/directory/”

• **kwargs (Any) – Additional keyword arguments.

trace_name = 'grib file'

patterns: tuple[str, ...] = ('*.grib', '*.grb', '*.grb2', '*.grib2')

mars

anemoi.inference.inputs.mars.rounded_area(area: list[float] | None) → list[float] | None

Round the area to a global extent if the surface is greater than 0.98.

Parameters:

area (Optional[List[float]]) – The area to be rounded.

Returns:

The rounded area or the original area if no rounding is needed.

Return type:

Optional[List[float]]

anemoi.inference.inputs.mars.grid_is_valid(grid: str | list[float] | None) → bool

Check if the grid is valid.

Parameters:

grid (Optional[Union[str, List[float]]]) – The grid to be checked.

Returns:

True if the grid is valid, False otherwise.

Return type:

bool

anemoi.inference.inputs.mars.area_is_valid(area: list[float] | None) → bool

Check if the area is valid.

Parameters:

area (Optional[List[float]]) – The area to be checked.

Returns:

True if the area is valid, False otherwise.

Return type:

bool

anemoi.inference.inputs.mars.postproc(grid: str | list[float] | None, area: list[float] | str | None) → dict[str, str | list[float]]

Post-process the grid and area.

Parameters:

• grid (Optional[Union[str, List[float]]]) – The grid to be post-processed.

• area (Optional[Union[List[float], str]]) – The area to be post-processed.

Returns:

The post-processed grid and area.

Return type:

Dict[str, Union[str, List[float]]]

anemoi.inference.inputs.mars.retrieve(requests: list[dict[str, Any]], grid: str | list[float] | None, area: list[float] | None, patch: Any | None = None, log: bool = True, **kwargs: Any) → Any

Retrieve data from MARS.

Parameters:

• requests (List[Dict[str, Any]]) – The list of requests to be retrieved.

• grid (Optional[Union[str, List[float]]]) – The grid for the retrieval.

• area (Optional[List[float]]) – The area for the retrieval.

• patch (Optional[Any], optional) – Optional patch for the request, by default None.

• log (bool, optional) – Whether to log the requests, by default True.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The retrieved data.

Return type:

Any

class anemoi.inference.inputs.mars.MarsInput(context: Context, metadata: Metadata, *, variables: list[str] | None = None, patches: list[tuple[dict[str, Any], dict[str, Any]]] | None = None, log: bool = True, pre_processors: list[str | dict[str, Any]] | None = None, namer: Any | None = None, purpose: str | None = None, **kwargs: Any)

Bases: GribInput

Get input fields from MARS.

Initialize the MarsInput.

Parameters:

• context (Any) – The context in which the input is used.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• variables (list[str] | None) – List of variables to be handled by the input, or None for a sensible default variables.

• namer (Optional[Any]) – Optional namer for the input.

• patches (Optional[List[Tuple[Dict[str, Any], Dict[str, Any]]]]) – Optional list of patches for the input.

• log (bool) – Whether to log the requests to MARS, by default True.

• **kwargs (Any) – Additional keyword to pass to the request to MARS.

trace_name = 'mars'

create_input_state(*, dates: list[str | datetime | int], ref_date_index=-1, **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• dates (list[Date]) – The list of dates for which to create the input state.

• ref_date_index (int = -1) – The index in dates to use as reference date for the state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

State

retrieve(variables: list[str], dates: list[str | datetime | int]) → Any

Retrieve data for the given variables and dates.

Parameters:

• variables (List[str]) – The list of variables to retrieve.

• dates (List[Any]) – The list of dates for which to retrieve the data.

Returns:

The retrieved data.

Return type:

Any

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for the given variables and dates.

Parameters:

• dates (List[Date]) – The list of dates for which to load the forcings state.

• current_state (State) – The current state to be updated with the loaded forcings state.

Returns:

The loaded forcings state.

Return type:

Any

patch(request: dict[str, Any]) → dict[str, Any]

Patch the given request with predefined patches.

Parameters:

request (Dict[str, Any]) – The request to be patched.

Returns:

The patched request.

Return type:

Dict[str, Any]

default_initial_date() → datetime

Return the initial date to use when none is given in the config (assuming a forecasting context).

The runner calls this when date isn’t set in the config or on the CLI. Each input decides how to pick the date: some read it from the data (e.g. the latest valid time in a file), others use a fixed rule (e.g. yesterday at 00:00), and some don’t implement this method at all.

netcdf

class anemoi.inference.inputs.netcdf.NetCDFFileInput(*args, **kwargs)

Bases: FieldlistInput

Handles netcdf files.

Initialise the FieldlistInput.

Parameters:

• context (Any) – The context in which the input is used.

• path (str) –

  Path, directory or glob pattern to file(s). Examples:

  • ”/path/to/file.grib”

  • ”/path/to/*.grib”

  • ”/path/to/**/*.grib2”

  • ”/path/to/directory/”

• **kwargs (Any) – Additional keyword arguments.

trace_name = 'netcdf file'

patterns: tuple[str, ...] = ('*.nc', '*.netcdf')

opendap

repeated_dates

class anemoi.inference.inputs.repeated_dates.RepeatedDatesInput(context: Context, metadata: Metadata, *, source: str, mode: str = 'constant', **kwargs: Any)

Bases: Input

This class is identical to the one used to in anemoi-datasets/create It uses a source of constants (e.g. a source containing the bathymetry) available only for a given date and returns its content whever date is requested by the runner

Initialise the Input object.

Parameters:

• context (Context) – The context for the input.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• variables (list of str or None) – List of variable names to be handled by the input, or None for all available variables.

• pre_processors (list of ProcessorConfig or None, optional) – List of pre-processors to apply to the input. If None, no pre-processing is performed.

• purpose (str or None, optional) – The purpose of the input (e.g., ‘forcings’, ‘constants’). Used for debugging and logging.

trace_name = 'repeated dates'

create_input_state(*, dates: list[str | datetime | int], ref_date_index=-1, **kwargs) → dict[str, Any]

Create the input state for the repeated-dates input.

Parameters:

• dates (list[Date]) – The dates for the input state.

• ref_date_index (int = -1) – The index in dates to use as reference date for the state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for repeated dates input.

Parameters:

• dates (list[Date]) – The list of dates for which to repeat the fields.

• current_state (State) – The current state to use for loading.

Returns:

The loaded and repeated forcings state.

Return type:

State

split

class anemoi.inference.inputs.split.SplitInput(context: Context, metadata: Metadata, *splits, **kwargs: Any)

Bases: Input

Initialise the Input object.

Parameters:

• context (Context) – The context for the input.

• metadata (Metadata) – Metadata corresponding to the dataset this input is handling.

• variables (list of str or None) – List of variable names to be handled by the input, or None for all available variables.

• pre_processors (list of ProcessorConfig or None, optional) – List of pre-processors to apply to the input. If None, no pre-processing is performed.

• purpose (str or None, optional) – The purpose of the input (e.g., ‘forcings’, ‘constants’). Used for debugging and logging.

trace_name = 'split input'

create_input_state(*, dates: list[str | datetime | int], ref_date_index=-1, **kwargs) → dict[str, Any]

Create the input state for the repeated-dates input.

Parameters:

• dates (list[Date]) – The dates for the input state.

• ref_date_index (int = -1) – The index in dates to use as reference date for the state.

• **kwargs (Any) – Additional keyword arguments.

Returns:

The created input state.

Return type:

State

load_forcings_state(*, dates: list[str | datetime | int], current_state: dict[str, Any]) → dict[str, Any]

Load the forcings state for repeated dates input.

Parameters:

• dates (list[Date]) – The list of dates for which to repeat the fields.

• current_state (State) – The current state to use for loading.

Returns:

The loaded and repeated forcings state.

Return type:

State