from __future__ import annotations import argparse import os import sys from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import torch import yaml PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) from model.gpinn import ( # noqa: E402 exact_poisson1d, exact_poisson2d, gPINN, gPINNBurgers, gPINNPoisson2D, ) DEFAULT_CONFIG = PROJECT_ROOT / "conf" / "config.yaml" CASES = ("1d", "2d", "burgers") def load_config(path: Path) -> dict: with path.open("r", encoding="utf-8") as stream: config = yaml.safe_load(stream) if not isinstance(config, dict) or "root" not in config: raise ValueError(f"config must contain a 'root' mapping: {path}") return config["root"] def project_path(value: str | Path) -> Path: path = Path(value).expanduser() return path if path.is_absolute() else PROJECT_ROOT / path def resolve_device(requested: str) -> torch.device: if requested == "auto": return torch.device("cuda" if torch.cuda.is_available() else "cpu") device = torch.device(requested) if device.type == "cuda" and not torch.cuda.is_available(): raise RuntimeError("CUDA/DCU was requested but torch.cuda.is_available() is false") return device def resolve_dtype(name: str) -> torch.dtype: try: return {"float32": torch.float32, "float64": torch.float64}[name] except KeyError as error: raise ValueError(f"unsupported dtype: {name}") from error def load_checkpoint(weight_dir: Path, filename: str) -> dict: checkpoint_path = weight_dir / filename if not checkpoint_path.is_file(): raise FileNotFoundError( f"checkpoint not found: {checkpoint_path}. Run scripts/train.py first." ) checkpoint = torch.load(checkpoint_path, map_location="cpu", weights_only=True) if not isinstance(checkpoint, dict) or "model_state" not in checkpoint: raise ValueError(f"invalid checkpoint format: {checkpoint_path}") return checkpoint def infer_poisson1d( weight_dir: Path, result_dir: Path, device: torch.device, dtype: torch.dtype, ) -> float: checkpoint = load_checkpoint(weight_dir, "gpinn_poisson1d.pt") layers = checkpoint.get("layers", [1, 30, 30, 30, 30, 1]) model = gPINN(layers).to(device=device, dtype=dtype) model.load_state_dict(checkpoint["model_state"], strict=True) model.eval() x = torch.linspace(0.0, np.pi, 1000, dtype=dtype, device=device).unsqueeze(-1) exact = exact_poisson1d(x.cpu().numpy().reshape(-1)) with torch.no_grad(): prediction = model(x).cpu().numpy().reshape(-1) relative_l2 = float(np.linalg.norm(prediction - exact) / np.linalg.norm(exact)) figure, axis = plt.subplots(figsize=(8, 4)) axis.plot(x.cpu().numpy(), exact, "k-", label="Exact") axis.plot(x.cpu().numpy(), prediction, "r--", label="gPINN") axis.set_xlabel("x") axis.set_ylabel("u") axis.legend() figure.tight_layout() output_path = result_dir / "gpinn_poisson1d.png" figure.savefig(output_path, dpi=150) plt.close(figure) print(f"[1D] relative L2={relative_l2:.6e}, plot={output_path}") return relative_l2 def infer_poisson2d( weight_dir: Path, result_dir: Path, device: torch.device, dtype: torch.dtype, default_exponent: float, ) -> float: checkpoint = load_checkpoint(weight_dir, "gpinn_poisson2d.pt") layers = checkpoint.get("layers", [2, 30, 30, 30, 1]) architecture = checkpoint.get("architecture", "gpinn") if architecture == "poisson2d_hard_bc": model = gPINNPoisson2D(layers) elif architecture == "gpinn": # The bundled legacy checkpoint predates the hard boundary transform. model = gPINN(layers) else: raise ValueError(f"unsupported 2D Poisson architecture: {architecture}") model = model.to(device=device, dtype=dtype) model.load_state_dict(checkpoint["model_state"], strict=True) model.eval() exponent = float(checkpoint.get("a", default_exponent)) axis = np.linspace(0.0, 1.0, 100) grid_x, grid_y = np.meshgrid(axis, axis) coordinates = torch.as_tensor( np.column_stack([grid_x.ravel(), grid_y.ravel()]), dtype=dtype, device=device ) with torch.no_grad(): prediction = model(coordinates).cpu().numpy().reshape(grid_x.shape) exact = exact_poisson2d(grid_x, grid_y, exponent) absolute_error = np.abs(prediction - exact) relative_l2 = float( np.linalg.norm(prediction.ravel() - exact.ravel()) / np.linalg.norm(exact.ravel()) ) figure, axes = plt.subplots(1, 3, figsize=(15, 4)) for plot_axis, title, field in zip( axes, ("Exact", "gPINN", "Absolute error"), (exact, prediction, absolute_error), strict=True, ): image = plot_axis.imshow(field, extent=(0, 1, 0, 1), origin="lower", cmap="jet") plot_axis.set_title(title) plot_axis.set_xlabel("x") plot_axis.set_ylabel("y") figure.colorbar(image, ax=plot_axis) figure.tight_layout() output_path = result_dir / "gpinn_poisson2d.png" figure.savefig(output_path, dpi=150) plt.close(figure) print(f"[2D] relative L2={relative_l2:.6e}, plot={output_path}") return relative_l2 def load_burgers_grid(data_path: Path) -> tuple[np.ndarray, np.ndarray, np.ndarray]: if not data_path.is_file(): raise FileNotFoundError(f"Burgers data not found: {data_path}") with np.load(data_path) as data: missing = {"t", "x", "usol"}.difference(data.files) if missing: raise ValueError(f"Burgers data is missing arrays: {sorted(missing)}") time_axis = np.asarray(data["t"]).reshape(-1) space_axis = np.asarray(data["x"]).reshape(-1) exact = np.asarray(data["usol"]) expected_shape = (space_axis.size, time_axis.size) if exact.shape != expected_shape: raise ValueError(f"usol shape must be {expected_shape}, got {exact.shape}") return time_axis, space_axis, exact def infer_burgers( weight_dir: Path, result_dir: Path, data_path: Path, device: torch.device, dtype: torch.dtype, ) -> float: checkpoint = load_checkpoint(weight_dir, "gpinn_burgers.pt") layers = checkpoint.get("layers", [2, 32, 32, 32, 1]) model = gPINNBurgers(layers).to(device=device, dtype=dtype) model.load_state_dict(checkpoint["model_state"], strict=True) model.eval() time_axis, space_axis, exact = load_burgers_grid(data_path) time_grid, space_grid = np.meshgrid(time_axis, space_axis) coordinates = torch.as_tensor( np.column_stack([space_grid.ravel(), time_grid.ravel()]), dtype=dtype, device=device, ) with torch.no_grad(): prediction = model(coordinates).cpu().numpy().reshape(exact.shape) absolute_error = np.abs(prediction - exact) relative_l2 = float( np.linalg.norm(prediction.ravel() - exact.ravel()) / np.linalg.norm(exact.ravel()) ) figure, axes = plt.subplots(1, 3, figsize=(15, 4)) for plot_axis, title, field in zip( axes, ("Exact", "gPINN", "Absolute error"), (exact, prediction, absolute_error), strict=True, ): image = plot_axis.contourf(time_axis, space_axis, field, 100, cmap="jet") plot_axis.set_title(title) plot_axis.set_xlabel("t") plot_axis.set_ylabel("x") figure.colorbar(image, ax=plot_axis) figure.tight_layout() output_path = result_dir / "gpinn_burgers.png" figure.savefig(output_path, dpi=150) plt.close(figure) print(f"[Burgers] relative L2={relative_l2:.6e}, plot={output_path}") return relative_l2 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run gPINN inference and visualization") parser.add_argument( "--config", type=Path, default=Path(os.environ.get("GPINN_CONFIG", DEFAULT_CONFIG)), help="YAML configuration file", ) parser.add_argument("--case", choices=(*CASES, "all"), default="all") parser.add_argument("--device", help="Override common.device, for example cpu or cuda:0") parser.add_argument("--data", type=Path, help="Override Burgers.npz path") parser.add_argument("--weight-dir", type=Path, help="Override checkpoint input directory") parser.add_argument("--result-dir", type=Path, help="Override plot output directory") return parser.parse_args() def main() -> None: args = parse_args() config_path = args.config.expanduser().resolve() config = load_config(config_path) common = config["common"] device = resolve_device(args.device or str(common["device"])) dtype = resolve_dtype(str(common["dtype"])) weight_dir = project_path(args.weight_dir or common["weight_dir"]) result_dir = project_path(args.result_dir or common["result_dir"]) data_path = project_path(args.data or config["burgers"]["data"]) result_dir.mkdir(parents=True, exist_ok=True) print(f"Config: {config_path}") print(f"Device: {device}") selected_cases = CASES if args.case == "all" else (args.case,) if "1d" in selected_cases: infer_poisson1d(weight_dir, result_dir, device, dtype) if "2d" in selected_cases: infer_poisson2d( weight_dir, result_dir, device, dtype, float(config["poisson2d"]["a"]), ) if "burgers" in selected_cases: infer_burgers(weight_dir, result_dir, data_path, device, dtype) if __name__ == "__main__": main()