NEDAS.core.context module

class NEDAS.core.context.Context(config: Config | None = None, config_file: str | None = None, parse_args: bool = False, **kwargs)

Bases: object

Runtime context manages the generation and interaction of dynamic objects in runtime

comm: parallel.Comm

comm_rec: parallel.Comm

comm_mem: parallel.Comm

progress: progress.Progress

grid: grid.GridType

grid_orig: grid.GridType

models: dict[str, Model]

datasets: dict[str, Dataset]

assimilator: Assimilator

updator: Updator

transform_funcs: list[Transform]

localization_funcs: dict[str, Callable]

inflation_func: Inflation

state: State

obs: Obs

debug: bool

interactive: bool

is_notebook: bool

time: datetime

iter: int

pid_show: int

nens: int

mem_list: dict[ProcIDMem, list[MemID]]

fs: FileSystem

io: IOBackend

jsub: JobSubmitter

check_interactive() → bool

If the runtime environment supports interactive output (with ANSI escape code).

check_notebook() → bool

If the runtime environment is a jupyter notebook

get_cols() → int

set_logging() → None

distribute_mem_tasks() → dict[int, list[int]]

Distribute mem_id across processors

update_assim_tools()

Update the assimilation tool components based on runtime configuration

property prev_time: datetime

Previous analysis time. Automatically updated when self.time changes.

Returns:

Previous analysis time.

Return type:

datetime

property next_time: datetime

Next analysis time. Automatically updated when self.time changes.

Returns:

Next analysis time.

Return type:

datetime

set_comm() → None

Initialize the MPI communicator, split the communicator if necessary.

For serial program, use a dummy communicator, set nproc to the number of available processors on the machine; for MPI program, use MPI.COMM_WORLD and check if size matchs with nproc.

Split the communicator into member and record groups, according to nproc and nproc_mem. See NEDAS.utils.parallel module for more details.

set_grid() → None

Initialize the analysis grid based on the configuration.

If grid_def['type'] is ‘custom’, will create a analysis grid based on provided parameters. If grid_def['type'] is a model name, will load the grid from the specified model class.

set_models() → None

Initialize model instances based on model_def[model_name] settings. Store the model instances in models[model_name].

set_datasets() → None

Initialize dataset instances based on dataset_def[dataset_name] settings. Store the dataset instances in datasets[dataset_name].

property total_tasks

property current_task

property message

property debug_message

timer(func: Callable)

Decorator to count the elapsed time for a function at runtime.

logger(func_name: str)

Decorator to register the func in call stack and show runtime messages.

print_1p(msg: str)

Customized print function for showing runtime message.

Only the processor with PID = self.pid_show will show the message, this avoids the redundancy if all processors are showing the same message.

log_event(msg: str, flag='')

show_greeting() → None

show_summary() → None

dump_config(config_file: str) → None

Dumps a snapshot of the current state to a yaml config file. The original config object remains unchanged in memory.

run_job(commands: str, parallel_mode: Literal['serial', 'mpi', 'openmp'] = 'serial', nproc: int = 1, offset: int = 0, **kwargs) → None

The user-facing method for running command on a computer. It re-configures the existing job submitter with runtime arguments and execute the command.

Parameters:

• commands (str) – Shell commands to be dispatched by job submitter

• parallel_mode (ParallelMode, optional) – parallel mode (‘serial’, ‘mpi’, ‘openmp’), default is ‘serial’

• nproc (int, optional) – number of processors (default is 1)

• offset (int, optional) – offset in full list of processors (default is 0)

• **kwargs – other keyword arguments to update the job submitter configuration