tasks

runner

class anemoi.inference.tasks.runner.CouplingForcings(context, input, variables, mask)

Bases: CoupledForcings

Just to have a different __repr__.

class anemoi.inference.tasks.runner.CoupledTensorHandler(context: Runner, metadata: Metadata, constant_forcings_input: Input, dynamic_forcings_input: Input, boundary_forcings_input: Input, trace_path: str | None = None)

Bases: TensorHandler

create_dynamic_coupled_forcings(variables: list[str], mask: ndarray[tuple[Any, ...], dtype[Any]]) → list[CoupledForcings]

initial_dynamic_forcings_providers(dynamic_forcings_providers: list[Forcings]) → list[Forcings]

Modify the dynamic forcings providers for the first step. For the initial state we need to load the dynamic coupled forcings from the default provider (i.e.: from disk).

class anemoi.inference.tasks.runner.CoupledRunner(config: RunConfiguration, coupled_input: CoupledInput)

Bases: Runner

Runner for coupled models.

This class handles the initialization and running of coupled models using the provided configuration and input.

input_states_hook(input_states: dict[str, dict[str, Any]]) → None

Hook used by coupled runners to send the input state.

output_states_hook(output_states: dict[str, dict[str, Any]]) → None

Hook used by coupled runners to send the input state.

class anemoi.inference.tasks.runner.CoupledInput(task: Task, transport: Transport, couplings: list[Coupling])

Bases: object

Input handler for coupled models.

This class manages the input data and state for coupled models, including loading and initializing forcings.

trace_name = 'coupled'

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

initial_state(state: dict[str, Any]) → None

output_state(state: dict[str, Any]) → None

class anemoi.inference.tasks.runner.NoModelCoupledRunner(config: RunConfiguration, coupled_input: CoupledInput)

Bases: NoModelMixing, CoupledRunner

Runner for testing coupled models.

class anemoi.inference.tasks.runner.RunnerTask(task_name: str, config: dict[str, Any], overrides: dict[str, Any] = {}, global_config: dict[str, Any] = {})

Bases: Task

Task for running coupled models.

This task initializes and runs coupled models using the provided configuration, overrides, and global configuration.

run(transport: Transport) → None

Run the task.