scripts/common.py

"""Configuration, exact MAT contract, sampling and checkpoint utilities."""
from __future__ import annotations

import hashlib
import json
import random
from pathlib import Path

import numpy as np
import torch
import yaml
from scipy.io import loadmat

ROOT = Path(__file__).resolve().parents[1]


def read_config(path=None):
    path = Path(path) if path else ROOT / "config/config.yaml"
    with path.open() as handle:
        cfg = yaml.safe_load(handle)
    if cfg["model"]["hidden_layers"] != [50] * 4:
        raise ValueError("Formal experiment requires exactly four 50-unit hidden layers")
    if cfg["training"]["iterations"] != 4000:
        raise ValueError("Formal experiment requires exactly 4000 iterations")
    return cfg


def resolve(path):
    p = Path(path)
    return p if p.is_absolute() else ROOT / p


def save_json(path, value):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False))
    tmp.replace(path)


def setup(cfg, device=None):
    seed = int(cfg["seed"])
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.set_num_threads(int(cfg["runtime"]["cpu_threads"]))
    dtype = getattr(torch, cfg["model"]["dtype"])
    torch.set_default_dtype(dtype)
    device = torch.device(device or cfg["runtime"]["device"])
    if device.type == "cuda" and not torch.cuda.is_available():
        raise RuntimeError("Requested accelerator is unavailable; run inside SLURM allocation")
    return device, dtype


def build_model(cfg, device):
    from model.TSAPINN import TSAPINN
    m = cfg["model"]
    return TSAPINN(**{k: m[k] for k in ["hidden_layers", "sigma", "zeta_sin", "zeta_cos", "reynolds", "bias_initialization"]}).to(device)


def load_dns(cfg):
    path = Path(cfg["data"]["path"])
    if not path.is_file():
        raise FileNotFoundError(f"Required author DNS data absent: {path}")
    raw = loadmat(path)
    for key, shape in cfg["data"]["expected_shapes"].items():
        if raw[key].shape != tuple(shape) or not np.isfinite(raw[key]).all():
            raise ValueError(f"Invalid DNS field {key}: expected finite {shape}")
    xy = np.asarray(raw["X_star"], dtype=np.float64)
    times = np.asarray(raw["t"], dtype=np.float64).ravel()
    if not np.all(np.diff(times) > 0):
        raise ValueError("DNS time must be strictly increasing")
    return {"xy": xy, "t": times, "uv": np.asarray(raw["U_star"]).transpose(2, 0, 1),
            "p": np.asarray(raw["p_star"]).T, "path": str(path),
            "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}


def sample_indices(cfg, dns):
    """Literal paper per-time counts, geometric boundary; fixed throughout training.

    Only 296 unique boundary nodes exist. Balanced sampling with replacement
    provides 100 direct DNS observations per side without inventing new labels.
    Boundary corner duplication and replacement are explicit implementation choices.
    """
    rng = np.random.RandomState(cfg["seed"])
    xy, times = dns["xy"], dns["t"]
    xmin, ymin = xy.min(0)
    xmax, ymax = xy.max(0)
    sides = [np.flatnonzero(np.isclose(xy[:, 0], xmin)), np.flatnonzero(np.isclose(xy[:, 0], xmax)),
             np.flatnonzero(np.isclose(xy[:, 1], ymin)), np.flatnonzero(np.isclose(xy[:, 1], ymax))]
    boundary = np.unique(np.concatenate(sides))
    interior = np.setdiff1d(np.arange(len(xy)), boundary)
    nt, tr = len(times), cfg["training"]
    nb, nr = int(tr["boundary_points_per_time"]), int(tr["residual_points_per_time"])
    if nb % 4:
        raise ValueError("Balanced side sampling needs boundary count divisible by 4")
    bc = np.stack([np.concatenate([rng.choice(side, nb // 4, replace=True) for side in sides]) for _ in times])
    res = np.stack([rng.choice(interior, nr, replace=False) for _ in times])
    val = np.stack([rng.choice(np.setdiff1d(interior, res[k]), tr["validation_points_per_time"], replace=False) for k in range(nt)])
    return {"boundary": bc, "residual": res, "validation": val}


def gather(dns, indices):
    nt, n = indices.shape
    xy = dns["xy"][indices]
    time = np.broadcast_to(dns["t"][:, None, None], (nt, n, 1))
    coordinates = np.concatenate((xy, time), 2).reshape(-1, 3)
    labels = dns["uv"][np.arange(nt)[:, None], indices].reshape(-1, 2)
    return coordinates, labels


def tensor(array, device, dtype):
    return torch.as_tensor(np.ascontiguousarray(array), dtype=dtype, device=device)


def predict(model, coordinates, device, dtype, batch_size):
    values = []
    model.eval()
    with torch.no_grad():
        for start in range(0, len(coordinates), batch_size):
            q = tensor(coordinates[start:start + batch_size], device, dtype)
            values.append(model.flow_taylor(q).cpu().numpy())
    return np.concatenate(values)


def relative_l2(prediction, reference):
    difference = np.asarray(prediction, dtype=np.float64) - reference
    numerator = np.sum(difference * difference, axis=0, dtype=np.float64)
    denominator = np.sum(np.asarray(reference, dtype=np.float64)**2, axis=0, dtype=np.float64)
    return np.sqrt(numerator / denominator)


def atomic_checkpoint(path, payload):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".tmp")
    torch.save(payload, tmp)
    tmp.replace(path)


def load_checkpoint(cfg, device, path=None):
    path = resolve(path or cfg["outputs"]["checkpoint"])
    checkpoint = torch.load(path, map_location=device, weights_only=False)
    saved_config = checkpoint["config"].copy()
    saved_config["data"] = saved_config["data"].copy()
    saved_config["data"]["path"] = cfg["data"]["path"]
    if saved_config != cfg:
        raise ValueError("Checkpoint configuration does not match requested configuration")
    model = build_model(cfg, device)
    model.load_state_dict(checkpoint["model_state_dict"], strict=True)
    model.eval()
    return model, checkpoint