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

narrative: precomputed plain-English day summaries #102

Merged
emi merged 1 commit from feat/grade-narrative into dev 2026-08-02 01:16:38 +00:00
Conversation 0 Commits 1 Files changed 7 +1079 -1
7 changed files with 1079 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
backend/api/narrative_routes.py Normal file
View file

@ -0,0 +1,104 @@
"""/api/v2/narrative — serve the precomputed plain-English summary for one day.
This endpoint is **read-only with respect to the model**. It never calls Ollama,
never blocks on one, and has no code path that could. It reads one row from the
derived store and returns it. That is the entire point of the design: inference
runs on the operator's desktop on a batch cadence
(``scripts/generate_narratives.py``), so thermograph.org keeps answering at
database speed whether or not that machine is powered on, whether or not the
WireGuard mesh is up, and whether or not a GPU is currently present in it.
A cache miss is an ordinary, expected outcome — a cell nobody has generated for
yet, or a day beyond what the last batch covered. It answers 200 with
``narrative: null`` and ``status: "pending"`` rather than 404, because "no
sentence yet" is not a client error and a frontend should render the page
without the sentence rather than treat the response as a failure.
"""
import datetime
import hashlib
import json
import os
from fastapi import APIRouter, HTTPException, Query, Request, Response
from data import grid
from data import narrator
from data import store
router = APIRouter(tags=["narrative"])
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""
def _parse_date(value: str | None) -> datetime.date:
"""Target date, defaulting to today. Mirrors web/app.py's _parse_target_date
contract (400 on an unparseable date rather than a silent fallback)."""
if not value:
return datetime.date.today()
try:
return datetime.date.fromisoformat(value)
except ValueError:
raise HTTPException(status_code=400, detail="date must be YYYY-MM-DD")
def _etag_for(cell_id: str, key: str, token: str, stamp) -> str:
"""Weak ETag over the row's identity *and* its generation stamp.
Folding the stamp in is what makes a pending response safely cacheable: while
the narrative is missing the tag is stable (so revalidation is a cheap 304),
and the moment the batch job writes the row the stamp changes, the tag
changes, and the client gets the sentence on its next revalidate."""
raw = f"{cell_id}:{key}:{token}:{stamp if stamp is not None else 'pending'}"
return f'W/"{hashlib.sha1(raw.encode()).hexdigest()[:20]}"'
def _not_modified(request: Request, etag: str) -> bool:
inm = request.headers.get("if-none-match")
if not inm:
return False
tags = {t.strip() for t in inm.split(",")}
return "*" in tags or etag in tags or etag.removeprefix("W/") in tags
def api_narrative(
request: Request,
lat: float = Query(..., ge=-90, le=90),
lon: float = Query(..., ge=-180, le=180),
date: str | None = Query(None, description="target date YYYY-MM-DD (default today)"),
):
"""The stored one-sentence summary of how unusual this day was here.
Percentile-relative, like every other Thermograph surface: it describes where
the day sits against ~45 years of this location's own history, and by
construction contains no temperature or rainfall figure (see data/narrator.py
— the model is never given one). Not a forecast.
"""
target = _parse_date(date)
cell = grid.snap(lat, lon)
key = narrator.narrative_key(target)
token = narrator.narrative_token()
row = store.get_narrative(cell["id"], key, token)
etag = _etag_for(cell["id"], key, token, (row or {}).get("generated_at"))
if _not_modified(request, etag):
return Response(status_code=304, headers={"ETag": etag})
payload = {
"cell": {"id": cell["id"],
"center_lat": cell["center_lat"],
"center_lon": cell["center_lon"]},
"date": target.isoformat(),
"narrative": row["text"] if row else None,
"status": "ready" if row else "pending",
"model": row["model"] if row else None,
"generated_at": row["generated_at"] if row else None,
}
return Response(
content=json.dumps(payload, separators=(",", ":")).encode(),
media_type="application/json",
headers={"ETag": etag},
)
router.add_api_route("/narrative", api_narrative, methods=["GET"])

325
backend/data/narrator.py Normal file
View file

