"""ClimODE's official global neural transport model. The equations and layer layout follow Aalto-QuML/ClimODE. The small wrapper at the bottom adds configuration-friendly construction and checkpoint loading; the core ``ClimateEncoderFreeUncertain`` class intentionally keeps the official state and tensor layout. """ from __future__ import annotations import importlib from pathlib import Path import sys from typing import Any, Sequence import torch import torch.nn as nn import torch.nn.functional as F try: from torchdiffeq import odeint as _torchdiffeq_odeint except ImportError: # pragma: no cover - exercised when optional dependency is absent _torchdiffeq_odeint = None def _euler_odeint(func, y0: torch.Tensor, t: torch.Tensor) -> torch.Tensor: """Equivalent fixed-step Euler solver used only when torchdiffeq is absent.""" states = [y0] for index in range(1, len(t)): previous = states[-1] dt = t[index] - t[index - 1] states.append(previous + dt * func(t[index - 1], previous)) return torch.stack(states, dim=0) def odeint(func, y0: torch.Tensor, t: torch.Tensor, method: str = "euler", **kwargs): if _torchdiffeq_odeint is not None: return _torchdiffeq_odeint(func, y0, t, method=method, **kwargs) if method != "euler": raise ImportError( "torchdiffeq is required for solver=%r; install the official dependency." % method ) return _euler_odeint(func, y0, t) class OptimVelocity(nn.Module): """Learn the initial per-channel velocity used to start the ODE system.""" def __init__(self, num_years: int, height: int, width: int, out_channels: int = 5): super().__init__() self.out_channels = out_channels self.v_x = nn.Parameter( torch.randn(num_years, 1, out_channels, height, width) ) self.v_y = nn.Parameter( torch.randn(num_years, 1, out_channels, height, width) ) def forward(self, data: torch.Tensor): u_y = torch.gradient(data, dim=3)[0] u_x = torch.gradient(data, dim=4)[0] divergence = torch.gradient(self.v_y, dim=3)[0] + torch.gradient( self.v_x, dim=4 )[0] adv = self.v_x * u_x + self.v_y * u_y + data * divergence return adv, self.v_x, self.v_y class BoundaryPad(nn.Module): """Reflect at the poles and wrap around the longitude seam.""" def forward(self, value: torch.Tensor) -> torch.Tensor: return F.pad(F.pad(value, (0, 0, 1, 1), "reflect"), (1, 1, 0, 0), "circular") class ResidualBlock(nn.Module): def __init__(self, in_channels: int, out_channels: int): super().__init__() self.activation = nn.LeakyReLU(0.3) self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=0) self.bn1 = nn.BatchNorm2d(out_channels) self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=0) self.bn2 = nn.BatchNorm2d(out_channels) self.drop = nn.Dropout(p=0.1) self.shortcut = ( nn.Conv2d(in_channels, out_channels, kernel_size=1) if in_channels != out_channels else nn.Identity() ) def forward(self, value: torch.Tensor) -> torch.Tensor: value_padded = F.pad( F.pad(value, (0, 0, 1, 1), "reflect"), (1, 1, 0, 0), "circular" ) hidden = self.activation(self.bn1(self.conv1(value_padded))) hidden = F.pad( F.pad(hidden, (0, 0, 1, 1), "reflect"), (1, 1, 0, 0), "circular" ) hidden = self.activation(self.bn2(self.conv2(hidden))) return self.drop(hidden) + self.shortcut(value) class ClimateResNet2D(nn.Module): def __init__( self, num_channels: int, layers: Sequence[int], hidden_size: Sequence[int], ): super().__init__() if len(layers) != len(hidden_size): raise ValueError("layers and hidden_size must have equal lengths") self.layer_cnn = nn.ModuleList( [nn.Sequential(*blocks_for_layer) for blocks_for_layer in self._split_blocks(layers, hidden_size, num_channels)] ) @staticmethod def _split_blocks( layers: Sequence[int], hidden_size: Sequence[int], num_channels: int ) -> list[list[nn.Module]]: groups: list[list[nn.Module]] = [] in_channels = num_channels for repetitions, out_channels in zip(layers, hidden_size): group = [ResidualBlock(in_channels, out_channels)] group.extend( ResidualBlock(out_channels, out_channels) for _ in range(1, int(repetitions)) ) groups.append(group) in_channels = out_channels return groups def forward(self, value: torch.Tensor) -> torch.Tensor: output = value.float() for layer in self.layer_cnn: output = layer(output) return output class SelfAttnConv(nn.Module): """Official key-query-value attention convolution.""" def __init__(self, in_channels: int, out_channels: int): super().__init__() self.query = self._conv(in_channels, in_channels // 8, stride=1) self.key = self._key_conv(in_channels, in_channels // 8, stride=2) self.value = self._key_conv(in_channels, out_channels, stride=2) self.post_map = nn.Sequential( nn.Conv2d(out_channels, out_channels, kernel_size=1, stride=1, padding=0) ) self.out_ch = out_channels @staticmethod def _conv(n_in: int, n_out: int, stride: int) -> nn.Sequential: return nn.Sequential( BoundaryPad(), nn.Conv2d(n_in, n_in // 2, kernel_size=3, stride=stride, padding=0), nn.LeakyReLU(0.3), BoundaryPad(), nn.Conv2d(n_in // 2, n_out, kernel_size=3, stride=stride, padding=0), nn.LeakyReLU(0.3), BoundaryPad(), nn.Conv2d(n_out, n_out, kernel_size=3, stride=stride, padding=0), ) @staticmethod def _key_conv(n_in: int, n_out: int, stride: int) -> nn.Sequential: return nn.Sequential( nn.Conv2d(n_in, n_in // 2, kernel_size=3, stride=stride, padding=0), nn.LeakyReLU(0.3), nn.Conv2d(n_in // 2, n_out, kernel_size=3, stride=stride, padding=0), nn.LeakyReLU(0.3), nn.Conv2d(n_out, n_out, kernel_size=3, stride=1, padding=0), ) def forward(self, value: torch.Tensor) -> torch.Tensor: size = value.size() value = value.float() query = self.query(value).flatten(-2, -1) key = self.key(value).flatten(-2, -1) val = self.value(value).flatten(-2, -1) beta = F.softmax(torch.bmm(query.transpose(1, 2), key), dim=1) output = torch.bmm(val, beta.transpose(1, 2)) output = output.view(-1, self.out_ch, size[-2], size[-1]).contiguous() return self.post_map(output) class ClimateEncoderFreeUncertain(nn.Module): """Official global ``Climate_encoder_free_uncertain`` implementation.""" def __init__( self, num_channels: int = 5, const_channels: int = 2, out_types: int = 5, method: str = "euler", use_att: bool = True, use_err: bool = True, use_pos: bool = False, ): super().__init__() self.layers = [5, 3, 2] self.hidden = [128, 64, 2 * out_types] input_channels = 30 + out_types * int(use_pos) + 34 * (1 - int(use_pos)) self.vel_f = ClimateResNet2D(input_channels, self.layers, self.hidden) if use_att: self.vel_att = SelfAttnConv(input_channels, 10) self.gamma = nn.Parameter(torch.tensor([0.1])) self.scales = num_channels self.const_channel = const_channels self.out_ch = out_types self.past_samples: torch.Tensor | int = 0 self.const_info: torch.Tensor | int = 0 self.lat_map: torch.Tensor | int = 0 self.lon_map: torch.Tensor | int = 0 self.method = method err_in = 9 + out_types * int(use_pos) + 34 * (1 - int(use_pos)) if use_err: self.noise_net = ClimateResNet2D(err_in, [3, 2, 2], [128, 64, 2 * out_types]) if use_pos: self.pos_enc = ClimateResNet2D(4, [2, 1, 1], [32, 16, out_types]) self.att = use_att self.err = use_err self.pos = use_pos self.pos_feat: torch.Tensor | int = 0 self.lsm: torch.Tensor | int = 0 self.oro: torch.Tensor | int = 0 def update_param(self, params: Sequence[torch.Tensor]) -> None: if len(params) != 4: raise ValueError("update_param expects past_samples, constants, latitude, longitude") self.past_samples, self.const_info, self.lat_map, self.lon_map = params def _time_features( self, t: torch.Tensor, height: int, width: int, batch_size: int ) -> tuple[torch.Tensor, ...]: t_emb = ((t * 100) % 24).view(1, 1, 1, 1).expand(batch_size, 1, height, width) sin_t = torch.sin(torch.pi * t_emb / 12 - torch.pi / 2) cos_t = torch.cos(torch.pi * t_emb / 12 - torch.pi / 2) sin_season = torch.sin(torch.pi * t_emb / (12 * 365) - torch.pi / 2) cos_season = torch.cos(torch.pi * t_emb / (12 * 365) - torch.pi / 2) return t_emb, torch.cat([sin_t, cos_t], dim=1), torch.cat( [sin_season, cos_season], dim=1 ) def pde(self, t: torch.Tensor, state: torch.Tensor) -> torch.Tensor: height, width = state.shape[-2:] ds = state[:, -self.out_ch :].view(-1, self.out_ch, height, width).float() velocity = state[:, : 2 * self.out_ch].view( -1, 2 * self.out_ch, height, width ).float() t_emb, day_emb, season_emb = self._time_features( t, height, width, ds.shape[0] ) ds_grad_x = torch.gradient(ds, dim=3)[0] ds_grad_y = torch.gradient(ds, dim=2)[0] nabla_u = torch.cat([ds_grad_x, ds_grad_y], dim=1) if self.pos: combined = torch.cat( [t_emb / 24, day_emb, season_emb, nabla_u, velocity, ds, self.pos_feat], dim=1, ) else: cos_lat, sin_lat = torch.cos(self.new_lat_map), torch.sin(self.new_lat_map) cos_lon, sin_lon = torch.cos(self.new_lon_map), torch.sin(self.new_lon_map) time_cyclic = torch.cat([day_emb, season_emb], dim=1) pos_feats = torch.cat( [ cos_lat, cos_lon, sin_lat, sin_lon, sin_lat * cos_lon, sin_lat * sin_lon, ], dim=1, ) pos_time = self.get_time_pos_embedding(time_cyclic, pos_feats) combined = torch.cat( [ t_emb / 24, day_emb, season_emb, nabla_u, velocity, ds, self.new_lat_map, self.new_lon_map, self.lsm, self.oro, pos_feats, pos_time, ], dim=1, ) dv = self.vel_f(combined) if self.att: dv = dv + self.gamma * self.vel_att(combined) v_x = velocity[:, : self.out_ch] v_y = velocity[:, self.out_ch :] advection = v_x * ds_grad_x + v_y * ds_grad_y advection = advection + ds * ( torch.gradient(v_x, dim=3)[0] + torch.gradient(v_y, dim=2)[0] ) return torch.cat([dv, advection], dim=1) @staticmethod def get_time_pos_embedding( time_features: torch.Tensor, position_features: torch.Tensor ) -> torch.Tensor: outputs = [feature.unsqueeze(1) * position_features for feature in time_features.unbind(1)] return torch.cat(outputs, dim=1) def noise_net_contrib( self, time: torch.Tensor, pos_enc: torch.Tensor, s_final: torch.Tensor, height: int, width: int, ) -> tuple[torch.Tensor, torch.Tensor]: t_emb = (time % 24).view(-1, 1, 1, 1, 1) sin_t = torch.sin(torch.pi * t_emb / 12 - torch.pi / 2).expand( len(s_final), s_final.shape[1], 1, height, width ) cos_t = torch.cos(torch.pi * t_emb / 12 - torch.pi / 2).expand( len(s_final), s_final.shape[1], 1, height, width ) sin_season = torch.sin(torch.pi * t_emb / (12 * 365) - torch.pi / 2).expand( len(s_final), s_final.shape[1], 1, height, width ) cos_season = torch.cos(torch.pi * t_emb / (12 * 365) - torch.pi / 2).expand( len(s_final), s_final.shape[1], 1, height, width ) pos_rep = pos_enc.expand(len(s_final), s_final.shape[1], -1, height, width) pos_rep = pos_rep.flatten(start_dim=0, end_dim=1) time_cyclic = torch.cat([sin_t, cos_t, sin_season, cos_season], dim=2) time_cyclic = time_cyclic.flatten(start_dim=0, end_dim=1) pos_time = self.get_time_pos_embedding(time_cyclic, pos_rep[:, 2:-2]) combined = torch.cat( [time_cyclic, s_final.flatten(start_dim=0, end_dim=1), pos_rep, pos_time], dim=1, ) final_out = self.noise_net(combined).view( len(time), -1, 2 * self.out_ch, height, width ) mean = s_final + final_out[:, :, : self.out_ch] std = F.softplus(final_out[:, :, self.out_ch :]) return mean, std def forward( self, time_steps: torch.Tensor, data: torch.Tensor, atol: float = 0.1, rtol: float = 0.1, ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]: if not isinstance(self.past_samples, torch.Tensor): raise RuntimeError("Call update_param before forward") height, width = self.past_samples.shape[-2:] values = data.float().view(-1, self.out_ch, height, width) final_data = torch.cat([self.past_samples, values], dim=1) init_time = time_steps[0].item() * 6 final_time = time_steps[-1].item() * 6 steps_val = final_time - init_time if self.pos: lat_map = self.lat_map.unsqueeze(0) * torch.pi / 180 lon_map = self.lon_map.unsqueeze(0) * torch.pi / 180 pos_rep = torch.cat([lat_map.unsqueeze(0), lon_map.unsqueeze(0), self.const_info], dim=1) self.pos_feat = self.pos_enc(pos_rep).expand( values.shape[0], -1, values.shape[-2], values.shape[-1] ) final_pos_enc = self.pos_feat else: self.oro = self.const_info[0, 0] self.lsm = self.const_info[0, 1] self.lsm = self.lsm.unsqueeze(0).expand(values.shape[0], -1, height, width) self.oro = F.normalize(self.oro).unsqueeze(0).expand(values.shape[0], -1, height, width) self.new_lat_map = self.lat_map.expand(values.shape[0], 1, height, width) * torch.pi / 180 self.new_lon_map = self.lon_map.expand(values.shape[0], 1, height, width) * torch.pi / 180 cos_lat, sin_lat = torch.cos(self.new_lat_map), torch.sin(self.new_lat_map) cos_lon, sin_lon = torch.cos(self.new_lon_map), torch.sin(self.new_lon_map) pos_feats = torch.cat( [cos_lat, cos_lon, sin_lat, sin_lon, sin_lat * cos_lon, sin_lat * sin_lon], dim=1, ) final_pos_enc = torch.cat( [self.new_lat_map, self.new_lon_map, pos_feats, self.lsm, self.oro], dim=1 ) integration_steps = max(int(steps_val) + 1, 1) new_time_steps = torch.linspace( init_time, final_time, steps=integration_steps, device=values.device ) ode_time = 0.01 * new_time_steps.float() final_result = odeint( self.pde, final_data, ode_time, method=self.method, atol=atol, rtol=rtol, ) s_final = final_result[:, :, -self.out_ch :].view( len(ode_time), -1, self.out_ch, height, width ) sampled = s_final[0 : len(s_final) : 6] if self.err: mean, std = self.noise_net_contrib( time_steps, final_pos_enc, sampled, height, width ) return mean, std, sampled return sampled, torch.zeros_like(sampled), sampled class ClimODE(ClimateEncoderFreeUncertain): """Configuration-friendly public model name.""" def __init__( self, num_channels: int = 5, const_channels: int = 2, out_types: int = 5, method: str = "euler", use_attention: bool = True, use_uncertainty: bool = True, use_positional_encoder: bool = False, ): super().__init__( num_channels=num_channels, const_channels=const_channels, out_types=out_types, method=method, use_att=use_attention, use_err=use_uncertainty, use_pos=use_positional_encoder, ) def _register_checkpoint_compat_modules() -> None: """Expose legacy top-level module names used by official pickle files.""" for legacy_name, current_name in ( ("model_function", "model.model_function"), ("model_utils", "model.model_utils"), ): module = importlib.import_module(current_name) sys.modules.setdefault(legacy_name, module) def load_checkpoint(path: str | Path, map_location: str | torch.device = "cpu") -> nn.Module: """Load an official full-object checkpoint or a state-dict checkpoint.""" checkpoint_path = Path(path) if not checkpoint_path.is_file(): raise FileNotFoundError(checkpoint_path) _register_checkpoint_compat_modules() try: checkpoint = torch.load(checkpoint_path, map_location=map_location, weights_only=False) except TypeError: # Older PyTorch does not expose weights_only. checkpoint = torch.load(checkpoint_path, map_location=map_location) if isinstance(checkpoint, nn.Module): return checkpoint model = ClimODE() if isinstance(checkpoint, dict): state_dict = checkpoint.get("state_dict", checkpoint.get("model", checkpoint)) else: state_dict = checkpoint model.load_state_dict(state_dict) return model # Names kept for the official checkpoint's pickle module/class references. Climate_encoder_free_uncertain = ClimateEncoderFreeUncertain Climate_ResNet_2D = ClimateResNet2D Self_attn_conv = SelfAttnConv boundarypad = BoundaryPad