    def __perform_extrapolation(self):
        """
        Extrapolate (advect) precipitation fields based on the velocity field to align
        them in time. This prepares the precipitation fields for autoregressive modeling.
        """
        # Determine the precipitation threshold mask if conditional is set
        if self.__config.conditional:
            self.__state.mask_threshold = np.logical_and.reduce(
                [
                    self.__precip[i, :, :] >= self.__config.precip_threshold
                    for i in range(self.__precip.shape[0])
                ]
            )
        else:
            self.__state.mask_threshold = None

        extrap_kwargs = self.__state.extrapolation_kwargs.copy()
        extrap_kwargs["xy_coords"] = self.__params.xy_coordinates
        extrap_kwargs["allow_nonfinite_values"] = (
            True if np.any(~np.isfinite(self.__precip)) else False
        )

        res = []

        def __extrapolate_single_field(precip, i):
            # Extrapolate a single precipitation field using the velocity field
            return self.__params.extrapolation_method(
                precip[i, :, :],
                self.__velocity,
                self.__config.ar_order - i,
                "min",
                **extrap_kwargs,
            )[-1]

        for i in range(self.__config.ar_order):
            if (
                not DASK_IMPORTED
            ):  # If Dask is not available, perform sequential extrapolation
                self.__precip[i, :, :] = __extrapolate_single_field(self.__precip, i)
            else:
                # If Dask is available, accumulate delayed computations for parallel execution
                res.append(dask.delayed(__extrapolate_single_field)(self.__precip, i))

        # If Dask is available, perform the parallel computation
        if DASK_IMPORTED and res:
            num_workers_ = min(self.__params.num_ensemble_workers, len(res))
            self.__precip = np.stack(
                list(dask.compute(*res, num_workers=num_workers_))
                + [self.__precip[-1, :, :]]
            )

        print("Extrapolation complete and precipitation fields aligned.")
Extrapolate (advect) precipitation fields based on the velocity field to align
them in time. This prepares the precipitation fields for autoregressive modeling.
pysteps/nowcasts/steps.py