@ -0,0 +1,325 @@
"""Plain-English grade narratives — a local LLM phrases one graded day.
Thermograph's payloads say a day sits in the 92nd percentile and earns the
"Very High" tier. That is precise and completely inert to read. This module
turns one graded day into a sentence a person actually absorbs, using a model
served by Ollama on the operator's desktop.
Three properties hold this together, and every one of them is load-bearing:
* **Generation never happens on the request path.** Nothing here is imported by
a route handler that has to answer quickly. ``scripts/generate_narratives.py``
fills the cache; ``api/narrative_routes.py`` only ever *reads* it. A public
endpoint therefore has no dependency on the desktop being powered on, the
mesh being up, or a model being loaded — the worst case is a cache miss, which
serves ``null`` and renders as nothing.
* **The model never sees a number it could misreport.** ``facts_from_grade``
hands it tier labels and percentile ordinals only — never a temperature, never
a raw reading. A model that hallucinates cannot invent "38°C" because it was
never told a degree value, and the sentence it writes is checked against
:data:`_BANNED` for units it had no business emitting. This is the whole
reason narratives are unit-free rather than rendered per-country: it removes
an entire class of wrong output, and as a bonus there is exactly one cached
narrative per (cell, date) instead of a °C and a °F variant.
* **The framing stays relative.** Thermograph grades how *unusual* weather is
against ~45 years of that exact location's history — it is not a forecast and
not a thermometer. The prompt says so, and the sanitizer drops output that
drifts into prediction or advice.
Configuration is entirely environmental, so the same code runs on the desktop
(Ollama on localhost) and on a host that reaches it across WireGuard:
* ``THERMOGRAPH_OLLAMA_URL`` — default ``http://127.0.0.1:11434``
* ``THERMOGRAPH_NARRATIVE_MODEL`` — default ``qwen3:14b``
* ``THERMOGRAPH_NARRATIVE_TIMEOUT`` — seconds, default 120 (generous on purpose:
a CPU-only fallback is roughly an order of magnitude slower than the GPU, and
a batch job would rather wait than lose the row)
"""
import datetime
import math
import os
import re
import httpx
# Bump to invalidate every stored narrative at once — it is half of the validity
# token (the model name is the other half), so a prompt or house-style change
# here orphans the old rows rather than silently mixing two generations of text.
# n2: fact lines state the direction ("warmer than 82% of nights…") instead of a
# bare "82nd percentile (High)", which models could and did read backwards.
NARRATIVE_VER = "n2"
OLLAMA_URL = os.environ.get("THERMOGRAPH_OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/")
MODEL = os.environ.get("THERMOGRAPH_NARRATIVE_MODEL", "qwen3:14b").strip()
TIMEOUT = float(os.environ.get("THERMOGRAPH_NARRATIVE_TIMEOUT", "120"))
# Hard ceiling on what we will store. The prompt asks for one sentence; this is
# the backstop for a model that ignores it. Sentence-truncated, not chopped
# mid-word (see _sanitize).
MAX_CHARS = 240
# The metrics a narrative may mention, in prompt order: payload key, the label
# used in the prompt, and the noun for what the percentile ranks against.
# Anything absent from the graded day is simply omitted — a cell with no
# feels-like reading yields a shorter fact list, never a "None" in the prompt.
#
# The noun matters more than it looks. An earlier version of this prompt wrote
# "Overnight low: 82nd percentile (High)", which a model read as "the low was
# high" and phrased as a *cool* night — an exact inversion of the fact, and one
# no output filter can catch because nothing about the sentence looks wrong.
# Stating the direction outright ("warmer than 82% of nights…") removes the
# ambiguity at the source; the tier is then along for corroboration, not
# interpretation.
_METRICS = (
("tmax", "Daytime high", "days"),
("tmin", "Overnight low", "nights"),
("feels", "Feels-like", "days"),
("precip", "Rain", "rainy days"),
)
# Units and forecast vocabulary the model was never given and must not produce.
# Matched case-insensitively against the finished sentence; a hit voids the
# generation rather than storing something that reads authoritative and is not.
_BANNED = re.compile(
r"(\d+\s*(°|deg|celsius|fahrenheit|c\b|f\b)" # any degree reading
r"|\b(mm|cm|inches|inch|mph|km/?h|knots?)\b" # any measured quantity
r"|\b(tomorrow|forecast|expect|will be|should be)\b" # prediction
r"|\b(wear|bring|advise|recommend|stay hydrated)\b)", # advice
re.I,
)
# qwen3 is a thinking model. We pass think=False, but a reasoning block leaking
# into the response is exactly the kind of thing that changes between releases,
# so it is stripped defensively too.
_THINK = re.compile(r"<think\b[^>]*>.*?</think\s*>", re.I | re.S)
# Models open with a conversational lead-in more often than they don't, and it
# arrives in two shapes: a bare interjection ("Sure!", "Okay,") and a labelled
# hand-off ("Here's a sentence:"), frequently both at once. Each is anchored to a
# known opener rather than to "anything before a colon", so a real narrative that
# happens to use a colon — "One thing stands out: the night was extraordinary" —
# survives untouched.
_LEADIN = re.compile(r"^\s*(?:sure|certainly|of course|okay|ok|absolutely|got it)"
r"\b[!,.…]*\s*", re.I)
_LABEL = re.compile(r"^\s*(?:here(?:'|’)?s|here is|sentence|summary|answer)"
r"\b[^:\n]{0,60}:\s*", re.I)
def _pct_int(pct) -> int:
"""A percentile as a whole number for the prompt, clamped to 1..99.
Same floor(x + 0.5) and same clamp as grading.pct_ordinal, deliberately: a
narrative saying "warmer than 92%" sitting beside a UI reading "92nd" is the
least confusing outcome, and the clamp keeps "warmer than 100% of days" —
which reads as a measurement error — off the page."""
return min(99, max(1, math.floor(float(pct) + 0.5)))
def narrative_key(target: datetime.date) -> str:
"""Derived-store key: one narrative per cell per date."""
return target.isoformat()
def narrative_token(model: str | None = None) -> str:
"""Validity token. Deliberately independent of the climate sync stamps.
The obvious move — folding ``recent_synced_at`` in, the way
``payloads.recent_token`` does — would invalidate every narrative every time
the forecast bundle refreshes (~4h). Regenerating the whole grid that often
is not affordable against a single desktop GPU, let alone the CPU fallback,
and it buys nothing: a sentence about a day being "cooler than four nights in
five" does not stop being true because the high moved a tenth of a degree.
Freshness is instead the generator's job — it re-runs for the current day on
its own cadence and overwrites in place (put_narrative upserts). This token
exists to orphan rows when the *phrasing contract* changes: the prompt/house
style (NARRATIVE_VER) or the model behind it."""
return f"{NARRATIVE_VER}:{model or MODEL}"
def _day_row(payload: dict, target: datetime.date) -> dict | None:
"""The graded row for ``target`` inside a /grade payload's ``recent`` array."""
want = target.isoformat()
for row in payload.get("recent") or ():
if row.get("date") == want:
return row
return None
def facts_from_grade(payload: dict, target: datetime.date,
place: str | None = None) -> dict | None:
"""Reduce a /grade payload to the handful of facts the model may use.
Returns None when the target day carries no gradeable metric at all — there
is nothing to say about it, and an empty fact list would invite the model to
invent something. Note what is *not* in the result: no temperatures, no
cell coordinates, no climatology arrays. See the module docstring.
``place`` overrides the payload's own label. The payload carries a
*reverse-geocoded* name for the ~2-mile cell, which in a dense city is a
neighbourhood — grading Shanghai yields "Nanjingdonglu Subdistrict", and a
sentence opening on that reads like it is about somewhere else. A caller that
already knows which city it asked about should say so."""
row = _day_row(payload, target)
if not row:
return None
metrics = []
for key, label, noun in _METRICS:
graded = row.get(key)
if not isinstance(graded, dict):
continue
grade = graded.get("grade")
if not grade:
continue
pct = graded.get("percentile")
# A dry day has a tier ("Dry") but no percentile — it is ranked among rain
# days only, and it isn't one. Carry pct=None and let build_prompt say so
# in words rather than emitting a percentage that doesn't exist.
metrics.append({"label": label, "grade": grade, "noun": noun,
"pct": _pct_int(pct) if pct is not None else None})
if not metrics:
return None
if place is None:
raw = payload.get("place")
place = (raw or {}).get("label") if isinstance(raw, dict) else raw
return {"place": place, "date": target.isoformat(), "metrics": metrics}
def build_prompt(facts: dict) -> str:
"""The full prompt for one narrative. Pure function of ``facts`` — same facts
in, same prompt out, which is what makes a regeneration reproducible."""
when = datetime.date.fromisoformat(facts["date"]).strftime("%-d %B")
lines = [
"You write one-sentence summaries for Thermograph, a service that grades "
"how unusual a day's weather is against about 45 years of that exact "
"location's own history.",
"",
"Rules:",
"- Describe how UNUSUAL the day is for this place at this time of year.",
"- Use only the facts listed below. Do not add any fact that is not there.",
"- Never state a temperature, a rainfall amount, or any other measurement. "
"You have not been given any, and inventing one is a serious error.",
"- Never forecast, predict, or give advice.",
"- One sentence, at most 25 words. Plain prose — no markdown, no quotes, "
"no preamble, no trailing commentary.",
"",
"Facts:",
]
if facts.get("place"):
lines.append(f"Place: {facts['place']}")
lines.append(f"Date: {when}")
for m in facts["metrics"]:
if m["pct"] is None:
# The only percentile-less case in practice is a dry day, whose tier
# ("Dry") is already a plain-English statement of the fact.
lines.append(f"{m['label']}: {m['grade'].lower()} — no measurable rain")
continue
# "warmer/heavier than N% of <noun> at this time of year here" — the
# direction is stated, so the tier cannot be read backwards. See _METRICS.
verb = "heavier" if m["label"] == "Rain" else "warmer"
lines.append(f"{m['label']}: {verb} than {m['pct']}% of {m['noun']} "
f"at this time of year here (tier: {m['grade']})")
lines += ["", "Sentence:"]
return "\n".join(lines)
def _sanitize(text: str) -> str | None:
"""Normalize a raw completion, or None if it isn't fit to store.
Rejecting is always safe here: the caller stores nothing and the endpoint
serves null, which renders as an absent sentence rather than a wrong one."""
if not text:
return None
text = _THINK.sub(" ", text)
text = _LABEL.sub("", _LEADIN.sub("", text))
text = re.sub(r"\s+", " ", text).strip()
text = text.strip('"“”‘’\'').strip()
if len(text) < 20: # too short to be a real sentence
return None
if _BANNED.search(text): # a measurement or a forecast it was never given
return None
if len(text) > MAX_CHARS:
# Truncate at the last sentence end that fits; if there isn't one, the
# model rambled past the limit without punctuation — treat that as a
# failed generation rather than storing a fragment.
cut = max(text.rfind(". ", 0, MAX_CHARS), text.rfind("! ", 0, MAX_CHARS),
text.rfind("? ", 0, MAX_CHARS))
if cut < 20:
return None
text = text[:cut + 1]
return text
def generate(facts: dict, *, model: str | None = None, client: httpx.Client | None = None) -> str | None:
"""One narrative for one graded day, or None on any failure.
Fail-soft by design and by convention (the derived store's readers behave the
same way): an unreachable Ollama, a cold model that blows the timeout, a
refusal, or output that trips the sanitizer all return None. The batch job
logs it, skips the cell, and moves on — a missing narrative is a non-event,
a wrong one is not."""
prompt = build_prompt(facts)
body = {
"model": model or MODEL,
"prompt": prompt,
"stream": False,
# Reasoning buys nothing for a one-sentence rephrasing and costs the bulk
# of the tokens — decisive when the GPU is unavailable and this is running
# on CPU at single-digit tokens/sec.
"think": False,
"options": {
# Low but not zero: greedy decoding on a small instruct model tends to
# produce the same stock opener for every cell, which reads worse
# across a grid than mild variation.
"temperature": 0.4,
"top_p": 0.9,
# ~25 words plus slack; a hard stop is the cheapest defence against a
# model that decides to write an essay.
"num_predict": 96,
"stop": ["\n\n"],
},
}
try:
if client is not None:
r = client.post(f"{OLLAMA_URL}/api/generate", json=body, timeout=TIMEOUT)
else:
with httpx.Client(timeout=TIMEOUT) as c:
r = c.post(f"{OLLAMA_URL}/api/generate", json=body)
r.raise_for_status()
return _sanitize((r.json() or {}).get("response") or "")
except Exception: # noqa: BLE001 - see docstring: every failure is a skip
return None
def health() -> dict:
"""Whether the configured Ollama can serve the configured model, and where it
would run. Used by the batch job to fail fast with a useful message instead of
grinding through a thousand cells that will each time out.
``processor`` is the interesting field in practice: Ollama reports "100% CPU"
when no accelerator is present, which is the difference between a full-grid
pass taking minutes and taking most of a day."""
out = {"url": OLLAMA_URL, "model": MODEL, "reachable": False,
"model_present": False, "processor": None}
try:
with httpx.Client(timeout=10.0) as c:
tags = c.get(f"{OLLAMA_URL}/api/tags").json()
out["reachable"] = True
out["model_present"] = any(
m.get("name") == MODEL or m.get("model") == MODEL
for m in tags.get("models") or ()
)
for m in (c.get(f"{OLLAMA_URL}/api/ps").json() or {}).get("models") or ():
if m.get("name") == MODEL or m.get("model") == MODEL:
total = m.get("size") or 0
gpu = m.get("size_vram") or 0
out["processor"] = ("100% CPU" if not gpu
else "100% GPU" if gpu >= total
else f"{round(100 * gpu / total)}% GPU")
except Exception: # noqa: BLE001 - health is advisory; callers handle False
pass
return out

