File size: 4,630 Bytes
30f7852 | 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 77 78 79 80 81 82 83 84 85 86 87 | """Evaluate 24-hour rainfall and produce per-step metrics and maps."""
import json
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import torch
import yaml
ROOT = Path(__file__).resolve().parents[1]
def categorical(prediction, target, threshold):
forecast, observed = prediction >= threshold, target >= threshold
hits = np.logical_and(forecast, observed).sum(); false = np.logical_and(forecast, ~observed).sum()
misses = np.logical_and(~forecast, observed).sum(); correct_negative = np.logical_and(~forecast, ~observed).sum()
total = hits + false + misses + correct_negative; eps = 1e-8
random_hits = (hits + misses) * (hits + false) / max(total, 1)
return {
"ACC": float((hits + correct_negative) / (total + eps)),
"TS": float(hits / (hits + misses + false + eps)),
"ETS": float((hits - random_hits) / (hits + misses + false - random_hits + eps)),
"BS": float((hits + false) / (hits + misses + eps)),
"FAR": float(false / (hits + false + eps)),
}
def fss(prediction, target, threshold, window):
forecast = torch.from_numpy((prediction >= threshold).astype(np.float32))[:, None]
observed = torch.from_numpy((target >= threshold).astype(np.float32))[:, None]
pf = torch.nn.functional.avg_pool2d(forecast, window, stride=1, padding=window // 2)
po = torch.nn.functional.avg_pool2d(observed, window, stride=1, padding=window // 2)
return float(1 - (pf - po).square().sum() / (pf.square().sum() + po.square().sum() + 1e-8))
def main():
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
archive = np.load(ROOT / config["paths"]["inference"], allow_pickle=False)
steps = int(config["data"]["time_steps"])
lead_hours = [int(value) for value in archive["lead_hours"]]
expected_3h = (steps, int(config["data"]["height"]), int(config["data"]["width"]))
for key in ("raw_3h", "target_3h", "corrected_3h"):
if archive[key].shape[1:] != expected_3h:
raise ValueError(f"{key} must have per-day shape {expected_3h}")
if lead_hours != config["data"]["lead_hours"]:
raise ValueError("inference NPZ lead_hours do not match configuration")
thresholds = [float(value) for value in config["evaluation"]["thresholds_mm_24h"]]
windows = [int(value) for value in config["evaluation"]["fss_windows"]]
metrics = {"units_24h": "mm/24h", "units_per_step": "mm/3h", "per_step_length": steps,
"lead_hours": lead_hours, "raw_24h": {}, "corrected_24h": {}, "per_step": {}}
for label in ("raw", "corrected"):
prediction_24h = archive[f"{label}_24h"]
metrics[f"{label}_24h"] = {str(int(t)): categorical(prediction_24h, archive["target_24h"], t) for t in thresholds}
metrics[f"{label}_24h"]["FSS@50"] = {str(w): fss(prediction_24h, archive["target_24h"], 50, w) for w in windows}
per_step = []
for step in range(steps):
item = {"step": step, "lead_hour": lead_hours[step]}
item.update({str(int(t)): categorical(archive[f"{label}_3h"][:, step], archive["target_3h"][:, step], t) for t in thresholds})
item["FSS@50"] = {str(w): fss(archive[f"{label}_3h"][:, step], archive["target_3h"][:, step], 50, w) for w in windows}
per_step.append(item)
metrics["per_step"][label] = per_step
output = ROOT / config["paths"]["evaluation_dir"]
output.mkdir(parents=True, exist_ok=True)
(output / "metrics.json").write_text(json.dumps(metrics, indent=2) + "\n")
figure, axes = plt.subplots(4, steps, figsize=(18, 9), constrained_layout=True)
rows = (archive["raw_3h"][0], archive["target_3h"][0], archive["corrected_3h"][0], archive["corrected_3h"][0] - archive["target_3h"][0])
labels = ("ECMWF", "CMPA target", "Corrected", "Error")
vmax = max(float(rows[0].max()), float(rows[1].max()), 1)
for row, (fields, label) in enumerate(zip(rows, labels)):
for step in range(steps):
cmap = "RdBu_r" if row == 3 else "Blues"
limit = vmax if row < 3 else max(vmax / 2, 1)
axes[row, step].imshow(fields[step], cmap=cmap, vmin=-limit if row == 3 else 0, vmax=limit)
axes[row, step].axis("off")
if row == 0: axes[row, step].set_title(f"+{archive['lead_hours'][step]} h")
if step == 0: axes[row, step].set_ylabel(label)
figure.savefig(output / "comparison.png", dpi=150); plt.close(figure)
print(f"metrics={output / 'metrics.json'} figure={output / 'comparison.png'}")
if __name__ == "__main__":
main()
|