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)

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