from __future__ import annotations import sys from collections.abc import Mapping from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import torch PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(Path(__file__).resolve().parent)) from common import ( # noqa: E402 build_laplace_data, checkpoint_state, load_config, project_path, relative_l2, resolve_device, resolve_dtype, ) from model.bpinn import build_model, posterior_predict # noqa: E402 DEFAULT_CONFIG = PROJECT_ROOT / "conf" / "config.yaml" def main() -> None: config_path = DEFAULT_CONFIG.resolve() config = load_config(config_path) common = config["common"] device = resolve_device(str(common["device"])) dtype = resolve_dtype(str(common["dtype"])) weight_dir = project_path(common["weight_dir"], PROJECT_ROOT) result_dir = project_path(common["result_dir"], PROJECT_ROOT) refined_path = weight_dir / config["training"]["refined_checkpoint_name"] base_path = weight_dir / config["training"]["checkpoint_name"] checkpoint_path = refined_path if refined_path.is_file() else base_path 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, Mapping): raise ValueError(f"invalid BPINN checkpoint: {checkpoint_path}") state, metadata = checkpoint_state(checkpoint) model_config = metadata.get("model_config", config["model"]) data_config = dict(metadata.get("data_config", config["data"])) seed = int(metadata.get("seed", common["seed"])) data = build_laplace_data(data_config, seed, device, dtype) model = build_model(model_config, dtype=dtype).to(device=device, dtype=dtype) model.load_state_dict(state, strict=True) model.eval() posterior_states = metadata.get("posterior_states") or [state] mean, standard_deviation, samples = posterior_predict( model, posterior_states, data["x_test"] ) mean_numpy = mean.cpu().numpy() std_numpy = standard_deviation.cpu().numpy() exact_numpy = data["u_test"].cpu().numpy() x_numpy = data["x_test"].cpu().numpy() error = relative_l2(mean_numpy, exact_numpy) rmse = float(np.sqrt(np.mean((mean_numpy - exact_numpy) ** 2))) max_error = float(np.max(np.abs(mean_numpy - exact_numpy))) result_dir.mkdir(parents=True, exist_ok=True) predictions_path = result_dir / config["inference"]["predictions_name"] figure_path = result_dir / config["inference"]["figure_name"] np.savez_compressed( predictions_path, x=x_numpy, exact=exact_numpy, mean=mean_numpy, std=std_numpy, samples=samples.cpu().numpy(), relative_l2=error, rmse=rmse, max_abs_error=max_error, ) x_axis = x_numpy.reshape(-1) exact_axis = exact_numpy.reshape(-1) mean_axis = mean_numpy.reshape(-1) std_axis = std_numpy.reshape(-1) absolute_error = np.abs(mean_axis - exact_axis) figure, axes = plt.subplots(1, 2, figsize=(12, 4)) axes[0].plot(x_axis, exact_axis, "k-", linewidth=1.5, label="Exact") axes[0].plot(x_axis, mean_axis, "r--", linewidth=1.5, label="BPINN") if np.any(std_axis > 0): axes[0].fill_between( x_axis, mean_axis - 2.0 * std_axis, mean_axis + 2.0 * std_axis, color="red", alpha=0.2, label="2 std", ) axes[0].set_xlabel("x") axes[0].set_ylabel("u") axes[0].legend() axes[1].semilogy(x_axis, np.maximum(absolute_error, 1.0e-16), "b-") axes[1].set_xlabel("x") axes[1].set_ylabel("Absolute error") figure.tight_layout() figure.savefig(figure_path, dpi=150) plt.close(figure) print(f"Config: {config_path}") print(f"Checkpoint: {checkpoint_path}") print(f"Device: {device}") print(f"Posterior states: {len(posterior_states)}") print(f"Relative L2={error:.6e}, RMSE={rmse:.6e}, MaxAbs={max_error:.6e}") print(f"Predictions: {predictions_path}") print(f"Plot: {figure_path}") if __name__ == "__main__": main()