model/graphdop.py

# coding=utf-8
#
# SPDX-License-Identifier: Apache-2.0
#
# Minimal reproduction of GraphDOP (ECMWF, "Towards skilful medium-range
# forecasts learnt directly from observations", 2025 preprint) following the
# encoder -- processor -- decoder design:
#
#   * Encoder: a GNN that projects gridded "observations" inside the input
#     window onto a latent mesh (a coarse regular lat/lon grid), using graph
#     edges with (forward bearing, haversine distance) features.
#   * Processor: a transformer that advances the latent atmospheric state
#     forward in time, once per output frame (latent-space rollout).
#   * Decoder: a GNN that maps the latent mesh back onto the target grid and
#     predicts per-channel observations with instrument-like output MLPs.
#
# Differences from the paper (documented in README.md): the paper consumes
# irregular, instrument-specific Level-1 observations with dynamic graphs built
# per batch (PyTorch Geometric); here the OneScience ERA5-h5 gridded pipeline is
# used as the observation placeholder, and the graphs are fixed regular-grid
# meshes. The weighted MSE objective is kept (per-channel weights).
import math

import torch
import torch.nn as nn
import torch.nn.functional as F


def _latlon_grid(shape):
    """Regular lat/lon coordinates for a (H, W) grid, North-to-South rows."""
    H, W = shape
    lat = torch.linspace(90.0, -90.0, H)
    lon = torch.linspace(0.0, 360.0 - 360.0 / W, W)
    return lat, lon


def _haversine(lat1, lon1, lat2, lon2):
    """Haversine distance in metres between points given in degrees."""
    R = 6371000.0
    p1 = torch.deg2rad(lat1)
    p2 = torch.deg2rad(lat2)
    dp = torch.deg2rad(lat2 - lat1)
    dl = torch.deg2rad(lon2 - lon1)
    a = torch.sin(dp / 2) ** 2 + torch.cos(p1) * torch.cos(p2) * torch.sin(dl / 2) ** 2
    return 2 * R * torch.asin(torch.sqrt(a.clamp(0, 1)))


def _bearing(lat1, lon1, lat2, lon2):
    """Initial forward bearing in radians from point 1 to point 2."""
    p1 = torch.deg2rad(lat1)
    p2 = torch.deg2rad(lat2)
    dl = torch.deg2rad(lon2 - lon1)
    y = torch.sin(dl) * torch.cos(p2)
    x = torch.cos(p1) * torch.sin(p2) - torch.sin(p1) * torch.cos(p2) * torch.cos(dl)
    return torch.atan2(y, x)


def build_mesh_graph(mesh_shape):
    """
    Build a fixed 8-neighbourhood graph over a regular latent mesh.
    Longitude wraps around; edge features are (forward bearing [rad],
    haversine distance [km]).
    """
    H, W = mesh_shape
    lat, lon = _latlon_grid(mesh_shape)
    lat = lat.view(-1, 1).expand(H, W)
    lon = lon.view(1, -1).expand(H, W)

    src_list, dst_list, feat_list = [], [], []
    for i in range(H):
        for j in range(W):
            for di, dj in ((-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 1), (1, -1), (1, 0), (1, 1)):
                ni, nj = i + di, (j + dj) % W
                if not (0 <= ni < H):
                    continue
                s = i * W + j
                d = ni * W + nj
                dist_km = _haversine(lat[i, j], lon[i, j], lat[ni, nj], lon[ni, nj]) / 1000.0
                bear = _bearing(lat[i, j], lon[i, j], lat[ni, nj], lon[ni, nj])
                src_list.append(s)
                dst_list.append(d)
                feat_list.append(torch.stack([bear / math.pi, dist_km / 1000.0]))
    edge_index = torch.stack([torch.as_tensor(src_list), torch.as_tensor(dst_list)], dim=0)
    edge_attr = torch.stack(feat_list)
    return edge_index, edge_attr


def _mlp(in_dim, out_dim, hidden_dim, n_layers=2):
    dims = [in_dim] + [hidden_dim] * (n_layers - 1) + [out_dim]
    layers = []
    for i in range(len(dims) - 1):
        layers.append(nn.Linear(dims[i], dims[i + 1]))
        if i < len(dims) - 2:
            layers.append(nn.GELU())
    return nn.Sequential(*layers)