81
backend/data/store.py
View file

@ -68,6 +68,15 @@ CREATE TABLE IF NOT EXISTS revgeo (
label TEXT, -- NULL == lookup ran but found no label
updated_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS narrative (
cell_id TEXT NOT NULL,
key TEXT NOT NULL, -- target date (ISO)
token TEXT NOT NULL, -- validity: narrator.narrative_token() (version + model)
text TEXT NOT NULL, -- the sentence; short enough not to be worth compressing
model TEXT NOT NULL, -- which model wrote it, for provenance in the payload
updated_at REAL NOT NULL,
PRIMARY KEY (cell_id, key)
);
"""
# Postgres mirror of _SCHEMA. UNLOGGED = fast, crash-non-durable (the cache is
@ -93,6 +102,24 @@ _PG_SCHEMA = (
updated_at DOUBLE PRECISION NOT NULL
)
""",
# LOGGED, unlike its two siblings above — the one table here that is not
# cheaply rebuildable. `derived` and `revgeo` are UNLOGGED because losing them
# costs a recompute from parquet or a Nominatim round-trip. Losing `narrative`
# costs an LLM pass over the whole grid on a single desktop GPU (and far worse
# on the CPU fallback), which is hours, not milliseconds. Durability is worth
# more than the write throughput here: narratives are written once per cell
# per day by a batch job, never on the request path.
"""
CREATE TABLE IF NOT EXISTS narrative (
cell_id TEXT NOT NULL,
key TEXT NOT NULL,
token TEXT NOT NULL,
text TEXT NOT NULL,
model TEXT NOT NULL,
updated_at DOUBLE PRECISION NOT NULL,
PRIMARY KEY (cell_id, key)
)
""",
)
# Statement text differs by dialect (placeholder style + upsert syntax); the SQLite
@ -113,11 +140,23 @@ if IS_POSTGRES:
"INSERT INTO revgeo (key, label, updated_at) VALUES (%s,%s,%s) "
"ON CONFLICT (key) DO UPDATE SET label=EXCLUDED.label, updated_at=EXCLUDED.updated_at"
)
_SQL_GET_NARRATIVE = (
"SELECT token, text, model, updated_at FROM narrative WHERE cell_id=%s AND key=%s")
_SQL_PUT_NARRATIVE = (
"INSERT INTO narrative (cell_id, key, token, text, model, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s) "
"ON CONFLICT (cell_id, key) DO UPDATE SET "
"token=EXCLUDED.token, text=EXCLUDED.text, model=EXCLUDED.model, "
"updated_at=EXCLUDED.updated_at"
)
else:
_SQL_GET_PAYLOAD = "SELECT token, payload FROM derived WHERE kind=? AND cell_id=? AND key=?"
_SQL_PUT_PAYLOAD = "INSERT OR REPLACE INTO derived VALUES (?,?,?,?,?,?)"
_SQL_GET_REVGEO = "SELECT label, updated_at FROM revgeo WHERE key=?"
_SQL_PUT_REVGEO = "INSERT OR REPLACE INTO revgeo VALUES (?,?,?)"
_SQL_GET_NARRATIVE = (
"SELECT token, text, model, updated_at FROM narrative WHERE cell_id=? AND key=?")
_SQL_PUT_NARRATIVE = "INSERT OR REPLACE INTO narrative VALUES (?,?,?,?,?,?)"
# SQLite connections aren't shareable across threads; uvicorn serves requests on
# a thread pool, so each thread lazily opens (and keeps, for its lifetime) its own
@ -312,15 +351,55 @@ def put_revgeo(key: str, label: str | None) -> None:
pass
# --- grade narratives --------------------------------------------------------
# Written only by scripts/generate_narratives.py, read only by
# api/narrative_routes.py. Same token-as-validity rule as `derived`: a row whose
# stored token doesn't match the caller's is a miss, so a prompt or model change
# can never serve text written under the old contract.
def get_narrative(cell_id: str, key: str, token: str) -> dict | None:
"""The stored sentence for one cell/date, or None when absent or token-invalid.
Returns the provenance alongside the text — the endpoint surfaces which model
wrote it and when, so a narrative that reads oddly can be traced to a specific
generation without guessing."""
try:
with _conn() as conn:
if conn is None:
return None
row = conn.execute(_SQL_GET_NARRATIVE, (cell_id, key)).fetchone()
if row is None or row[0] != token:
return None
return {"text": row[1], "model": row[2], "generated_at": row[3]}
except Exception: # noqa: BLE001
return None
def put_narrative(cell_id: str, key: str, token: str, text: str, model: str) -> None:
"""Upsert one narrative. Overwriting in place is the freshness mechanism for
the current day — see narrator.narrative_token for why the token deliberately
doesn't move with the climate sync stamps."""
try:
with _conn() as conn:
if conn is None:
return
conn.execute(_SQL_PUT_NARRATIVE, (cell_id, key, token, text, model, time.time()))
conn.commit()
except Exception: # noqa: BLE001
pass
def stats() -> dict:
"""Row counts + on-disk size, for the migrate script's summary output."""
out = {"db_path": os.path.abspath(DB_PATH), "derived": 0, "revgeo": 0, "bytes": 0}
out = {"db_path": os.path.abspath(DB_PATH), "derived": 0, "revgeo": 0,
"narrative": 0, "bytes": 0}
try:
with _conn() as conn:
if conn is None:
return out
out["derived"] = conn.execute("SELECT COUNT(*) FROM derived").fetchone()[0]
out["revgeo"] = conn.execute("SELECT COUNT(*) FROM revgeo").fetchone()[0]
out["narrative"] = conn.execute("SELECT COUNT(*) FROM narrative").fetchone()[0]
# Recent writes may still live in the WAL sidecar — count both files.
out["bytes"] = sum(os.path.getsize(p) for p in (DB_PATH, DB_PATH + "-wal")
if os.path.exists(p))

