official-src/aardvark/e2e_model.py

import pickle

import torch
import torch.nn as nn
import numpy as np

from models import ConvCNPWeather
from misc_downscaling_functionality import ConvCNPWeatherOnToOff


class ConvCNPWeatherE2E(nn.Module):
    """
    Complete Aardvark weather model. This chains together the trained encoder,
    processor and decoder modules to create complete forecasts. It can be finetuned
    end to end to optimise predictions for a specific variable and location.
    """

    def __init__(
        self,
        device,
        lead_time,
        se_model_path,
        forecast_model_path,
        sf_model_path,
        return_gridded=False,
        aux_data_path=None,
    ):

        super().__init__()

        self.device = device
        self.lead_time = lead_time
        self.return_gridded = return_gridded

        # Load encoder
        self.se_model = self.load_se_model(se_model_path)

        # Load processor
        self.forecast_model = nn.ModuleList(
            [
                self.load_forecast_model(forecast_model_path, l + 1)
                for l in range(lead_time)
            ]
        )

        # Load decoder
        self.sf_model = self.load_sf_model(sf_model_path, lead_time)
        self.decoder = None

        # Setup normalisation factors
        self.forecast_input_means = (
            self.to_tensor(np.load(aux_data_path + "norm_factors/mean_4u_1.npy"))
            .unsqueeze(0)
            .unsqueeze(0)
            .unsqueeze(0)
        )
        self.forecast_input_stds = (
            self.to_tensor(np.load(aux_data_path + "norm_factors/std_4u_1.npy"))
            .unsqueeze(0)
            .unsqueeze(0)
            .unsqueeze(0)
        )

        self.forecast_pred_diff_means = (
            self.to_tensor(np.load(aux_data_path + "norm_factors/mean_diff_4u_1.npy"))
            .unsqueeze(0)
            .unsqueeze(0)
            .unsqueeze(0)
        )
        self.forecast_pred_diff_stds = (
            self.to_tensor(np.load(aux_data_path + "norm_factors/std_diff_4u_1.npy"))
            .unsqueeze(0)
            .unsqueeze(0)
            .unsqueeze(0)
        )

    def to_tensor(self, arr):
        return torch.from_numpy(arr).float().to(self.device)

    def load_se_model(self, se_model_path):
        """
        Load the trained encoder module
        """

        with open(se_model_path + "/config.pkl", "rb") as handle:
            forecast_config = pickle.load(handle)

        model = ConvCNPWeather(
            in_channels=forecast_config["in_channels"],
            out_channels=forecast_config["out_channels"],
            int_channels=forecast_config["int_channels"],
            device="cuda",
            res=forecast_config["res"],
            gnp=bool(0),
            decoder=forecast_config["decoder"],
            mode=forecast_config["mode"],
            film=bool(0),
        )

        best_epoch = np.argmin(np.load("{}/losses_0.npy".format(se_model_path)))
        state_dict = torch.load(
            "{}/epoch_{}".format(se_model_path, best_epoch),
            map_location="cuda",
        )["model_state_dict"]
        state_dict = {k[7:]: v for k, v in zip(state_dict.keys(), state_dict.values())}
        model.load_state_dict(state_dict)
        model = model.to("cuda")
        return model

    def load_forecast_model(self, forecast_model_path, lead_time):
        """
        Load the trained processor module
        """

        with open(forecast_model_path + "/config.pkl", "rb") as handle:
            forecast_config = pickle.load(handle)

        model = ConvCNPWeather(
            in_channels=forecast_config["in_channels"],
            out_channels=forecast_config["out_channels"],
            int_channels=forecast_config["int_channels"],
            device="cuda",
            res=forecast_config["res"],
            gnp=bool(0),
            decoder=forecast_config["decoder"],
            mode=forecast_config["mode"],
            film=False,
        )
        state_dict = torch.load(
            f"{forecast_model_path}/forecast_{lead_time}/epoch_0",
            map_location="cuda",
        )["model_state_dict"]
        state_dict = {k[7:]: v for k, v in zip(state_dict.keys(), state_dict.values())}
        model.load_state_dict(state_dict)
        model = model.to("cuda")
        return model

    def load_sf_model(self, sf_model_path, lead_time):
        """
        Load the trained decoder module
        """

        with open(sf_model_path + "config.pkl", "rb") as handle:
            config = pickle.load(handle)

        model = ConvCNPWeatherOnToOff(
            in_channels=config["in_channels"],
            out_channels=config["out_channels"],
            int_channels=config["int_channels"],
            device="cuda",
            res=config["res"],
            decoder=config["decoder"],
            mode=config["mode"],
            film=False,
        )

        best_epoch = np.argmin(
            np.load("{}/lt_{}/losses_0.npy".format(sf_model_path, lead_time))
        )
        full_state_dict = torch.load(
            sf_model_path + f"/lt_{lead_time}/epoch_{best_epoch}", map_location="cuda"
        )
        state_dict = full_state_dict["model_state_dict"]
        state_dict = {k[7:]: v for k, v in zip(state_dict.keys(), state_dict.values())}
        model.load_state_dict(state_dict)
        model = model.to("cuda")
        model.eval()

        return model

    def process_se_output(self, task, x):
        """
        Reshape and normalise encoder output for input to processor
        """

        task["forecast"]["y_context"][:, :24, ...] = x.permute(0, 3, 2, 1)
        if self.return_gridded:
            return task, x.permute(0, 3, 2, 1)
        return task

    def process_forecast_output(self, task, x, last=False):
        """
        Reshape and normalise processor output for input to decoder
        """

        base_context = task["forecast"]["y_context"][:, :-11, ...].permute(0, 2, 3, 1)

        base_context = (
            base_context * self.forecast_input_stds + self.forecast_input_means
        ).permute(0, 3, 2, 1)

        x = self.forecast_pred_diff_means + x * self.forecast_pred_diff_stds

        unnorm_x = x + base_context.permute(0, 2, 3, 1)

        forecast = unnorm_x

        x = (unnorm_x - self.forecast_input_means) / self.forecast_input_stds

        task["downscaling"]["y_context"][:, :24, ...] = x.permute(0, 3, 2, 1)
        task["forecast"]["y_context"] = torch.cat(
            [
                x.permute(0, 3, 2, 1),
                task["forecast"]["y_context"][:, 24:, ...],
            ],
            axis=1,
        )
        if self.return_gridded:
            return task, forecast
        return task

    def forward(self, task, film_index=None):
        """
        Produce a forecast
        """

        # Generate initial state
        x = self.se_model(task["assimilation"], film_index=None)
        if self.return_gridded:
            task, initial_state = self.process_se_output(task, x)
        else:
            task = self.process_se_output(task, x)

        # Generate forecast
        for lt in range(self.lead_time):
            x = self.forecast_model[lt](task["forecast"], film_index=None)
            if self.return_gridded:
                task, forecast = self.process_forecast_output(task, x)
            else:
                task = self.process_forecast_output(task, x)

        # Generate station forecast
        x = self.sf_model(task["downscaling"], film_index=None)

        if self.return_gridded:
            initial_state = (
                initial_state.permute(0, 3, 2, 1) * self.forecast_input_stds
                + self.forecast_input_means
            )
            return x, forecast, initial_state
        return x