    def __update_state(self, state, params):
        """
        Update the state during the nowcasting loop. This function handles the AR model iteration,
        noise generation, recomposition, and mask application for each ensemble member.
        """
        precip_forecast_out = [None] * params["n_ens_members"]

        # Update the deterministic AR(p) model if noise or sprog mask is used
        if params["noise_method"] is None or params["mask_method"] == "sprog":
            self.__update_deterministic_ar_model(state, params)

        # Worker function for each ensemble member
        def worker(j):
            self.__apply_ar_model_to_cascades(j, state, params)
            precip_forecast_out[j] = self.__recompose_and_apply_mask(j, state, params)

        # Use Dask for parallel execution if available
        if (
            DASK_IMPORTED
            and params["n_ens_members"] > 1
            and params["num_ensemble_workers"] > 1
        ):
            res = []
            for j in range(params["n_ens_members"]):
                res.append(dask.delayed(worker)(j))
            dask.compute(*res, num_workers=params["num_ensemble_workers"])
        else:
            for j in range(params["n_ens_members"]):
                worker(j)

        return np.stack(precip_forecast_out), state
Update the state during the nowcasting loop. This function handles the AR model iteration,
noise generation, recomposition, and mask application for each ensemble member.
pysteps/nowcasts/steps.py