meshgraphnet.py

"""MeshGraphNet (Pfaff et al., 2021) for cylinder_flow, laid out like a transformers LLM.

    MeshGraphNetForSimulation      backbone + decoder head + loss + rollout  (~ LlamaForCausalLM)
      model: MeshGraphNetModel     backbone                                   (~ LlamaModel)
        embeddings                 normalizers + node/edge encoder MLPs       (~ embed_tokens)
        layers[i]                  message passing: edge_block + node_block   (~ LlamaDecoderLayer)
"""

from dataclasses import dataclass
from typing import Optional

import torch
from torch import nn
from transformers import PretrainedConfig, PreTrainedModel
from transformers.utils import ModelOutput


class MeshGraphNetConfig(PretrainedConfig):
    r"""Configuration for MeshGraphNet (Pfaff et al., 2021) on cylinder_flow.

    Defaults reproduce the DeepMind / medav cylinder_flow model.

    Args:
        hidden_size: Latent width of every node/edge representation.
        num_hidden_layers: Number of message-passing layers (processor steps).
        num_mlp_hidden_layers: Hidden layers inside every MLP (output layer excluded).
        velocity_size: Velocity components per node.
        num_node_types: Size of the node-type one-hot vector.
        edge_input_size: Raw edge features (relative mesh position + its norm).
        output_size: Predicted quantities per node (velocity change).
        split_first_linear: Hold the first Linear of each layer's edge/node MLP
            as one Linear per input (``W [h_s; h_r; e] = W_s h_s + W_r h_r + W_e e``).
            Sender/receiver projections are then computed once per node and
            gathered, instead of building the per-edge concatenation.
            Weights and outputs are identical to ``False``.
        layer_norm_eps: LayerNorm epsilon.
        normalizer_eps: Lower bound on the feature-normalizer std.
    """

    model_type = "meshgraphnet"

    def __init__(
        self,
        hidden_size: int = 128,
        num_hidden_layers: int = 15,
        num_mlp_hidden_layers: int = 2,
        velocity_size: int = 2,
        num_node_types: int = 9,
        edge_input_size: int = 3,
        output_size: int = 2,
        split_first_linear: bool = False,
        layer_norm_eps: float = 1e-5,
        normalizer_eps: float = 1e-8,
        **kwargs,
    ):
        self.hidden_size = hidden_size
        self.num_hidden_layers = num_hidden_layers
        self.num_mlp_hidden_layers = num_mlp_hidden_layers
        self.velocity_size = velocity_size
        self.num_node_types = num_node_types
        self.edge_input_size = edge_input_size
        self.output_size = output_size
        self.split_first_linear = split_first_linear
        self.layer_norm_eps = layer_norm_eps
        self.normalizer_eps = normalizer_eps
        super().__init__(**kwargs)

    @property
    def node_input_size(self) -> int:
        return self.velocity_size + self.num_node_types


# Node types of the DeepMind datasets. Only NORMAL and OUTFLOW nodes are predicted.
NORMAL, OBSTACLE, AIRFOIL, HANDLE, INFLOW, OUTFLOW, WALL_BOUNDARY = range(7)


def prediction_mask(node_type: torch.Tensor) -> torch.Tensor:
    return (node_type == NORMAL) | (node_type == OUTFLOW)


def prepare_inputs(cells, mesh_pos, node_type, velocity) -> dict:
    """One mesh time step -> model inputs (the counterpart of a tokenizer).

    cells [M, 3] triangles -> edge_index [2, E] (senders, receivers), both directions.
    """
    cells = torch.as_tensor(cells).long()
    raw = torch.cat([cells[:, 0:2], cells[:, 1:3], cells[:, [2, 0]]])
    edges = torch.stack([raw.max(1).values, raw.min(1).values], 1).unique(dim=0)
    a, b = edges[:, 0], edges[:, 1]
    return {
        "velocity": torch.as_tensor(velocity).float(),
        "node_type": torch.as_tensor(node_type).long().reshape(-1),
        "mesh_pos": torch.as_tensor(mesh_pos).float(),
        "edge_index": torch.stack([torch.cat([a, b]), torch.cat([b, a])]),
    }