class GNNLayer(nn.Module):
    """Message-passing layer with edge features (mean-aggregate, residual)."""

    def __init__(self, dim, edge_dim=2, hidden_dim=64):
        super().__init__()
        self.edge_mlp = _mlp(2 * dim + edge_dim, dim, hidden_dim)
        self.node_mlp = _mlp(dim, dim, hidden_dim)
        self.norm = nn.LayerNorm(dim)

    def forward(self, x, edge_index, edge_attr):
        B, N, D = x.shape
        src, dst = edge_index
        offsets = torch.arange(B, device=x.device) * N
        src_b = (src.unsqueeze(0) + offsets.view(B, 1)).reshape(-1)
        dst_b = (dst.unsqueeze(0) + offsets.view(B, 1)).reshape(-1)
        edge_attr_b = edge_attr.unsqueeze(0).expand(B, -1, -1).reshape(-1, edge_attr.size(1))
        xb = x.reshape(B * N, D)
        msg = self.edge_mlp(torch.cat([xb[src_b], xb[dst_b], edge_attr_b], dim=1))
        agg = torch.zeros_like(xb)
        agg.index_add_(0, dst_b, msg)
        cnt = torch.bincount(dst_b, minlength=B * N).clamp(min=1).unsqueeze(1)
        agg = agg / cnt
        agg = agg.reshape(B, N, D)
        return self.norm(x + self.node_mlp(agg))


class ObsEncoder(nn.Module):
    """
    Maps the observation grid onto the latent mesh with a per-cell input MLP,
    an adaptive pooling to the mesh resolution, and graph message passing.
    """

    def __init__(self, in_channels, latent_dim, mesh_shape, num_layers=2, hidden_dim=64):
        super().__init__()
        self.in_channels = in_channels
        self.input_mlp = _mlp(in_channels, latent_dim, hidden_dim)
        self.gnn = nn.ModuleList([GNNLayer(latent_dim, hidden_dim=hidden_dim) for _ in range(num_layers)])
        self.mesh_shape = mesh_shape
        self.edge_index, self.edge_attr = build_mesh_graph(mesh_shape)

    def forward(self, x):
        B, C, H, W = x.shape
        feat = x.permute(0, 2, 3, 1).reshape(-1, C)
        feat = self.input_mlp(feat).reshape(B, H, W, -1).permute(0, 3, 1, 2)
        mesh = F.adaptive_avg_pool2d(feat, self.mesh_shape)
        mesh = mesh.permute(0, 2, 3, 1).reshape(B, -1, mesh.size(1))
        edge_index, edge_attr = self.edge_index.to(x.device), self.edge_attr.to(x.device)
        for layer in self.gnn:
            mesh = layer(mesh, edge_index, edge_attr)
        return mesh


class LatentProcessor(nn.Module):
    """Transformer over latent mesh tokens that advances the state in time."""

    def __init__(self, latent_dim, mesh_shape, num_blocks=1, n_heads=4, hidden_dim=128):
        super().__init__()
        n_nodes = mesh_shape[0] * mesh_shape[1]
        self.pos_emb = nn.Parameter(torch.zeros(1, n_nodes, latent_dim))
        nn.init.trunc_normal_(self.pos_emb, std=0.02)
        block = nn.TransformerEncoderLayer(
            d_model=latent_dim, nhead=n_heads, dim_feedforward=hidden_dim,
            dropout=0.0, activation="gelu", batch_first=True, norm_first=True,
        )
        self.blocks = nn.ModuleList([block for _ in range(num_blocks)])

    def forward(self, mesh):
        tokens = mesh + self.pos_emb
        for block in self.blocks:
            tokens = block(tokens)
        return tokens


