def _generate_perturbations(pert_gen, num_workers, seed):
    """Generate perturbations based on the estimated forecast error statistics."""
    m, n = pert_gen["m"], pert_gen["n"]
    dist_param = pert_gen["dist_param"]
    std = pert_gen["std"]
    acf_fft_ampl = pert_gen["acf_fft_ampl"]
    weights = pert_gen["weights"]

    perturb = stats.norm.rvs(size=(m, n), random_state=seed)
    perturb_fft = np.fft.rfft2(perturb)

    out = np.zeros((m, n))

    def worker(i):
        if std[i] > 0.0:
            filtered_noise = np.fft.irfft2(acf_fft_ampl[i] * perturb_fft, s=(m, n))
            filtered_noise /= np.std(filtered_noise)
            filtered_noise = stats.lognorm.ppf(
                stats.norm.cdf(filtered_noise), *dist_param[i]
            )
        else:
            filtered_noise = np.ones(weights[i].shape)

        return weights[i] * filtered_noise

    if DASK_IMPORTED and num_workers > 1:
        res = []
        for i in range(weights.shape[0]):
            res.append(dask.delayed(worker)(i))
        res = dask.compute(*res, num_workers=num_workers, scheduler="threads")
        for r in res:
            out += r
    else:
        for i in range(weights.shape[0]):
            out += worker(i)

    return out
Generate perturbations based on the estimated forecast error statistics.
pysteps/nowcasts/linda.py