NEDAS.core.perturb module

class NEDAS.core.perturb.PerturbField(**kwargs)

Bases: object

Generates and applies perturbation on given 2D field(s)

mask: ndarray

perturb_type: str

other_opts: list[str] = []

params: dict[str, dict[str, Any]] = {}

grid: RegularGrid | IrregularGrid | Grid1D

perturb_methods: dict[str, Callable]

parse_perturb_opts(**kwargs) → None

generate_perturb(grid: RegularGrid | IrregularGrid | Grid1D, fields: dict[str, ndarray], prev_perturb: dict[str, Any], dt: float = 1, n: int = 0) → dict[str, ndarray]

Add random perturbation to the given 2D fields

Parameters:

• grid (GridType) – Grid object describing the 2d domain

• fields (dict[str, np.ndarray]) – the input fields

• prev_perturb (dict[str, Any]) – previous perturbation data, dict[str, None] if unavailable

• dt (float) – interval (hours) between time steps

• n (int)

Returns:

the generated perturbations

Return type:

dict[str, np.ndarray]

add_perturb(fields: dict[str, ndarray], perturb: dict[str, ndarray], **kwargs) → dict[str, ndarray]

Add perturbations to each field

perturb_random_gaussian(rec: dict[str, Any], s: int) → ndarray

Generate a random perturbation using the Gaussian random field method

perturb_random_powerlaw(rec: dict[str, Any], s: int) → ndarray

Generate a random perturbation using the powerlaw method

perturb_random_displace(rec: dict[str, Any], s: int) → ndarray

Generate a random perturbation using the displacement method (returns a vector field)

make_correlated_perturb(prev_perturb: ndarray, perturb: ndarray, corr: float) → ndarray

Create perturbations that are correlated in time

make_wind_perturb_from_press(perturb: dict[str, ndarray]) → dict[str, ndarray]

Legacy option in TOPAZ prsflg==1,2 options in force_perturb program, reproduced here. Expecting the vnames ‘atmos_surf_press’ for pressure field and ‘atmos_surf_velocity’ for the wind field. Derive the wind perturbation from pressure perturbations, so that they are in wind-pressure relation (prsflg==1) Additionally, derived wind perturbations are rescaled to match the specified amp (prsflg==2)

class NEDAS.core.perturb.Perturbation(c: Context)

Bases: object

Perturbation top-level manager

nfld: int = 0

perturb: dict[str, Any] = {}

task_list: dict[int, list[dict]] = {}

distribute_perturb_tasks(c: Context) → dict[int, list[dict]]

count_num_fields(c: Context)

init_file_locks(c: Context) → None

Initialize file locks for nc-backed perturb variables (e.g. iced, iceh).

Mirrors the updator pattern: register only the output file paths THIS rank will personally write, then call build_file_locks(), which does its own internal allgather to reconstruct the full per-file writer ordering (point-to-point handoff chain, see utils/parallel.py).

The file-collection loop is wrapped in try/except so that a rank that fails mid-loop still participates in the collective allgather (inside build_file_locks) and Barrier, preventing an MPI deadlock. Any error is re-raised after the barrier so the outer try/except in __call__ can handle it uniformly.

rec_dt(model, variable_list: list) → float

Return the shared dt for all variables in a perturb record.

Raises ValueError if variables have different dt, since one time loop cannot serve variables with different temporal resolutions.

nstep(c: Context, model, dt: float) → int

Number of time steps to perturb for a variable with the given dt.

A variable is BC/forcing if its dt is strictly less than the model restart_dt (sub-restart temporal resolution). It is perturbed at every dt step over the forecast period. IC/restart variables are perturbed once at t=0.

Uses model.restart_dt as the threshold when available; falls back to cycle_period so models without restart_dt get the conservative nstep=1.

prepare_perturb_dir(c)

Prepare and clear the directory where perturbation data will be stored (offline mode)

save_perturb_data(c: Context, **rec)

Save a copy of perturbation data, for use by the next analysis cycle

load_perturb_data(c: Context, **rec)

Load the perturbation data

collect_fields(c: Context, t: datetime, k: int, **rec) → dict[str, ndarray]

Collect all model fields to be perturbed

output_perturbed_fields(c: Context, fields: dict[str, ndarray], t: datetime, k: int, **rec) → None