| """Compute paper-aligned metrics and plot input, target, and prediction.""" |
|
|
| import json |
| from pathlib import Path |
|
|
| import matplotlib |
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import yaml |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def metrics(candidate_n, candidate_p, target_n, target_p): |
| rmse = np.sqrt(np.mean((candidate_n - target_n) ** 2, axis=(0, 2))) |
| sample_variable_rmse = np.sqrt(np.mean((candidate_n - target_n) ** 2, axis=2)) |
| mass_h = np.mean(np.abs(candidate_p[:, 1].sum(1) - target_p[:, 1].sum(1)) / candidate_p.shape[2]) |
| mass_r = np.mean(np.abs(candidate_p[:, 2].sum(1) - target_p[:, 2].sum(1)) / candidate_p.shape[2]) |
| h_bias = np.mean(candidate_p[:, 1] - target_p[:, 1]) |
| return {"J": float(sample_variable_rmse.mean()), "rmse_u": float(rmse[0]), |
| "rmse_h": float(rmse[1]), "rmse_r": float(rmse[2]), |
| "mass_error_h_per_point": float(mass_h), "mass_error_r_per_point": float(mass_r), |
| "h_bias": float(h_bias)} |
|
|
|
|
| def main(): |
| config = yaml.safe_load((ROOT / "conf/config.yaml").read_text()) |
| data = np.load(ROOT / config["paths"]["inference"]) |
| if str(data["format_version"]) != config["data"]["format_version"]: |
| raise ValueError("prediction format mismatch") |
| input_n, target_n, prediction_n = data["inputs"][:, :3], data["targets"], data["predictions"] |
| input_p, target_p, prediction_p = data["xa"], data["targets_physical"], data["predictions_physical"] |
| expected = (len(target_n), 3, 250) |
| if any(array.shape != expected for array in (input_n, target_n, prediction_n, input_p, target_p, prediction_p)): |
| raise ValueError("evaluation arrays must have shape [B,3,250]") |
| baseline = metrics(input_n, input_p, target_n, target_p) |
| prediction = metrics(prediction_n, prediction_p, target_n, target_p) |
| improvement = {key: float(100 * (baseline[key] - prediction[key]) / baseline[key]) |
| for key in ("J", "rmse_u", "rmse_h", "rmse_r", "mass_error_h_per_point", "mass_error_r_per_point") |
| if baseline[key] != 0} |
| if baseline["h_bias"] != 0: |
| improvement["absolute_h_bias"] = float( |
| 100 * (abs(baseline["h_bias"]) - abs(prediction["h_bias"])) / abs(baseline["h_bias"]) |
| ) |
| report = {"samples": len(target_n), "baseline_input": baseline, "prediction": prediction, |
| "relative_improvement_percent": improvement, |
| "note": "Structured synthetic engineering validation; not paper performance."} |
| values = list(baseline.values()) + list(prediction.values()) + list(improvement.values()) |
| if not np.isfinite(values).all(): |
| raise FloatingPointError("non-finite evaluation metric") |
| output = ROOT / config["paths"]["evaluation_dir"] |
| output.mkdir(parents=True, exist_ok=True) |
| (output / "metrics.json").write_text(json.dumps(report, indent=2) + "\n") |
| x = np.arange(250) * float(config["data"]["domain_km"]) / 250 |
| figure, axes = plt.subplots(3, 1, figsize=(11, 8), sharex=True) |
| for index, (axis, variable) in enumerate(zip(axes, ("u", "h", "r"))): |
| axis.plot(x, input_p[0, index], color="steelblue", label="input X^a", linewidth=1.4) |
| axis.plot(x, target_p[0, index], color="black", label="QPEns target", linewidth=1.5) |
| axis.plot(x, prediction_p[0, index], color="firebrick", label="CNN prediction", linewidth=1.3) |
| if variable == "r": |
| axis.fill_between(x, 0, data["radar"][0, 0] * max(target_p[0, 2].max(), 1e-6), color="gold", alpha=0.2, label="radar mask") |
| axis.set_ylabel(variable); axis.grid(alpha=0.2) |
| axes[0].legend(ncol=3); axes[-1].set_xlabel("distance (km)") |
| figure.suptitle("MassConservingCNN structured synthetic validation") |
| figure.tight_layout(); figure.savefig(output / "input_target_prediction.png", dpi=150); plt.close(figure) |
| print(f"evaluation={output.relative_to(ROOT)} J={prediction['J']:.6f} h_mass={prediction['mass_error_h_per_point']:.6f}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
