File size: 3,785 Bytes
9be39c5 | 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 | """Compute the paper's rainfall metrics and visualize the 90-minute forecast."""
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 to_rainfall(values, config):
data = config["data"]
radar_db = values * (float(data["radar_db_max"]) - float(data["radar_db_min"])) + float(data["radar_db_min"])
return 10 ** ((radar_db - 10 * np.log10(float(data["zr_a"]))) / (10 * float(data["zr_b"])))
def main():
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
data = np.load(ROOT / config["paths"]["inference_dir"] / "predictions.npz")
prediction, target = data["predictions"], data["targets"]
predicted_rain, target_rain = to_rainfall(prediction, config), to_rainfall(target, config)
threshold = float(config["data"]["rainfall_threshold_mm_h"])
forecast, observed = predicted_rain >= threshold, target_rain >= threshold
hits, misses = np.logical_and(forecast, observed).sum(), np.logical_and(~forecast, observed).sum()
false_alarms = np.logical_and(forecast, ~observed).sum()
eps = 1e-9
frame_correlation, per_step = [], []
for step in range(prediction.shape[1]):
left, right = prediction[:, step].reshape(-1), target[:, step].reshape(-1)
correlation = float(np.dot(left, right) / (np.sqrt(np.dot(left, left) * np.dot(right, right)) + eps))
frame_correlation.append(correlation)
step_forecast, step_observed = forecast[:, step], observed[:, step]
step_hits = np.logical_and(step_forecast, step_observed).sum()
step_misses = np.logical_and(~step_forecast, step_observed).sum()
step_false_alarms = np.logical_and(step_forecast, ~step_observed).sum()
per_step.append({
"lead_minutes": int(data["forecast_lead_minutes"][step]),
"rainfall_mse": float(np.mean((predicted_rain[:, step] - target_rain[:, step]) ** 2)),
"csi": float(step_hits / (step_hits + step_misses + step_false_alarms + eps)),
"far": float(step_false_alarms / (step_hits + step_false_alarms + eps)),
"pod": float(step_hits / (step_hits + step_misses + eps)),
"correlation": correlation,
})
metrics = {
"samples": int(len(prediction)),
"binary_cross_entropy": float(-(target * np.log(prediction.clip(1e-7, 1 - 1e-7)) +
(1 - target) * np.log((1 - prediction).clip(1e-7, 1))).mean()),
"rainfall_mse": float(np.mean((predicted_rain - target_rain) ** 2)),
"csi": float(hits / (hits + misses + false_alarms + eps)),
"far": float(false_alarms / (hits + false_alarms + eps)),
"pod": float(hits / (hits + misses + eps)),
"correlation": float(np.mean(frame_correlation)),
"per_forecast_step": 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")
steps = [0, 2, 5, 8, 11, 14]
figure, axes = plt.subplots(3, len(steps), figsize=(15, 7))
for column, step in enumerate(steps):
axes[0, column].imshow(target[0, step, 0], cmap="turbo", vmin=0, vmax=1)
axes[1, column].imshow(prediction[0, step, 0], cmap="turbo", vmin=0, vmax=1)
axes[2, column].imshow(np.abs(target[0, step, 0] - prediction[0, step, 0]), cmap="magma", vmin=0, vmax=1)
axes[0, column].set_title(f"+{(step + 1) * 6} min")
for axis in axes[:, column]:
axis.axis("off")
figure.tight_layout()
figure.savefig(output / "comparison.png", dpi=150)
plt.close(figure)
if __name__ == "__main__":
main()
|