class MeshGraphNetNormalizer(nn.Module):
    """Feature normalizer that accumulates mean/std in training mode (DeepMind normalization.py).

    Statistics are accumulated for the first ``max_accumulations`` training calls, then frozen.
    """

    def __init__(self, size: int, eps: float = 1e-8, max_accumulations: int = 10**6):
        super().__init__()
        self.eps, self.max_accumulations = eps, max_accumulations
        self.register_buffer("acc_count", torch.zeros(()))
        self.register_buffer("num_accumulations", torch.zeros(()))
        self.register_buffer("acc_sum", torch.zeros(size))
        self.register_buffer("acc_sum_squared", torch.zeros(size))

    def _stats(self):
        count = self.acc_count.clamp(min=1)
        mean = self.acc_sum / count
        std = (self.acc_sum_squared / count - mean**2).clamp(min=0).sqrt().clamp(min=self.eps)
        return mean, std

    def forward(self, x):
        if self.training:  # no host sync: the update is masked once accumulation is done
            keep = (self.num_accumulations < self.max_accumulations).to(x.dtype)
            self.acc_count += keep * x.shape[0]
            self.num_accumulations += keep
            self.acc_sum += keep * x.detach().sum(0)
            self.acc_sum_squared += keep * (x.detach() ** 2).sum(0)
        mean, std = self._stats()
        return (x - mean) / std

    def inverse(self, x):
        mean, std = self._stats()
        return x * std + mean


class MeshGraphNetMLP(nn.Module):
    """``sizes=[in, h, h, out]`` -> Linear ReLU Linear ReLU Linear (+ LayerNorm)."""

    def __init__(self, sizes: list[int], layer_norm: bool, eps: float = 1e-5):
        super().__init__()
        self.layers = nn.ModuleList(nn.Linear(i, o) for i, o in zip(sizes[:-1], sizes[1:]))
        self.layer_norm = nn.LayerNorm(sizes[-1], eps=eps) if layer_norm else None

    def forward(self, x):
        for i, layer in enumerate(self.layers):
            x = layer(torch.relu(x) if i > 0 else x)
        return self.layer_norm(x) if self.layer_norm is not None else x


class MeshGraphNetEmbeddings(nn.Module):
    def __init__(self, config: MeshGraphNetConfig):
        super().__init__()
        d, hidden, eps = config.hidden_size, [config.hidden_size] * config.num_mlp_hidden_layers, config.layer_norm_eps
        self.num_node_types = config.num_node_types
        self.node_normalizer = MeshGraphNetNormalizer(config.node_input_size, config.normalizer_eps)
        self.edge_normalizer = MeshGraphNetNormalizer(config.edge_input_size, config.normalizer_eps)
        self.node_encoder = MeshGraphNetMLP([config.node_input_size, *hidden, d], True, eps)
        self.edge_encoder = MeshGraphNetMLP([config.edge_input_size, *hidden, d], True, eps)

    def forward(self, velocity, node_type, mesh_pos, edge_index):
        senders, receivers = edge_index
        one_hot = nn.functional.one_hot(node_type, self.num_node_types).to(velocity.dtype)
        rel = mesh_pos[senders] - mesh_pos[receivers]
        node_features = torch.cat([velocity, one_hot], -1)
        edge_features = torch.cat([rel, rel.norm(dim=-1, keepdim=True)], -1)
        return (self.node_encoder(self.node_normalizer(node_features)),
                self.edge_encoder(self.edge_normalizer(edge_features)))


class MeshGraphNetEdgeBlock(nn.Module):
    """New edge message = MLP([h_sender, h_receiver, e]).

    split_first_linear=True holds the first Linear as ``sender``/``receiver``/``edge``:
    sender/receiver projections run once per node and are gathered (no [E, 3d] concat).
    """

    def __init__(self, config: MeshGraphNetConfig):
        super().__init__()
        d = config.hidden_size
        self.split = config.split_first_linear
        if self.split:
            self.sender = nn.Linear(d, d, bias=False)
            self.receiver = nn.Linear(d, d, bias=False)
            self.edge = nn.Linear(d, d)
            for m in (self.sender, self.receiver, self.edge):
                m.concat_in_features = 3 * d  # init as one Linear(3d, d)
        else:
            self.input = nn.Linear(3 * d, d)
        self.mlp = MeshGraphNetMLP([d] * (config.num_mlp_hidden_layers + 1), True, config.layer_norm_eps)

    def forward(self, h, e, edge_index):
        s, r = edge_index
        if self.split:
            x = self.sender(h)[s] + self.receiver(h)[r] + self.edge(e)
        else:
            x = self.input(torch.cat([h[s], h[r], e], -1))
        return self.mlp(torch.relu(x))


class MeshGraphNetNodeBlock(nn.Module):
    """Node update = MLP([h, sum of incoming messages]).

    split_first_linear=True holds the first Linear as ``node``/``aggregated``.
    """

    def __init__(self, config: MeshGraphNetConfig):
        super().__init__()
        d = config.hidden_size
        self.split = config.split_first_linear
        if self.split:
            self.node = nn.Linear(d, d)
            self.aggregated = nn.Linear(d, d, bias=False)
            for m in (self.node, self.aggregated):
                m.concat_in_features = 2 * d  # init as one Linear(2d, d)
        else:
            self.input = nn.Linear(2 * d, d)
        self.mlp = MeshGraphNetMLP([d] * (config.num_mlp_hidden_layers + 1), True, config.layer_norm_eps)

    def forward(self, h, messages, edge_index):
        agg = torch.zeros_like(h).index_add_(0, edge_index[1], messages)
        x = self.node(h) + self.aggregated(agg) if self.split else self.input(torch.cat([h, agg], -1))
        return self.mlp(torch.relu(x))


