File size: 4,361 Bytes
b11ef36 | 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 | """Compute paper metrics, Theil-Sen trends, and the 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 metrics(observed, predicted):
error = predicted - observed
denominator = np.maximum(np.sum(observed, axis=0), 1e-12)
r2_denominator = np.maximum(np.sum((observed - observed.mean(0)) ** 2, axis=0), 1e-12)
ioa_denominator = np.maximum(np.sum((np.abs(predicted - observed.mean(0)) +
np.abs(observed - observed.mean(0))) ** 2, axis=0), 1e-12)
return {"RMSE": np.sqrt(np.mean(error ** 2, axis=0)),
"r2": 1 - np.sum(error ** 2, axis=0) / r2_denominator,
"FAC2": np.mean((predicted / np.maximum(observed, 1e-12) >= .5) &
(predicted / np.maximum(observed, 1e-12) <= 2), axis=0),
"MB": np.mean(error, axis=0), "MGE": np.mean(np.abs(error), axis=0),
"NMB": np.sum(error, axis=0) / denominator,
"NMGE": np.sum(np.abs(error), axis=0) / denominator,
"COE": 1 - np.sum(np.abs(error), axis=0) /
np.maximum(np.sum(np.abs(observed - observed.mean(0)), axis=0), 1e-12),
"IOA": 1 - np.sum(error ** 2, axis=0) / ioa_denominator}
def theil_sen(x, y, max_pairs=200000, seed=0):
"""Median pairwise slope, with deterministic subsampling for large series."""
x, y = np.asarray(x, float), np.asarray(y, float)
total = len(x) * (len(x) - 1) // 2
if total <= max_pairs:
slopes = [(y[j] - y[i]) / (x[j] - x[i]) for i in range(len(x) - 1)
for j in range(i + 1, len(x)) if x[j] != x[i]]
else:
rng = np.random.default_rng(seed)
left = rng.integers(0, len(x) - 1, max_pairs)
right = rng.integers(left + 1, len(x), max_pairs)
valid = x[right] != x[left]
slopes = (y[right[valid]] - y[left[valid]]) / (x[right[valid]] - x[left[valid]])
slope = float(np.median(slopes))
return slope, float(np.median(y - slope * x))
def main():
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
data = np.load(ROOT / config["paths"]["inference"])
names = [str(value) for value in data["pollutant_names"]]
scores = metrics(data["observed"], data["predicted"])
report = {"metrics": {name: {metric: float(values[i]) for metric, values in scores.items()}
for i, name in enumerate(names)}, "theil_sen_per_year": {}}
trends = []
for i, name in enumerate(names):
raw = theil_sen(data["ttrend"], data["observed"][:, i], seed=i)
normalized = theil_sen(data["ttrend"], data["normalized"][:, i], seed=100 + i)
report["theil_sen_per_year"][name] = {"observed_slope": raw[0],
"normalized_slope": normalized[0]}
trends.append((raw[0], normalized[0]))
output = ROOT / config["paths"]["evaluation_dir"]
output.mkdir(parents=True, exist_ok=True)
(output / "metrics.json").write_text(json.dumps(report, indent=2) + "\n")
figure, axes = plt.subplots(2, 1, figsize=(11, 8), gridspec_kw={"height_ratios": [2, 1]})
order = np.argsort(data["ttrend"])
sample = order[::max(1, len(order) // 600)]
axes[0].plot(data["ttrend"][sample], data["observed"][sample, 0], ".", alpha=.25, label="observed PM2.5")
axes[0].plot(data["ttrend"][sample], data["normalized"][sample, 0], ".", alpha=.4, label="weather-normalized PM2.5")
axes[0].set(ylabel="Concentration (ug m-3)", title="Observed and meteorologically normalized test samples")
axes[0].legend()
positions = np.arange(len(names)); width = .36
axes[1].bar(positions - width / 2, [v[0] for v in trends], width, label="observed")
axes[1].bar(positions + width / 2, [v[1] for v in trends], width, label="normalized")
axes[1].axhline(0, color="black", linewidth=.7); axes[1].set_xticks(positions, names)
axes[1].set(ylabel="Theil-Sen slope per year", title="Robust concentration trends")
axes[1].legend(); figure.tight_layout(); figure.savefig(output / "comparison.png", dpi=160)
plt.close(figure)
print(f"evaluation={output.relative_to(ROOT)} pollutants={len(names)}")
if __name__ == "__main__":
main()
|