File size: 4,216 Bytes
03573b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """Compute paper precipitation diagnostics and render a comparison figure."""
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 pattern_correlation(target, prediction):
target_anomaly = target - target.mean()
prediction_anomaly = prediction - prediction.mean()
denominator = np.sqrt(np.sum(target_anomaly ** 2) * np.sum(prediction_anomaly ** 2))
return float(np.sum(target_anomaly * prediction_anomaly) / max(float(denominator), 1e-12))
def precipitation_metrics(values, years, wet, extreme, percentile):
daily_domain_mean = values.mean(axis=(1, 2, 3))
yearly_p95 = [float(np.percentile(daily_domain_mean[years == year], percentile)) for year in np.unique(years)]
return {"P_mean_mm_day": float(values.mean()),
"yearly_P95_mm_day": yearly_p95,
"P95_mean_over_years_mm_day": float(np.mean(yearly_p95)),
"wet_day_fraction_gt_1mm": float(np.mean(values > wet)),
"extreme_day_fraction_gt_25mm": float(np.mean(values > extreme))}
def main():
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
data = np.load(ROOT / config["paths"]["inference"])
prediction, target, years = data["prediction"], data["target"], data["years"]
if prediction.shape != target.shape or prediction.shape[1:] != (1, 216, 488):
raise ValueError("evaluation requires complete [N,1,216,488] fields")
evaluation = config["evaluation"]
target_metrics = precipitation_metrics(target, years, float(evaluation["wet_day_threshold_mm"]),
float(evaluation["extreme_day_threshold_mm"]),
float(evaluation["p95_percentile"]))
prediction_metrics = precipitation_metrics(prediction, years, float(evaluation["wet_day_threshold_mm"]),
float(evaluation["extreme_day_threshold_mm"]),
float(evaluation["p95_percentile"]))
target_pattern, prediction_pattern = target.mean(0).squeeze(), prediction.mean(0).squeeze()
report = {"format_version": str(data["format_version"]), "units": str(data["units"]),
"sample_count": len(target), "target": target_metrics, "prediction": prediction_metrics,
"bias_mm_day": float((prediction - target).mean()),
"pattern_correlation": pattern_correlation(target_pattern, prediction_pattern),
"definitions": {"P": "mean precipitation over all days and grid cells",
"P95": "95th percentile of daily domain-mean P per year, then averaged",
"wet_day": "grid-cell day precipitation > 1 mm/day",
"extreme_day": "grid-cell day precipitation > 25 mm/day"}}
numbers = [report["bias_mm_day"], report["pattern_correlation"]]
numbers += list(target_metrics.values())[:1] + list(prediction_metrics.values())[:1]
if not np.isfinite(numbers).all():
raise FloatingPointError("non-finite evaluation metrics")
output = ROOT / config["paths"]["evaluation_dir"]
output.mkdir(parents=True, exist_ok=True)
(output / "metrics.json").write_text(json.dumps(report, indent=2) + "\n")
vmax = float(np.percentile(target[0], 99.5))
figure, axes = plt.subplots(1, 3, figsize=(13, 4))
for axis, field, title in zip(axes, (data["bilinear_precipitation"][0, 0], target[0, 0], prediction[0, 0]),
("Bilinear input", "Target", "SRCNN")):
image = axis.imshow(field, cmap="Blues", vmin=0, vmax=max(vmax, 1.0), aspect="auto")
axis.set_title(title); axis.set_xlabel("x"); axis.set_ylabel("y")
figure.colorbar(image, ax=axes, label="mm/day", shrink=0.8)
figure.subplots_adjust(left=0.06, right=0.9, bottom=0.12, top=0.88, wspace=0.25)
figure.savefig(output / "precipitation_comparison.png", dpi=150); plt.close(figure)
print(f"evaluation={output.relative_to(ROOT)} pattern_correlation={report['pattern_correlation']:.4f}")
if __name__ == "__main__":
main()
|