import pickle import torch import numpy as np import torch.nn as nn from set_convs import convDeepSet from unet_wrap_padding import Unet from vit import * from models import * hadisd_publisher_shifts = { "tas": 273.15, "u": 0.0, "v": 0.0, "psl": 0.0, "ws": 0.0, } hadisd_publisher_scales = { "tas": 10, "u": 10, "v": 10, "psl": 100, "ws": 10.0, } def hadisd_normalisation_factors(var: str): path = "/home/azureuser/aux_data/norm_factors/" return { "mean": np.load(path + f"mean_hadisd_{var}.npy"), "std": np.load(path + f"std_hadisd_{var}.npy"), } def unnormalise_hadisd_var(x, var): factors = hadisd_normalisation_factors(var) hadisd_shift = hadisd_publisher_shifts[var] hadisd_scale = hadisd_publisher_scales[var] return hadisd_shift + hadisd_scale * (factors["mean"] + factors["std"] * x) class DownscalingRmseLoss(nn.Module): def __init__(self): super().__init__() def forward(self, target, output, prev_step, fix_sigma=None, expand=True): target = torch.flatten(target.to(output.device)) output = torch.flatten(output) tmp = torch.isnan(target) clean_target = target[~tmp] clean_output = output[~tmp] return torch.mean((clean_target - clean_output) ** 2) class ConvCNPWeatherOnToOff(nn.Module): """ ConvCNP for decoder """ def __init__( self, in_channels, out_channels, int_channels, device, res, data_path="../data/", mode="end_to_end", decoder=None, film=False, ): super().__init__() # Setup self.device = device self.in_channels = in_channels self.out_channels = out_channels self.int_channels = int_channels self.decoder = decoder self.int_x = 256 self.int_y = 128 self.mode = mode self.film = film # Load lon-lat of internal discretisation self.era5_x = ( torch.from_numpy( np.load(data_path + "grid_lon_lat/era5_x_{}.npy".format(res)) ).float() / 360 ) self.era5_y = ( torch.from_numpy( np.load(data_path + "grid_lon_lat/era5_y_{}.npy".format(res)) ).float() / 360 ) # Setup setconv self.sc_out = convDeepSet( 0.001, "OnToOff", density_channel=False, device=self.device ) if self.mode not in ["downscaling", "end_to_end"]: unet_out_channels = out_channels else: unet_out_channels = int_channels # UNet backbone if self.decoder == "base": self.decoder_lr = Unet( in_channels=in_channels, out_channels=unet_out_channels, div_factor=1, film=film, ) else: raise Exception(f"Expected to use base decoder, but got {self.decoder}") # Postprocessing MLP self.mlp = DownscalingMLP( in_channels=24 + 9, out_channels=1, h_channels=64, h_layers=2, ) def forward(self, task, film_index): x = task["y_context"] batch_size = x.shape[0] # UNet backbone x = self.decoder_lr(x, film_index=task["lt"]) # Transform to station predictions with setconv num_channels = x.shape[3] x = x.permute(0, 3, 1, 2) assert list(x.shape) == [batch_size, num_channels, 240, 121] x_target = task["x_target"] num_stations = x_target.shape[2] x = self.sc_out( x_in=task["x_context"], wt=x, x_out=[x_target[:, 0, :], x_target[:, 1, :]], ) assert x.shape[0] == batch_size assert x.shape[2] == num_stations # Concatenate auxiliary data at stations alt_target = task["alt_target"] assert torch.isnan(alt_target).sum() == 0 assert alt_target.shape[0] == batch_size assert alt_target.shape[2] == num_stations aux_time = task["aux_time"].squeeze(-1).repeat(1, 1, num_stations) assert aux_time.shape[0] == batch_size assert aux_time.shape[2] == num_stations x = torch.cat([x, alt_target, x_target, aux_time], dim=1).permute(0, 2, 1) assert x.shape[0] == batch_size assert x.shape[1] == num_stations tmp = self.mlp(x) assert list(tmp.shape) == [batch_size, num_stations, 1] y_hat = tmp.squeeze(-1) assert list(y_hat.shape) == [batch_size, num_stations] return y_hat class ResidualBlock(nn.Module): def __init__(self, n_channels): super().__init__() self.block = nn.Sequential(nn.Linear(n_channels, n_channels), nn.ReLU()) def forward(self, x): return self.block(x) + x class DownscalingMLP(nn.Module): """ MLP for handling auxiliary data at station locations """ def __init__(self, in_channels, out_channels, h_channels, h_layers): super().__init__() self.mlp = nn.Sequential( nn.Linear(in_channels, h_channels), *[ResidualBlock(h_channels) for _ in range(h_layers)], nn.Linear(h_channels, out_channels), ) def forward(self, x): return self.mlp(x) def find_epoch(results_dir: str) -> int: losses = np.load(results_dir + "losses_0.npy") epoch = np.argmin(losses) return epoch def load_config(results_dir: str) -> dict: with open(results_dir + "config.pkl", "rb") as f: return pickle.load(f) def load_model(results_dir, device, config_dir=None, epoch=None): """ Load the ConvCNPWeatherOnToOff to generate forecasts """ config_dir = config_dir or results_dir config = load_config(config_dir) epoch = epoch or find_epoch(results_dir) model = ConvCNPWeatherOnToOff( in_channels=config["in_channels"], out_channels=config["out_channels"], int_channels=config["int_channels"], device=device, res=config["res"], decoder=config["decoder"], mode=config["mode"], film=False, ) full_state_dict = torch.load(results_dir + f"epoch_{epoch}", map_location=device) 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(device) model.eval() return model def load_model_e2e(results_dir, lead_time, device): """ Load the full E2E model to generate forecasts """ config = load_config(results_dir) epoch = find_epoch(results_dir) model = ConvCNPWeatherE2E( device="cuda", lead_time=lead_time, se_model_path="../results/assimilation/all_v4_vit_ps3", forecast_model_path="../results/forecast/vit_fnl_randomlt_mlp", sf_model_path="../results/downscaling_NEW", ) full_state_dict = torch.load(results_dir + f"epoch_{epoch}", map_location=device) 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(device) model.eval() return model