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.

trace_name = 'cds'

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

Create the input state for the given date.

Parameters:

• date (Optional[Date]) – The date 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

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_input_state(*, date: str | datetime | int | None, **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• date (Optional[Date]) – The date 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, **kwargs: Any)

Bases: Input

Handles anemoi-datasets dataset as input.

property ds: Any

Return 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(*, date: str | datetime | int | None = None, constant: bool = False, **kwargs) → dict[str, Any]

Create the input state for the given date.

Parameters:

• date (Optional[Any]) – The date for which to create the input state.

• constant (bool) – Whether the field is constant or dynamic

• **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, **kwargs: Any)

Bases: DatasetInput

Handles anemoi-datasets dataset as input.

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.

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.

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.

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.

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.

trace_name = 'dummy'

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

Create the input state for the given date.

Parameters:

• date (Optional[Date]) – The date 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

ekd

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

Find a variable in an earthkit FieldList/FieldArray.

Parameters:

• data (Any) – The data to search (FieldList or FieldArray).

• 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 (FieldArray subset).

Return type:

Any

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.

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

Substitute placeholders in the template with metadata values.

Parameters:

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

• field (Any) – 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.

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 (FieldList/FieldArray).

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.

trace_name = 'empty'

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

Create an empty input state.

Parameters:

• date (Date or None) – The date 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 of 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, **kwargs: dict[str, Any])

Bases: GribInput

Get input fields from FDB.

trace_name = 'fdb'

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

Create the input state dictionary.

Parameters:

• date (Optional[Date]) – The date 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.

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.

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.

trace_name = 'mars'

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

Create the input state for the given date.

Parameters:

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

• ref_date_index (int = -1) – If 0 takes the first date, if -1 takes the last date in sequence.

• **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]

netcdf

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

Bases: FieldlistInput

Handles netcdf files.

trace_name = 'netcdf file'

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

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

trace_name = 'repeated dates'

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

Create the input state for the repeated-dates input.

Parameters:

• date (Date or None) – The date for 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 repeated dates input.

Parameters:

• dates (list of 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

trace_name = 'split input'

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

Create the input state for the repeated-dates input.

Parameters:

• date (Date or None) – The date for 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 repeated dates input.

Parameters:

• dates (list of 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