Jinemi/thermograph
4
0
Fork 0
Code Issues 2 Pull requests 3 Projects Releases Packages Wiki Activity Actions

Simplify the score module and page: dedup builders, consolidate metric metadata (#209)

Browse source 
Backend (scoring.py):
- Fold the parallel WEIGHTS/METRIC_LABELS/DIRECTION_WORDS/TEMP_DIR_METRICS maps
  into one METRICS descriptor so a scored metric is defined in one place.
- Share one _build_entry between the seasonal and annual paths.
- Add _freq_for so the annual entry reads the wet-day / heat-stress-day share
  directly instead of re-running the full precip_divergence just for it.
- Drop unconsumed payload: per-q v6/pct and the overall bias.
- Drop _slice's dead 'annual' branch; derive SLICES from SEASONS.

Frontend (score.js):
- One scoreCell() and signedPts() helper for the four score-cell sites and the
  two signed-points chips; reuse .section-title / .table-wrap instead of cloning.

Behavior-preserving (identical scores); bumps the score cache version.
This commit is contained in:
Emi Griffith 2026-07-19 21:02:40 -07:00 committed by GitHub
parent 41e84060ab
commit 3bd41a41ff
2 changed files with 100 additions and 123 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

221
scoring.py
View file

@ -6,9 +6,9 @@ at that percentile and ask where it lands in the baseline distribution. The gap
between where it lands and ``q`` — in percentile points, so it is unit-free and
comparable across metrics — is the divergence. A metric's score is the mean
absolute divergence over the categories, mapped to 0-100; its sign (bias) says
which way it drifted. Metrics are computed per meteorological season and pooled
into an annual view, then weighted (temps / humidity / feels-like heaviest) into
one overall score.
which way it drifted. Metrics are computed per meteorological season and the
seasonal differentials are averaged into an annual view, then weighted (temps /
humidity / feels-like heaviest) into one overall score.
Baseline is the ENTIRE record, including the recent years — the same all-years
climatology every other view grades against. The recent window's ~13% overlap
@ -24,20 +24,25 @@ RECENT_YEARS = 6
QS = (10, 25, 50, 75, 90) # the scored percentile categories
SEASONS = {"djf": (12, 1, 2), "mam": (3, 4, 5),
"jja": (6, 7, 8), "son": (9, 10, 11)}
SLICES = ("annual", "djf", "mam", "jja", "son")
SLICES = ("annual", *SEASONS) # payload slice keys
# All scored metrics. wetbulb is derived at the read boundary (climate.py); the
# rest are raw daily columns. Order here is the canonical compute order; the
# frontend renders in its own display order (Precip first).
SCORE_METRICS = ("tmax", "tmin", "feels", "humid", "wetbulb", "wind", "gust", "precip")
# Temps / humidity / feels-like dominate; wet bulb medium; wind + precip lightest.
WEIGHTS = {"feels": 2.0, "humid": 2.0, "tmax": 1.5, "tmin": 1.5,
"wetbulb": 1.25, "precip": 0.75, "wind": 0.5, "gust": 0.5}
# Metrics whose signed drift feeds the headline "warmer / cooler" (so wind and
# precip signs never cancel the temperature story in the overall bias).
TEMP_DIR_METRICS = ("tmax", "tmin", "feels", "wetbulb")
# Per-metric descriptor — display label, overall-score weight, and the
# (positive-drift, negative-drift) direction words — in ONE table so a scored
# metric is defined in a single place. Temps / humidity / feels-like dominate the
# weighting; wet bulb medium; wind + precip lightest. wetbulb is derived at the
# read boundary (climate.py); the rest are raw daily columns. Insertion order is
# the canonical compute order (the frontend renders in its own display order).
METRICS = {
"tmax": {"label": "High temp", "weight": 1.5, "dir": ("warmer", "cooler")},
"tmin": {"label": "Low temp", "weight": 1.5, "dir": ("warmer", "cooler")},
"feels": {"label": "Feels like", "weight": 2.0, "dir": ("warmer", "cooler")},
"humid": {"label": "Humidity", "weight": 2.0, "dir": ("muggier", "drier")},
"wetbulb": {"label": "Wet bulb", "weight": 1.25, "dir": ("warmer", "cooler")},
"wind": {"label": "Wind", "weight": 0.5, "dir": ("windier", "calmer")},
"gust": {"label": "Gusts", "weight": 0.5, "dir": ("gustier", "calmer")},
"precip": {"label": "Precip", "weight": 0.75, "dir": ("wetter", "drier")},
}
SCORE_METRICS = tuple(METRICS)
MIN_BASELINE_YEARS = 25 # below this, no scoring at all (payload carries a reason)
MIN_SLICE_SAMPLES = 300 # recent-slice floor (~540 expected per 6-yr season)
@ -49,20 +54,6 @@ SATURATION = 15.0 # mean |divergence| (pct points) that maps to score 10
# sustained exertion. Reported as a recent-vs-baseline share alongside the score.
WETBULB_STRESS_F = 78.8 # 26 °C
METRIC_LABELS = {
"tmax": "High temp", "tmin": "Low temp", "feels": "Feels like",
"humid": "Humidity", "wetbulb": "Wet bulb", "wind": "Wind",
"gust": "Gusts", "precip": "Precip",
}
# (word for a positive drift, word for a negative drift).
DIRECTION_WORDS = {
"tmax": ("warmer", "cooler"), "tmin": ("warmer", "cooler"),
"feels": ("warmer", "cooler"), "wetbulb": ("warmer", "cooler"),
"humid": ("muggier", "drier"), "wind": ("windier", "calmer"),
"gust": ("gustier", "calmer"), "precip": ("wetter", "drier"),
}
# score threshold -> (label, css for positive bias, css for negative bias). The
# css classes are the existing 9-step temperature scale, so no new color tokens
# are needed — the frontend's TIER_COLORS resolves them for free. Lower-bound
@ -92,19 +83,16 @@ def recent_baseline_split(df: pl.DataFrame):
return df.filter(pl.col("date") > cutoff), df
def _slice(df: pl.DataFrame, key: str) -> pl.DataFrame:
"""A slice of the record: the whole thing for ``"annual"``, else the pooled
days of one meteorological season (DJF wraps the year end, but the samples are
pooled across all years so the boundary is irrelevant)."""
if key == "annual":
return df
return df.filter(pl.col("date").dt.month().is_in(list(SEASONS[key])))
def _slice(df: pl.DataFrame, season: str) -> pl.DataFrame:
"""The pooled days of one meteorological season (DJF wraps the year end, but the
samples are pooled across all years so the boundary is irrelevant)."""
return df.filter(pl.col("date").dt.month().is_in(list(SEASONS[season])))
def divergence(base: np.ndarray, rec: np.ndarray) -> dict | None:
"""Core math for one metric within one slice. For each category ``q`` in
``QS``: the recent value at that percentile, where it lands in the baseline,
and the signed gap ``d = landed − q``. Returns per-category detail plus the
and the signed gap ``d = landed − q``. Returns per-category ``d`` plus the
mean-absolute-divergence (``mad``, drives the score) and mean signed
divergence (``bias``, drives the direction). ``None`` when the recent slice is
too thin to trust."""
@ -112,11 +100,10 @@ def divergence(base: np.ndarray, rec: np.ndarray) -> dict | None:
return None
per_q, deltas = [], []
for q in QS:
v6 = float(np.percentile(rec, q))
pct = grading.empirical_percentile(base, v6)
pct = grading.empirical_percentile(base, float(np.percentile(rec, q)))
if pct is None:
continue
per_q.append({"q": q, "v6": round(v6, 2), "pct": pct, "d": round(pct - q, 1)})
per_q.append({"q": q, "d": round(pct - q, 1)})
deltas.append(pct - q)
if not per_q:
return None
@ -131,23 +118,38 @@ def precip_divergence(base: np.ndarray, rec: np.ndarray) -> dict | None:
*amounts* shifted (percentile divergence over rain days only) and how the
wet-day *frequency* shifted (points of the wet-day share). They average into
one mad/bias; the frequency alone carries it when rain days are too few for a
stable amount distribution."""
stable amount distribution. (The wet-day share for the card's freq read-out is
computed separately by ``_freq_for``.)"""
if rec.size < MIN_SLICE_SAMPLES or base.size == 0:
return None
thr = grading.RAIN_THRESHOLD
f6 = float(np.mean(rec >= thr)) * 100.0
f45 = float(np.mean(base >= thr)) * 100.0
freq_d = f6 - f45
freq_d = (float(np.mean(rec >= thr)) - float(np.mean(base >= thr))) * 100.0
wet_rec = rec[rec >= thr]
amount = divergence(base[base >= thr], wet_rec) if wet_rec.size >= MIN_WET_DAYS else None
if amount is not None:
mad = 0.5 * amount["mad"] + 0.5 * abs(freq_d)
bias = 0.5 * amount["bias"] + 0.5 * freq_d
else:
mad, bias = abs(freq_d), freq_d
return {"per_q": amount["per_q"] if amount else [],
"mad": round(mad, 1), "bias": round(bias, 1),
"freq": {"f6": round(f6, 1), "f45": round(f45, 1), "d": round(freq_d, 1)}}
return {"per_q": amount["per_q"],
"mad": round(0.5 * amount["mad"] + 0.5 * abs(freq_d), 1),
"bias": round(0.5 * amount["bias"] + 0.5 * freq_d, 1)}
return {"per_q": [], "mad": round(abs(freq_d), 1), "bias": round(freq_d, 1)}
def _freq_for(metric: str, base: np.ndarray, rec: np.ndarray) -> dict | None:
"""The day-count read-out a metric card carries, or ``None``. Precip: the
wet-day share; wet bulb: the heat-stress-day share. Both recent-vs-baseline in
percentage points, over the whole slice — counts, not distribution shifts."""
if metric == "precip":
return _day_share(base, rec, grading.RAIN_THRESHOLD)
if metric == "wetbulb":
return _day_share(base, rec, WETBULB_STRESS_F, {"threshold_f": WETBULB_STRESS_F})
return None
def _day_share(base: np.ndarray, rec: np.ndarray, thr: float, extra: dict | None = None) -> dict:
"""Share of days at or above ``thr`` — recent (``f6``) vs baseline (``f45``) and
the gap (``d``), in percentage points."""
f6 = float(np.mean(rec >= thr)) * 100.0
f45 = float(np.mean(base >= thr)) * 100.0
return {"f6": round(f6, 1), "f45": round(f45, 1), "d": round(f6 - f45, 1), **(extra or {})}
def score_of(mad: float) -> int:
@ -166,68 +168,59 @@ def tier_of(score: int, bias: float):
def _direction(metric: str, bias: float) -> str:
up, down = DIRECTION_WORDS[metric]
up, down = METRICS[metric]["dir"]
return up if bias >= 0 else down
def _null_entry(metric: str, reason: str) -> dict:
return {"key": metric, "label": METRIC_LABELS[metric], "score": None,
"weight": WEIGHTS[metric], "reason": reason}
return {"key": metric, "label": METRICS[metric]["label"], "score": None,
"weight": METRICS[metric]["weight"], "reason": reason}
def _build_entry(metric: str, mad: float, bias: float, per_q: list,
n_recent: int, n_base: int, freq: dict | None) -> dict:
"""Assemble a scored metric entry from its divergence magnitude/direction — the
one place the entry shape is defined, shared by the seasonal and annual paths."""
score = score_of(mad)
tier, css = tier_of(score, bias)
direction = _direction(metric, bias)
entry = {
"key": metric, "label": METRICS[metric]["label"],
"score": score, "mad": round(mad, 1), "bias": round(bias, 1),
"tier": tier, "class": css, "direction": direction,
"grade": tier if score < 15 else f"{tier}, {direction}",
"weight": METRICS[metric]["weight"], "per_q": per_q,
"n_recent": n_recent, "n_base": n_base,
}
if freq is not None:
entry["freq"] = freq
return entry
def _metric_entry(metric: str, base: np.ndarray, rec: np.ndarray) -> dict:
div = precip_divergence(base, rec) if metric == "precip" else divergence(base, rec)
if div is None:
return _null_entry(metric, "not enough recent data")
score = score_of(div["mad"])
tier, css = tier_of(score, div["bias"])
direction = _direction(metric, div["bias"])
entry = {
"key": metric, "label": METRIC_LABELS[metric],
"score": score, "mad": div["mad"], "bias": div["bias"],
"tier": tier, "class": css, "direction": direction,
"grade": tier if score < 15 else f"{tier}, {direction}",
"weight": WEIGHTS[metric], "per_q": div["per_q"],
"n_recent": int(rec.size), "n_base": int(base.size),
}
if "freq" in div:
entry["freq"] = div["freq"]
if metric == "wetbulb":
# How often the peak wet bulb reaches heat-stress levels — a "wet-bulb day"
# vs a normal one — recent share vs the full record.
entry["freq"] = _stress_freq(base, rec, WETBULB_STRESS_F)
return entry
def _stress_freq(base: np.ndarray, rec: np.ndarray, thr: float) -> dict:
"""Share of days at or above ``thr`` — recent vs baseline, in percentage
points. The rest of the days are 'normal'."""
f6 = float(np.mean(rec >= thr)) * 100.0
f45 = float(np.mean(base >= thr)) * 100.0
return {"f6": round(f6, 1), "f45": round(f45, 1),
"d": round(f6 - f45, 1), "threshold_f": thr}
return _build_entry(metric, div["mad"], div["bias"], div["per_q"],
int(rec.size), int(base.size), _freq_for(metric, base, rec))
def _overall(entries: dict) -> dict | None:
"""Weighted roll-up of the present metrics in one slice. Weights renormalize
over whatever is present (a metric missing for the source drops out cleanly).
The total reads as a direction-agnostic NET CHANGE: it is colored by magnitude
on the intensity ramp and labeled by tier alone, not 'warmer/cooler' — the
per-metric cards still carry direction. A signed ``bias`` (from the
temperature-direction metrics) stays in the payload for reference."""
The total reads as a direction-agnostic NET CHANGE: colored by magnitude on the
intensity ramp and labeled by tier alone, not 'warmer/cooler' — the per-metric
cards carry direction."""
present = [e for e in entries.values() if e.get("score") is not None]
if not present:
return None
wsum = sum(e["weight"] for e in present)
mad = sum(e["weight"] * e["mad"] for e in present) / wsum
dirs = [e for e in present if e["key"] in TEMP_DIR_METRICS]
bias = (sum(e["weight"] * e["bias"] for e in dirs) / sum(e["weight"] for e in dirs)
if dirs else 0.0)
score = score_of(mad)
tier, css = tier_of(score, 1.0) # magnitude only — net change, not a direction
return {"score": score, "mad": round(mad, 1), "bias": round(bias, 1),
"tier": tier, "class": css, "grade": tier, "descriptor": "net change"}
return {"score": score, "mad": round(mad, 1), "tier": tier, "class": css,
"grade": tier, "descriptor": "net change"}
def _slice_entry(metric: str, history: pl.DataFrame, base_s: pl.DataFrame,
@ -243,9 +236,9 @@ def _annual_entry(metric: str, seasonal: list, history: pl.DataFrame,
differentials rather than from an annually-pooled distribution. Pooling all
days together widens the reference spread and hides a shift confined to one
season (a hot-summer trend barely moves the all-year percentiles); averaging
the four seasons' divergences keeps it visible. Frequency read-outs
(precip wet days, wet-bulb heat-stress days) are day counts, so those stay
pooled over the whole year."""
the four seasons' divergences keeps it visible. Frequency read-outs (precip wet
days, wet-bulb heat-stress days) are day counts, so those stay pooled over the
whole year."""
if metric not in history.columns:
return _null_entry(metric, "not available for this location")
present = [e for e in seasonal if e.get("score") is not None]
@ -253,31 +246,17 @@ def _annual_entry(metric: str, seasonal: list, history: pl.DataFrame,
return _null_entry(metric, "not enough recent data")
mad = sum(e["mad"] for e in present) / len(present)
bias = sum(e["bias"] for e in present) / len(present)
score = score_of(mad)
tier, css = tier_of(score, bias)
direction = _direction(metric, bias)
per_q = []
for q in QS:
ds = [pq["d"] for e in present for pq in e.get("per_q", []) if pq["q"] == q]
ds = [pq["d"] for e in present for pq in e["per_q"] if pq["q"] == q]
if ds:
per_q.append({"q": q, "d": round(sum(ds) / len(ds), 1)})
entry = {
"key": metric, "label": METRIC_LABELS[metric],
"score": score, "mad": round(mad, 1), "bias": round(bias, 1),
"tier": tier, "class": css, "direction": direction,
"grade": tier if score < 15 else f"{tier}, {direction}",
"weight": WEIGHTS[metric], "per_q": per_q,
"n_recent": sum(e["n_recent"] for e in present),
"n_base": sum(e["n_base"] for e in present),
}
if metric == "precip":
div = precip_divergence(grading._finite(baseline["precip"]), grading._finite(recent["precip"]))
if div:
entry["freq"] = div["freq"]
elif metric == "wetbulb":
entry["freq"] = _stress_freq(grading._finite(baseline["wetbulb"]),
grading._finite(recent["wetbulb"]), WETBULB_STRESS_F)
return entry
freq = None
if metric in ("precip", "wetbulb"):
freq = _freq_for(metric, grading._finite(baseline[metric]), grading._finite(recent[metric]))
return _build_entry(metric, mad, bias, per_q,
sum(e["n_recent"] for e in present),
sum(e["n_base"] for e in present), freq)
def build_scores(history: pl.DataFrame) -> dict:
@ -293,16 +272,14 @@ def build_scores(history: pl.DataFrame) -> dict:
# Each season scored on its own distribution first…
slices = {}
for key in SEASONS:
base_s, rec_s = _slice(baseline, key), _slice(recent, key)
for season in SEASONS:
base_s, rec_s = _slice(baseline, season), _slice(recent, season)
entries = {m: _slice_entry(m, history, base_s, rec_s) for m in SCORE_METRICS}
slices[key] = {"overall": _overall(entries), "metrics": entries}
slices[season] = {"overall": _overall(entries), "metrics": entries}
# …then the annual headline is the average of those seasonal differentials.
annual = {}
for m in SCORE_METRICS:
seasonal = [slices[k]["metrics"][m] for k in SEASONS]
annual[m] = _annual_entry(m, seasonal, history, baseline, recent)
annual = {m: _annual_entry(m, [slices[s]["metrics"][m] for s in SEASONS],
history, baseline, recent) for m in SCORE_METRICS}
slices = {"annual": {"overall": _overall(annual), "metrics": annual}, **slices}
return {

2
views.py
View file

@ -108,7 +108,7 @@ def forecast_key(today, days: int) -> str:
# is the plain `history_token` (expires when the archive tail advances). The key
# just carries the scoring-math version, so a math change invalidates only score
# rows without disturbing any other kind.
SCORE_VER = "s3"
SCORE_VER = "s4"
def score_key() -> str: