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

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.

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.

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