def _estimate_ar1_params(
    field_src, field_dst, estim_weights, interp_weights, num_workers=1
):
    """Constrained optimization of AR(1) parameters."""

    def objf(p, *args):
        i = args[0]
        field_ar = p * field_src
        return np.nansum(estim_weights[i] * (field_dst - field_ar) ** 2.0)

    bounds = (-0.98, 0.98)

    def worker(i):
        return opt.minimize_scalar(objf, method="bounded", bounds=bounds, args=(i,)).x

    if DASK_IMPORTED and num_workers > 1:
        res = []
        for i in range(len(estim_weights)):
            res.append(dask.delayed(worker)(i))

        psi = dask.compute(*res, num_workers=num_workers, scheduler="threads")
    else:
        psi = []
        for i in range(len(estim_weights)):
            psi.append(worker(i))

    return [np.sum([psi_ * interp_weights[i] for i, psi_ in enumerate(psi)], axis=0)]
Constrained optimization of AR(1) parameters.
pysteps/nowcasts/linda.py