grib

Subpackages

grib templates

encoding

anemoi.inference.grib.encoding.encode_time_processing(*, result: dict[str, Any], template: Field, variable: Variable, date: datetime, step: timedelta, previous_step: timedelta | None, start_steps: dict[str, timedelta], edition: int, ensemble: bool) → None

Encode time processing information into the result dictionary.

Parameters:

result (dict[str, Any]) – The result dictionary to update.
template (ekd.Field) – The template field.
variable (Variable) – The variable containing time processing information.
date (datetime) – The date and time.
step (timedelta) – The current step.
previous_step (timedelta | None) – The previous step.
start_steps (dict[str, timedelta]) – The start steps dictionary.
edition (int) – The GRIB edition.
ensemble (bool) – Whether the data is part of an ensemble.

anemoi.inference.grib.encoding.grib_keys(*, values: ndarray[tuple[Any, ...], dtype[Any]], template: Field, variable: Variable, ensemble: bool, param: int | float | str | None, date: datetime, step: timedelta, previous_step: timedelta | None, start_steps: dict[str, timedelta], keys: dict[str, Any], grib1_keys: dict[int | float | str, dict[str, Any]] = {}, grib2_keys: dict[int | float | str, dict[str, Any]] = {}) → dict[str, Any]

Generate GRIB keys for encoding.

Parameters:

values (FloatArray) – The values to encode.
template (ekd.Field) – The template to use.
variable (Variable) – The variable containing GRIB keys.
ensemble (bool) – Whether the data is part of an ensemble.
param (int | float | str | None) – The parameter value.
date (datetime) – The date and time.
step (Any) – The current step.
previous_step (timedelta | None) – The previous step.
start_steps (dict[str, timedelta]) – The start steps dictionary.
keys (dict[str, Any]) – The initial keys dictionary.
grib1_keys (dict[int | float | str, dict[str, Any]], optional) – Additional GRIB1 keys.
grib2_keys (dict[int | float | str, dict[str, Any]], optional) – Additional GRIB2 keys.

Returns:

The generated GRIB keys.

Return type:

dict[str, Any]

anemoi.inference.grib.encoding.check_encoding(handle: Any, keys: dict[str, Any], first: bool = True) → None

Check if the GRIB encoding matches the expected keys.

Parameters:

handle (Any) – The GRIB handle.
keys (Dict[str, Any]) – The expected keys.
first (bool, optional) – Whether this is the first check.

Raises:

ValueError – If the GRIB field could not be encoded.

anemoi.inference.grib.encoding.encode_message(*, values: Any | None, template: Any, metadata: dict[str, Any], check_nans: bool = False, missing_value: int | float = 9999) → Any

Encode a GRIB message.

Parameters:

values (Optional[Any]) – The values to encode.
template (Any) – The template to use.
metadata (Dict[str, Any]) – The metadata for the GRIB message.
check_nans (bool, optional) – Whether to check for NaNs in the values.
missing_value (Union[int, float], optional) – The value to use for missing data.

Returns:

The encoded GRIB handle.

Return type:

Any

class anemoi.inference.grib.encoding.GribWriter(out: Path | IOBase, split_output: bool = True)

Bases: object

Write GRIB messages to one or more files.

close() → None: Close all open files.

write(*, values: Any | None, template: Any, metadata: dict[str, Any], check_nans: bool = False, missing_value: int | float = 9999) → tuple

Write a GRIB message to the target file.

Parameters:

values (Optional[Any]) – The values to encode.
template (Any) – The template to use.
metadata (Dict[str, Any]) – The metadata for the GRIB message.
check_nans (bool, optional) – Whether to check for NaNs in the values.
missing_value (Union[int, float], optional) – The value to use for missing data.

Returns:

The encoded GRIB handle and the file path.

Return type:

tuple

target(handle: Any) → tuple[IOBase, Path | str]

Determine the target file for the GRIB message.

Parameters:: handle (Any) – The GRIB handle.
Returns:: The file object and the file path.
Return type:: tuple