    def __prepare_radar_and_NWP_fields(self):
        """
        Prepare radar and NWP precipitation fields for nowcasting.
        This includes generating a threshold mask, transforming fields into
        Lagrangian coordinates, cascade decomposing/recomposing, and checking
        for zero-precip areas. The results are stored in class attributes.
        """
        # determine the precipitation threshold mask
        if self.__config.conditional:
            # TODO: is this logical_and correct here? Now only those places where precip is in all images is saved?
            self.__params.mask_threshold = np.logical_and.reduce(
                [
                    self.__precip[i, :, :] >= self.__config.precip_threshold
                    for i in range(self.__precip.shape[0])
                ]
            )
        else:
            self.__params.mask_threshold = None

        # we need to know the zerovalue of precip to replace the mask when decomposing after
        # extrapolation
        self.__params.precip_zerovalue = np.nanmin(self.__precip)

        # 1. Start with the radar rainfall fields. We want the fields in a
        # Lagrangian space

        # Advect the previous precipitation fields to the same position with the
        # most recent one (i.e. transform them into the Lagrangian coordinates).

        self.__state.extrapolation_kwargs["xy_coords"] = self.__params.xy_coordinates
        res = []

        def transform_to_lagrangian(precip, i):
            return self.__params.extrapolation_method(
                precip[i, :, :],
                self.__velocity,
                self.__config.ar_order - i,
                "min",
                allow_nonfinite_values=True,
                **self.__state.extrapolation_kwargs.copy(),
            )[-1]

        if not DASK_IMPORTED:
            # Process each earlier precipitation field directly
            for i in range(self.__config.ar_order):
                self.__precip[i, :, :] = transform_to_lagrangian(self.__precip, i)
        else:
            # Use Dask delayed for parallelization if DASK_IMPORTED is True
            for i in range(self.__config.ar_order):
                res.append(dask.delayed(transform_to_lagrangian)(self.__precip, i))
            num_workers_ = (
                len(res)
                if self.__config.num_workers > len(res)
                else self.__config.num_workers
            )
            self.__precip = np.stack(
                list(dask.compute(*res, num_workers=num_workers_))
                + [self.__precip[-1, :, :]]
            )

        # Replace non-finite values with the minimum value for each field
        self.__precip = self.__precip.copy()
        for i in range(self.__precip.shape[0]):
            self.__precip[i, ~np.isfinite(self.__precip[i, :])] = np.nanmin(
                self.__precip[i, :]
            )

        # Perform the cascade decomposition for the input precip fields and,
        # if necessary, for the (NWP) model fields
        # Compute the cascade decompositions of the input precipitation fields
        precip_forecast_decomp = []
        for i in range(self.__config.ar_order + 1):
            precip_forecast = self.__params.decomposition_method(
                self.__precip[i, :, :],
                self.__params.bandpass_filter,
                mask=self.__params.mask_threshold,
                fft_method=self.__params.fft,
                output_domain=self.__config.domain,
                normalize=True,
                compute_stats=True,
                compact_output=True,
            )
            precip_forecast_decomp.append(precip_forecast)

        # Rearrange the cascaded into a four-dimensional array of shape
        # (n_cascade_levels,ar_order+1,m,n) for the autoregressive model
        self.__state.precip_cascades = nowcast_utils.stack_cascades(
            precip_forecast_decomp, self.__config.n_cascade_levels
        )

        precip_forecast_decomp = precip_forecast_decomp[-1]
        self.__state.mean_extrapolation = np.array(precip_forecast_decomp["means"])
        self.__state.std_extrapolation = np.array(precip_forecast_decomp["stds"])

        # If necessary, recompose (NWP) model forecasts
        self.__state.precip_models_cascades = None

        if self.__params.precip_models_provided_is_cascade:
            self.__state.precip_models_cascades = self.__precip_models
            # Inline logic of _compute_cascade_recomposition_nwp
            temp_precip_models = []
            for i in range(self.__precip_models.shape[0]):
                precip_model = []
                for time_step in range(self.__precip_models.shape[1]):
                    # Use the recomposition method to rebuild the rainfall fields
                    recomposed = self.__params.recomposition_method(
                        self.__precip_models[i, time_step]
                    )
                    precip_model.append(recomposed)
                temp_precip_models.append(precip_model)

            self.__precip_models = np.stack(temp_precip_models)

        # Check for zero input fields in the radar and NWP data.
        self.__params.zero_precip_radar = check_norain(
            self.__precip,
            self.__config.precip_threshold,
            self.__config.norain_threshold,
            self.__params.noise_kwargs["win_fun"],
        )
        # The norain fraction threshold used for nwp is the default value of 0.0,
        # since nwp does not suffer from clutter.
        self.__params.zero_precip_model_fields = check_norain(
            self.__precip_models,
            self.__config.precip_threshold,
            self.__config.norain_threshold,
            self.__params.noise_kwargs["win_fun"],
        )
Prepare radar and NWP precipitation fields for nowcasting.
This includes generating a threshold mask, transforming fields into
Lagrangian coordinates, cascade decomposing/recomposing, and checking
for zero-precip areas. The results are stored in class attributes.
pysteps/blending/steps.py