class MeshGraphNetLayer(nn.Module):
    def __init__(self, config: MeshGraphNetConfig):
        super().__init__()
        self.edge_block = MeshGraphNetEdgeBlock(config)
        self.node_block = MeshGraphNetNodeBlock(config)

    def forward(self, h, e, edge_index):
        messages = self.edge_block(h, e, edge_index)
        return h + self.node_block(h, messages, edge_index), e + messages


@dataclass
class MeshGraphNetOutput(ModelOutput):
    loss: Optional[torch.FloatTensor] = None
    delta_velocity: Optional[torch.FloatTensor] = None  # [N, 2], de-normalized
    hidden_states: Optional[tuple[torch.FloatTensor, ...]] = None  # node states per layer


class MeshGraphNetPreTrainedModel(PreTrainedModel):
    config_class = MeshGraphNetConfig
    base_model_prefix = "model"
    _no_split_modules = ["MeshGraphNetLayer"]

    def _init_weights(self, module):
        # Sonnet (DeepMind) default: truncated normal, std 1/sqrt(fan_in), cut at 2 std; zero bias.
        if isinstance(module, nn.Linear):
            std = getattr(module, "concat_in_features", module.in_features) ** -0.5
            nn.init.trunc_normal_(module.weight, std=std, a=-2 * std, b=2 * std)
            if module.bias is not None:
                nn.init.zeros_(module.bias)
        elif isinstance(module, nn.LayerNorm):
            nn.init.ones_(module.weight)
            nn.init.zeros_(module.bias)


class MeshGraphNetModel(MeshGraphNetPreTrainedModel):
    def __init__(self, config: MeshGraphNetConfig):
        super().__init__(config)
        self.embeddings = MeshGraphNetEmbeddings(config)
        self.layers = nn.ModuleList(MeshGraphNetLayer(config) for _ in range(config.num_hidden_layers))
        self.post_init()

    def forward(self, velocity, node_type, mesh_pos, edge_index, output_hidden_states=False):
        h, e = self.embeddings(velocity, node_type, mesh_pos, edge_index)
        hidden_states = (h,) if output_hidden_states else None
        for layer in self.layers:
            h, e = layer(h, e, edge_index)
            if output_hidden_states:
                hidden_states += (h,)
        return h, hidden_states


class MeshGraphNetForSimulation(MeshGraphNetPreTrainedModel):
    """Predicts the per-node velocity change for one time step (dt = 0.01)."""

    prepare_inputs = staticmethod(prepare_inputs)

    def __init__(self, config: MeshGraphNetConfig):
        super().__init__(config)
        hidden = [config.hidden_size] * config.num_mlp_hidden_layers
        self.model = MeshGraphNetModel(config)
        self.decoder = MeshGraphNetMLP([config.hidden_size, *hidden, config.output_size], False)
        self.output_normalizer = MeshGraphNetNormalizer(config.output_size, config.normalizer_eps)
        self.post_init()

    def forward(self, velocity, node_type, mesh_pos, edge_index, target_velocity=None,
                output_hidden_states=False):
        h, hidden_states = self.model(velocity, node_type, mesh_pos, edge_index, output_hidden_states)
        out = self.decoder(h)
        loss = None
        if target_velocity is not None:
            # DeepMind: MSE of the normalized velocity change on NORMAL/OUTFLOW nodes.
            target = self.output_normalizer(target_velocity - velocity)
            mask = prediction_mask(node_type).to(out.dtype)
            loss = (((target - out) ** 2).sum(-1) * mask).sum() / mask.sum()
        return MeshGraphNetOutput(loss, self.output_normalizer.inverse(out), hidden_states)

    @torch.no_grad()
    def rollout(self, velocity, node_type, mesh_pos, edge_index, num_steps: int):
        """Autoregressive prediction (~ generate). Boundary nodes keep their initial
        velocity. Returns [num_steps + 1, N, 2]."""
        mask = prediction_mask(node_type).unsqueeze(-1)
        trajectory = [velocity]
        for _ in range(num_steps):
            delta = self(velocity, node_type, mesh_pos, edge_index).delta_velocity
            velocity = torch.where(mask, velocity + delta, velocity)
            trajectory.append(velocity)
        return torch.stack(trajectory)