187
backend/scripts/generate_narratives.py Normal file
View file

@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""Fill the narrative cache — the batch half of the plain-English day summaries.
This is the only thing in the codebase that calls the LLM. It walks the city
list, asks the running API for each city's graded day, has a local model phrase
it, and writes the sentence to the derived store. ``/api/v2/narrative`` then
serves those rows without ever touching a model.
# on the desktop, where Ollama is
python scripts/generate_narratives.py --limit 25
# from a host that reaches the desktop across the mesh
THERMOGRAPH_OLLAMA_URL=http://10.x.x.x:11434 \
python scripts/generate_narratives.py
Two connections, deliberately separate:
* ``--api-base`` (or ``THERMOGRAPH_API_BASE_INTERNAL``) — where to read graded
days from. Reusing the real endpoint means this script inherits the whole
fetch/grade/cache pipeline instead of reimplementing it, and warms the grade
cache as a side effect.
* ``THERMOGRAPH_OLLAMA_URL`` — where inference runs. It needs no access to
anything else, and nothing waits on it but this process.
Safe to interrupt and safe to re-run: work is per-city and committed as it goes,
and an existing valid row is skipped unless ``--force``. Nothing here writes to
climate data — the only table it touches is ``narrative``.
"""
import argparse
import datetime
import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import httpx # noqa: E402
from data import cities # noqa: E402
from data import grid # noqa: E402
from data import narrator # noqa: E402
from data import store # noqa: E402
DEFAULT_API = os.environ.get("THERMOGRAPH_API_BASE_INTERNAL", "http://127.0.0.1:8000").rstrip("/")
# The app mounts its API under THERMOGRAPH_BASE (default "/thermograph"), but a
# deployment can and does run with it empty — the LAN dev container serves
# /api/v2/... flat. Guessing wrong turns every city into a 404, so the prefix is
# probed once against the I/O-free /api/version rather than assumed.
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
_PREFIXES = tuple(dict.fromkeys([f"/{_BASE}" if _BASE else "", ""]))
def _detect_prefix(client: httpx.Client, api: str) -> str | None:
for prefix in _PREFIXES:
try:
if client.get(f"{api}{prefix}/api/version", timeout=15).status_code == 200:
return prefix
except Exception: # noqa: BLE001 - try the next candidate
continue
return None
def _grade(client: httpx.Client, api: str, prefix: str, lat: float, lon: float,
target: datetime.date) -> dict | None:
"""One city's /grade payload, or None if the API couldn't produce it.
``after=0`` and a short history window keep the response small: the narrative
only ever describes the target day, so grading two weeks either side of it
would be work this script pays for and then discards."""
try:
r = client.get(f"{api}{prefix}/api/v2/grade",
params={"lat": lat, "lon": lon, "date": target.isoformat(),
"days": 1, "after": 0})
r.raise_for_status()
return r.json()
except Exception as exc: # noqa: BLE001
print(f" grade failed: {type(exc).__name__}", flush=True)
return None
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument("--api-base", default=DEFAULT_API,
help=f"backend base URL to read graded days from (default {DEFAULT_API})")
ap.add_argument("--date", default=None, help="target date YYYY-MM-DD (default today)")
ap.add_argument("--limit", type=int, default=0, help="stop after N cities (0 = all)")
ap.add_argument("--slug", action="append", default=[],
help="generate only this city slug (repeatable)")
ap.add_argument("--force", action="store_true",
help="regenerate even when a valid row already exists")
ap.add_argument("--dry-run", action="store_true",
help="print the sentence instead of storing it")
args = ap.parse_args()
target = (datetime.date.fromisoformat(args.date) if args.date
else datetime.date.today())
token = narrator.narrative_token()
key = narrator.narrative_key(target)
# Fail fast and loudly rather than timing out a thousand times in a row. The
# processor line is the one people actually need: an accelerator that has
# silently dropped out turns a minutes-long pass into an all-day one.
hp = narrator.health()
print(f"ollama : {hp['url']} reachable={hp['reachable']} "
f"model={hp['model']} present={hp['model_present']}")
if not hp["reachable"]:
print("ERROR: cannot reach Ollama. Set THERMOGRAPH_OLLAMA_URL.", file=sys.stderr)
return 1
if not hp["model_present"]:
print(f"ERROR: model {hp['model']!r} not pulled. Run: ollama pull {hp['model']}",
file=sys.stderr)
return 1
work = cities.all_cities()
if args.slug:
wanted = set(args.slug)
work = [c for c in work if c["slug"] in wanted]
missing = wanted - {c["slug"] for c in work}
if missing:
print(f"ERROR: unknown slug(s): {', '.join(sorted(missing))}", file=sys.stderr)
return 1
if args.limit > 0:
work = work[:args.limit]
print(f"target : {target} token={token}")
print(f"cities : {len(work)}\n")
made = skipped = failed = 0
started = time.time()
with httpx.Client(timeout=120.0) as api_client, httpx.Client(timeout=narrator.TIMEOUT) as llm:
prefix = _detect_prefix(api_client, args.api_base)
if prefix is None:
print(f"ERROR: no Thermograph API at {args.api_base} "
f"(tried {', '.join(repr(p) for p in _PREFIXES)})", file=sys.stderr)
return 1
print(f"api : {args.api_base}{prefix}/api/v2\n")
for n, city in enumerate(work, 1):
label = f"{city['name']}, {city['country']}"
cell = grid.snap(city["lat"], city["lon"])
if not args.force and store.get_narrative(cell["id"], key, token):
skipped += 1
continue
print(f"[{n}/{len(work)}] {label}", flush=True)
payload = _grade(api_client, args.api_base, prefix,
city["lat"], city["lon"], target)
if payload is None:
failed += 1
continue
# Name the city we asked about, not the cell's reverse-geocoded
# neighbourhood — see narrator.facts_from_grade.
facts = narrator.facts_from_grade(payload, target,
place=cities.display_name(city))
if facts is None:
# No gradeable metric for this day — usually a cell whose recent
# bundle hasn't synced yet. Not an error; it'll be picked up on a
# later pass once the data lands.
print(" no graded data for the target day — skipping", flush=True)
skipped += 1
continue
text = narrator.generate(facts, client=llm)
if not text:
print(" generation failed or was rejected by the sanitizer", flush=True)
failed += 1
continue
print(f" {text}", flush=True)
if not args.dry_run:
store.put_narrative(cell["id"], key, token, text, narrator.MODEL)
made += 1
elapsed = time.time() - started
print(f"\ndone in {elapsed:.0f}s — {made} written, {skipped} skipped, {failed} failed")
if args.dry_run:
print("(dry run — nothing was stored)")
# A run that produced nothing while having work to do is a failure worth a
# non-zero exit, so a cron wrapper notices.
return 1 if made == 0 and failed > 0 else 0
if __name__ == "__main__":
raise SystemExit(main())

109
backend/tests/api/test_narrative_routes.py Normal file
View file

@ -0,0 +1,109 @@
"""/api/v2/narrative — the read-only half of the plain-English day summaries.
The load-bearing assertion here is the negative one: this route must never reach
for a model. ``narrator.generate`` is replaced with a bomb for every test, so any
future edit that makes the request path call inference fails loudly instead of
quietly coupling a public endpoint to a desktop GPU.
"""
import datetime
import pytest
from fastapi.testclient import TestClient
from data import grid
from data import narrator
from data import store
from web import app as appmod
Q = {"lat": 47.6062, "lon": -122.3321}
TODAY = datetime.date.today()
@pytest.fixture
def client(monkeypatch):
def _boom(*a, **kw): # pragma: no cover - failing here is the point
raise AssertionError("the request path must never call the LLM")
monkeypatch.setattr(narrator, "generate", _boom)
return TestClient(appmod.app)
@pytest.fixture
def seeded():
"""Store one narrative for today's cell and hand back its identity."""
cell = grid.snap(Q["lat"], Q["lon"])
key = narrator.narrative_key(TODAY)
token = narrator.narrative_token()
store.put_narrative(cell["id"], key, token,
"A cool night for late July here.", "qwen3:14b")
yield {"cell": cell, "key": key, "token": token}
def test_missing_narrative_is_pending_not_an_error(client):
r = client.get("/thermograph/api/v2/narrative",
params={"lat": -33.8688, "lon": 151.2093})
assert r.status_code == 200
body = r.json()
assert body["narrative"] is None
assert body["status"] == "pending"
assert body["model"] is None
def test_stored_narrative_is_served_with_provenance(client, seeded):
r = client.get("/thermograph/api/v2/narrative", params=Q)
assert r.status_code == 200
body = r.json()
assert body["narrative"] == "A cool night for late July here."
assert body["status"] == "ready"
assert body["model"] == "qwen3:14b"
assert body["generated_at"] > 0
assert body["date"] == TODAY.isoformat()
assert body["cell"]["id"] == seeded["cell"]["id"]
def test_conditional_revalidation_returns_304(client, seeded):
first = client.get("/thermograph/api/v2/narrative", params=Q)
etag = first.headers["ETag"]
again = client.get("/thermograph/api/v2/narrative", params=Q,
headers={"If-None-Match": etag})
assert again.status_code == 304
assert again.content == b""
def test_the_pending_etag_changes_once_the_row_lands(client, seeded):
"""A client that cached a pending response has to notice the sentence
appearing — so the two states must not share an ETag."""
ready = client.get("/thermograph/api/v2/narrative", params=Q).headers["ETag"]
pending = client.get("/thermograph/api/v2/narrative",
params={"lat": -33.8688, "lon": 151.2093}).headers["ETag"]
assert ready != pending
def test_a_row_written_under_another_token_is_not_served(client):
"""A prompt or model change must orphan old text rather than serve it."""
cell = grid.snap(35.6762, 139.6503)
store.put_narrative(cell["id"], narrator.narrative_key(TODAY),
"n0:some-old-model", "Written under the old contract.",
"some-old-model")
r = client.get("/thermograph/api/v2/narrative",
params={"lat": 35.6762, "lon": 139.6503})
assert r.status_code == 200
assert r.json()["narrative"] is None
assert r.json()["status"] == "pending"
def test_an_explicit_past_date_is_its_own_row(client, seeded):
r = client.get("/thermograph/api/v2/narrative",
params={**Q, "date": "2020-01-01"})
assert r.status_code == 200
assert r.json()["date"] == "2020-01-01"
assert r.json()["narrative"] is None
def test_an_unparseable_date_is_rejected(client):
r = client.get("/thermograph/api/v2/narrative", params={**Q, "date": "last-tuesday"})
assert r.status_code == 400
@pytest.mark.parametrize("bad", [{"lat": 91, "lon": 0}, {"lat": 0, "lon": 181}])
def test_out_of_range_coordinates_are_rejected(client, bad):
assert client.get("/thermograph/api/v2/narrative", params=bad).status_code == 422