class ObsDecoder(nn.Module):
    """
    Maps the latent mesh back onto the target grid (bilinear upsample) and
    predicts per-channel observations with an output MLP.
    """

    def __init__(self, latent_dim, out_channels, grid_shape, mesh_shape, num_layers=2, hidden_dim=64):
        super().__init__()
        self.gnn = nn.ModuleList([GNNLayer(latent_dim, hidden_dim=hidden_dim) for _ in range(num_layers)])
        self.grid_shape = grid_shape
        self.mesh_shape = mesh_shape
        self.edge_index, self.edge_attr = build_mesh_graph(mesh_shape)
        self.output_mlp = _mlp(latent_dim, out_channels, hidden_dim)

    def forward(self, mesh):
        B, N, D = mesh.shape
        edge_index = self.edge_index.to(mesh.device)
        edge_attr = self.edge_attr.to(mesh.device)
        for layer in self.gnn:
            mesh = layer(mesh, edge_index, edge_attr)
        H, W = self.grid_shape
        Hm, Wm = self.mesh_shape
        mesh = mesh.transpose(1, 2).reshape(B, D, Hm, Wm)
        grid = F.interpolate(mesh, size=self.grid_shape, mode="bilinear", align_corners=False)
        grid = grid.permute(0, 2, 3, 1).reshape(B, H * W, D)
        return self.output_mlp(grid).reshape(B, H, W, -1).permute(0, 3, 1, 2)


class GraphDOP(nn.Module):
    """
    Config-driven GraphDOP wrapper.

    Args:
        in_channels: Number of observation channels per frame.
        out_channels: Number of forecast channels per frame.
        input_steps: Number of input (observation window) frames.
        output_steps: Number of forecast frames.
        grid_shape: Spatial shape of the (gridded) observation field.
        mesh_shape: Latent mesh resolution (each dimension, powers of two fine).
        latent_dim: Feature dimension of latent mesh tokens.
        num_encoder_layers / num_decoder_layers: GNN message-passing layers.
        num_processor_blocks: Transformer blocks in the processor.
        n_heads: Attention heads of the processor.
        channel_weights: Per-channel weights for the weighted MSE objective.
    """

    def __init__(
        self,
        in_channels=6,
        out_channels=6,
        input_steps=2,
        output_steps=2,
        grid_shape=(32, 32),
        mesh_shape=(8, 8),
        latent_dim=64,
        num_encoder_layers=2,
        num_decoder_layers=2,
        num_processor_blocks=1,
        n_heads=4,
        hidden_dim=64,
        channel_weights=None,
    ):
        super().__init__()
        self.in_channels = int(in_channels)
        self.out_channels = int(out_channels)
        self.input_steps = int(input_steps)
        self.output_steps = int(output_steps)
        self.grid_shape = (int(grid_shape[0]), int(grid_shape[1]))
        self.mesh_shape = (int(mesh_shape[0]), int(mesh_shape[1]))

        self.encoder = ObsEncoder(
            self.in_channels, int(latent_dim), self.mesh_shape, num_layers=int(num_encoder_layers), hidden_dim=int(hidden_dim)
        )
        self.processor = LatentProcessor(
            int(latent_dim), self.mesh_shape, num_blocks=int(num_processor_blocks), n_heads=int(n_heads), hidden_dim=int(hidden_dim)
        )
        self.decoder = ObsDecoder(
            int(latent_dim), self.out_channels, self.grid_shape, self.mesh_shape,
            num_layers=int(num_decoder_layers), hidden_dim=int(hidden_dim),
        )

        if channel_weights is None:
            channel_weights = torch.ones(self.out_channels)
        self.register_buffer("channel_weights", torch.as_tensor(channel_weights, dtype=torch.float32))

    def forward(self, x):
        """
        Args:
            x: Observation frames, shape [batch, input_steps, C, H, W].
        Returns:
            Forecast frames, shape [batch, output_steps, C, H, W].
        """
        latents = torch.stack([self.encoder(x[:, t]) for t in range(self.input_steps)], dim=0)
        latent = latents.mean(dim=0)
        outs = []
        for _ in range(self.output_steps):
            latent = self.processor(latent)
            outs.append(self.decoder(latent))
        return torch.stack(outs, dim=1)

    def wmse_loss(self, pred, target):
        """Weighted mean squared error objective (Eq. 1 of the paper)."""
        diff = (pred - target) ** 2
        w = self.channel_weights.view(1, 1, self.out_channels, 1, 1)
        return (diff * w).mean()