269
backend/tests/data/test_narrator.py Normal file
View file

@ -0,0 +1,269 @@
"""Narrative generation: fact extraction, prompt building, and output sanitizing.
Hermetic — no Ollama is contacted. ``generate`` is exercised through an injected
stub client, and every other function here is pure.
"""
import datetime
import pytest
from data import narrator
TARGET = datetime.date(2026, 7, 25)
def _payload(**over):
"""A /grade payload trimmed to what facts_from_grade reads."""
day = {
"date": "2026-07-25",
"tmax": {"value": 72.2, "percentile": 47.8, "grade": "Normal", "class": "normal"},
"tmin": {"value": 54.3, "percentile": 20.2, "grade": "Low", "class": "cold"},
"feels": {"value": 53.1, "percentile": 27.9, "grade": "Below Normal", "class": "cool"},
"precip": {"value": 0.0, "percentile": None, "grade": "Dry", "class": "dry"},
}
day.update(over.pop("day", {}))
base = {
"place": "Ringaudai, Lithuania",
"recent": [
{"date": "2026-07-24", "tmax": {"value": 60.9, "percentile": 4.0,
"grade": "Very Low", "class": "very-cold"}},
day,
],
}
base.update(over)
return base
class _Resp:
def __init__(self, payload):
self._payload = payload
def raise_for_status(self):
return None
def json(self):
return self._payload
class _Client:
"""Stand-in for httpx.Client that records the request and replays a response."""
def __init__(self, response=None, raises=None):
self.response = response
self.raises = raises
self.calls = []
def post(self, url, json=None, timeout=None):
self.calls.append({"url": url, "json": json})
if self.raises is not None:
raise self.raises
return _Resp(self.response)
# --- fact extraction ---------------------------------------------------------
def test_facts_pick_the_target_day_not_a_neighbour():
facts = narrator.facts_from_grade(_payload(), TARGET)
assert facts["date"] == "2026-07-25"
assert facts["place"] == "Ringaudai, Lithuania"
grades = {m["label"]: m["grade"] for m in facts["metrics"]}
# "Very Low" belongs to the 24th; picking it up would mean the wrong row.
assert grades == {"Daytime high": "Normal", "Overnight low": "Low",
"Feels-like": "Below Normal", "Rain": "Dry"}
def test_facts_never_carry_a_raw_measurement():
"""The safety property the whole design rests on: a model that is never given
a temperature cannot misreport one."""
facts = narrator.facts_from_grade(_payload(), TARGET)
blob = repr(facts) + narrator.build_prompt(facts)
for reading in ("72.2", "54.3", "53.1", "60.9"):
assert reading not in blob
def test_facts_omit_metrics_the_day_does_not_have():
facts = narrator.facts_from_grade(
_payload(day={"tmin": None, "feels": None, "precip": None}), TARGET)
assert [m["label"] for m in facts["metrics"]] == ["Daytime high"]
def test_facts_are_none_when_nothing_is_gradeable():
empty = _payload(day={"tmax": None, "tmin": None, "feels": None, "precip": None})
assert narrator.facts_from_grade(empty, TARGET) is None
def test_facts_are_none_when_the_day_is_absent():
assert narrator.facts_from_grade(_payload(), datetime.date(2026, 8, 9)) is None
def test_facts_accept_a_dict_place_label():
facts = narrator.facts_from_grade(_payload(place={"label": "Kaunas, Lithuania"}), TARGET)
assert facts["place"] == "Kaunas, Lithuania"
def test_an_explicit_place_overrides_the_reverse_geocoded_one():
"""A cell in a dense city reverse-geocodes to a neighbourhood; a caller that
knows it asked about Shanghai should not get a sentence about a subdistrict."""
facts = narrator.facts_from_grade(
_payload(place="Nanjingdonglu Subdistrict"), TARGET, place="Shanghai, China")
assert facts["place"] == "Shanghai, China"
assert "Nanjingdonglu" not in narrator.build_prompt(facts)
def test_dry_day_states_the_tier_without_a_percentage():
"""Rain is ranked among rain days only, so a dry day has a tier and no
percentile — it must not render as a percentage that doesn't exist."""
facts = narrator.facts_from_grade(_payload(), TARGET)
rain = next(m for m in facts["metrics"] if m["label"] == "Rain")
assert rain["pct"] is None
prompt = narrator.build_prompt(facts)
assert "Rain: dry — no measurable rain" in prompt
assert "None" not in prompt
assert "%" not in prompt.split("Rain:")[1]
# --- prompt ------------------------------------------------------------------
def test_prompt_is_deterministic_and_carries_every_fact():
facts = narrator.facts_from_grade(_payload(), TARGET)
a, b = narrator.build_prompt(facts), narrator.build_prompt(facts)
assert a == b
assert "Place: Ringaudai, Lithuania" in a
assert "Date: 25 July" in a
assert ("Daytime high: warmer than 48% of days at this time of year here "
"(tier: Normal)") in a
assert ("Overnight low: warmer than 20% of nights at this time of year here "
"(tier: Low)") in a
def test_prompt_states_the_direction_so_a_tier_cannot_be_read_backwards():
"""The regression this guards: "Overnight low: 82nd percentile (High)" was
phrased by the model as a *cool* night — an inversion no output filter can
detect. The percentage must carry the direction itself."""
warm_night = _payload(day={"tmin": {"value": 78.0, "percentile": 82.4,
"grade": "High", "class": "hot"}})
prompt = narrator.build_prompt(narrator.facts_from_grade(warm_night, TARGET))
assert "Overnight low: warmer than 82% of nights" in prompt
def test_rain_is_described_as_heavier_not_warmer():
wet = _payload(day={"precip": {"value": 12.0, "percentile": 73.0,
"grade": "Heavy", "class": "wet-6"}})
prompt = narrator.build_prompt(narrator.facts_from_grade(wet, TARGET))
assert "Rain: heavier than 73% of rainy days" in prompt
@pytest.mark.parametrize("pct, shown", [(47.8, 48), (20.2, 20), (0.1, 1), (99.9, 99)])
def test_prompt_percentages_match_the_ui_rounding_and_clamp(pct, shown):
"""Mirrors grading.pct_ordinal — a narrative disagreeing with the number
printed next to it is the kind of thing users report as a bug."""
day = _payload(day={"tmax": {"value": 1.0, "percentile": pct,
"grade": "Normal", "class": "normal"}})
prompt = narrator.build_prompt(narrator.facts_from_grade(day, TARGET))
assert f"Daytime high: warmer than {shown}% of days" in prompt
def test_prompt_omits_the_place_line_when_unknown():
facts = narrator.facts_from_grade(_payload(place=None), TARGET)
assert "Place:" not in narrator.build_prompt(facts)
# --- sanitizing --------------------------------------------------------------
@pytest.mark.parametrize("raw, expected", [
(' "A cool night for late July here." ',
"A cool night for late July here."),
("Sure! Here's a sentence: A cool night for late July in this corner of Lithuania.",
"A cool night for late July in this corner of Lithuania."),
("<think>weighing the tiers</think> A cool night for late July here.",
"A cool night for late July here."),
("A cool night for\nlate July here.",
"A cool night for late July here."),
("Okay, here is the summary: A cool night for late July here.",
"A cool night for late July here."),
])
def test_sanitize_normalizes_common_model_habits(raw, expected):
assert narrator._sanitize(raw) == expected
def test_sanitize_keeps_a_colon_that_belongs_to_the_sentence():
"""Preamble stripping is anchored to known openers on purpose — "everything
before the first colon" would eat half of a perfectly good narrative."""
raw = "One thing stands out: the night was colder than four in five late-July nights."
assert narrator._sanitize(raw) == raw
@pytest.mark.parametrize("raw", [
"",
" ",
"Sure!", # too short to be a sentence
"The high reached 31°C, well above normal for July.", # a reading it was never given
"Rainfall hit 12 mm, far more than usual for the season.", # ditto
"Tomorrow will be even more unusual for this location.", # forecast
"Expect an unusually cold night for late July in this area.", # forecast
"A cold night by local standards, so wear an extra layer tonight.", # advice
])
def test_sanitize_rejects_output_it_should_not_store(raw):
assert narrator._sanitize(raw) is None
def test_sanitize_truncates_a_long_answer_at_a_sentence_boundary():
first = "A distinctly cool night for late July in this part of Lithuania. "
text = narrator._sanitize(first + "B" * 400)
assert text == first.strip()
assert len(text) <= narrator.MAX_CHARS
def test_sanitize_rejects_a_long_answer_with_nowhere_to_cut():
assert narrator._sanitize("A" * 400) is None
# --- generate ----------------------------------------------------------------
def test_generate_returns_the_sanitized_sentence():
facts = narrator.facts_from_grade(_payload(), TARGET)
client = _Client(response={"response": ' "A cool night for late July here." '})
assert narrator.generate(facts, client=client) == "A cool night for late July here."
def test_generate_sends_a_bounded_non_thinking_request():
"""Reasoning tokens dominate cost for a one-sentence rewrite, and an unbounded
generation on a CPU fallback is how a batch job stops finishing."""
facts = narrator.facts_from_grade(_payload(), TARGET)
client = _Client(response={"response": "A cool night for late July here."})
narrator.generate(facts, client=client, model="test-model")
sent = client.calls[0]["json"]
assert sent["model"] == "test-model"
assert sent["stream"] is False
assert sent["think"] is False
assert sent["options"]["num_predict"] <= 128
assert client.calls[0]["url"].endswith("/api/generate")
def test_generate_is_none_when_the_model_is_unreachable():
facts = narrator.facts_from_grade(_payload(), TARGET)
client = _Client(raises=OSError("connection refused"))
assert narrator.generate(facts, client=client) is None
def test_generate_is_none_when_output_fails_the_sanitizer():
facts = narrator.facts_from_grade(_payload(), TARGET)
client = _Client(response={"response": "The high hit 31°C today."})
assert narrator.generate(facts, client=client) is None
def test_generate_is_none_on_an_empty_response():
facts = narrator.facts_from_grade(_payload(), TARGET)
assert narrator.generate(facts, client=_Client(response={})) is None
# --- token -------------------------------------------------------------------
def test_token_tracks_the_version_and_the_model():
base = narrator.narrative_token("qwen3:14b")
assert base != narrator.narrative_token("llama3:8b")
assert base.startswith(narrator.NARRATIVE_VER)
def test_key_is_the_iso_date():
assert narrator.narrative_key(TARGET) == "2026-07-25"

5
backend/web/app.py
View file

@ -20,6 +20,7 @@ from fastapi.responses import JSONResponse, RedirectResponse
from accounts import api_accounts
from api import content_routes
from api import internal_routes
from api import narrative_routes
from core import audit
from data import climate
from notifications import digest
@ -897,6 +898,10 @@ app.include_router(v2, prefix=f"{BASE}/api/v2")
# service (Stage 3) instead of anything in this process. See
# backend/api/content_routes.py.
app.include_router(content_routes.router, prefix=f"{BASE}/api/v2")
# Precomputed plain-English day summaries. Read-only: this router never calls the
# LLM, so mounting it adds no runtime dependency on the inference host — see
# backend/api/narrative_routes.py.
app.include_router(narrative_routes.router, prefix=f"{BASE}/api/v2")
# Internal control surface for the thermograph-daemon Go binary (the gateway
# bot + recurring jobs that used to run in _lifespan). Deliberately NOT under
# BASE — the daemon hits THERMOGRAPH_API_BASE_INTERNAL directly, same posture