feat(weather-recon): Phase 1 — curated stations + Directus weather schema (#184)

packages/tools/weather-recon/.gitignore

@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+*.egg-info/
+data/isd-history.csv

packages/tools/weather-recon/Dockerfile

@@ -0,0 +1,17 @@
+# One image serves both roles: the Prefect worker and the flow-run Jobs it spawns.
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+# Deps first (stable layer), pinned to the in-cluster Prefect server version.
+RUN pip install "prefect==3.7.4" "prefect-kubernetes>=0.5" "httpx>=0.27"
+
+COPY pyproject.toml ./
+COPY weather_recon ./weather_recon
+COPY data/stations.csv ./data/stations.csv
+RUN pip install --no-deps .
+
+# Flow-run jobs override this; the worker Deployment uses it as-is.
+CMD ["prefect", "worker", "start", "--pool", "weather-recon-k8s", "--type", "kubernetes"]

packages/tools/weather-recon/README.md

@@ -0,0 +1,37 @@
+# weather-recon
+
+Loads September 2001 weather data into Directus for the rt911 Weather app
+(issue #184; design spec: `plans/weather-app-design.md`). Phase 1 ships the
+curated station reference table and the three collections; observation /
+radar / forecast / almanac flows follow in Phase 2.
+
+## Layout
+
+- `scripts/build_stations.py` — dev-run: NOAA `isd-history.csv` + curated ICAO
+  list → `data/stations.csv` (committed). Re-run only to change the curation.
+- `weather_recon/flow.py` — `load-weather-stations`: validates the CSV, ensures
+  the `weather_stations` / `weather_observations` / `weather_forecasts`
+  collections, reloads `weather_stations` (delete-all + insert; ICAO-keyed).
+- `weather_recon/directus.py` — REST client + collection specs (ported from
+  flight-recon; same 1 MB payload batching and cast-json rules).
+
+## Running locally (no work pool)
+
+    pip install -e ".[dev]"
+    export DIRECTUS_URL=...            # e.g. port-forwarded rt911-api
+    export DIRECTUS_API_TOKEN=...      # from the rt911 namespace secret
+    python -m weather_recon.flow data/stations.csv
+
+## Cluster deployment (mirrors flight-recon; see its k8s/ for templates)
+
+1. Build + import the image into k3s containerd:
+   `docker build -t weather-recon:0.1.0 . && docker save weather-recon:0.1.0 | sudo k3s ctr images import -`
+2. Infra repo (GitOps): `weather-recon-k8s` work pool, worker Deployment
+   running this image, and a `weather-recon-secrets` Secret providing
+   `DIRECTUS_API_TOKEN` — copy the flight-recon manifests.
+3. Register the deployment: see `weather_recon/deploy.py` docstring.
+
+## Tests
+
+    pytest tests/ -v
+    ruff check weather_recon/ tests/ scripts/

packages/tools/weather-recon/pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "weather-recon"
+version = "0.1.0"
+description = "Load Sept 2001 weather stations/observations/forecasts into Directus (Prefect flows)"
+requires-python = ">=3.12"
+dependencies = [
+    "prefect>=3.7,<4",
+    "prefect-kubernetes>=0.5",
+    "httpx>=0.27",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8", "ruff>=0.4", "timezonefinder>=6"]
+
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["weather_recon"]
+
+[tool.ruff]
+line-length = 100

packages/tools/weather-recon/scripts/build_stations.py

@@ -0,0 +1,129 @@
+"""
+Build data/stations.csv from NOAA's isd-history.csv and the curated ICAO list.
+
+Dev-run only (output is committed):
+
+    python scripts/build_stations.py            # uses cached data/isd-history.csv if present
+    python scripts/build_stations.py --refresh  # re-download the master list
+
+Reports any curated ICAO with no ISD row covering 2001-09-09..12 — trim or
+substitute those and re-run until the report is empty.
+"""
+
+import argparse
+import csv
+import sys
+from pathlib import Path
+
+ISD_HISTORY_URL = "https://www.ncei.noaa.gov/pub/data/noaa/isd-history.csv"
+ROOT = Path(__file__).resolve().parents[1]
+CACHE = ROOT / "data" / "isd-history.csv"
+OUT = ROOT / "data" / "stations.csv"
+FIELDS = ["station_id", "name", "lat", "lon", "elevation_m", "country", "tz", "isd_id"]
+
+# Curated station list: major-city METAR sites across the US, Canada and
+# Mexico with 2001 coverage. 188 sites; every US state, 17 CA / 18 MX.
+CURATED_ICAOS = {
+    # US — Northeast
+    "KBOS", "KPWM", "KBGR", "KBTV", "KMHT", "KPVD", "KBDL", "KALB", "KSYR",
+    "KROC", "KBUF", "KJFK", "KLGA", "KEWR", "KPHL", "KABE", "KAVP", "KCXY",
+    "KPIT", "KDCA", "KIAD", "KBWI",
+    # US — Southeast
+    "KRIC", "KORF", "KRDU", "KGSO", "KCLT", "KCAE", "KCHS", "KSAV", "KATL",
+    "KTLH", "KJAX", "KMCO", "KTPA", "KMIA", "KPBI", "KEYW",
+    "KFMY",  # KRSW substitute: Fort Myers Page Field (KRSW has no 2001-09 rows)
+    "KBHM", "KMGM", "KMOB", "KHSV", "KBNA", "KMEM", "KTYS", "KCHA", "KSDF",
+    "KLEX", "KCRW", "KJAN",
+    # US — Midwest
+    "KCLE", "KCMH", "KCVG", "KDAY", "KTOL", "KIND", "KFWA", "KSBN", "KDTW",
+    "KGRR", "KLAN", "KORD", "KMDW", "KRFD", "KPIA", "KSPI", "KMLI", "KMKE",
+    "KMSN", "KGRB", "KMSP", "KDLH", "KRST", "KFAR", "KBIS", "KFSD", "KRAP",
+    "KDSM", "KCID", "KOMA", "KLNK", "KICT", "KSTL", "KMCI", "KSGF", "KCOU",
+    # US — South-central
+    "KOKC", "KTUL", "KLIT", "KFSM", "KDFW", "KDAL", "KIAH", "KAUS",
+    "KSAT", "KELP", "KLBB", "KAMA", "KMAF", "KCRP", "KBRO", "KSHV",
+    "KMSY", "KLCH",
+    # US — Mountain
+    "KDEN", "KCOS", "KPUB", "KGJT", "KCYS", "KCPR", "KBIL", "KGTF", "KBZN",
+    "KMSO", "KHLN", "KBOI", "KIDA", "KPIH", "KSLC", "KLAS", "KRNO", "KELY",
+    "KPHX", "KTUS", "KFLG", "KYUM", "KABQ", "KROW",
+    # US — Pacific
+    "KSEA", "KGEG", "KYKM", "KPDX", "KEUG", "KMFR", "KSFO", "KOAK", "KSJC",
+    "KSMF", "KFAT", "KBFL", "KLAX", "KSAN", "KSBA", "KMRY", "KRDD",
+    # US — Alaska & Hawaii
+    "PANC", "PAFA", "PAJN", "PHNL", "PHOG", "PHTO", "PHLI",
+    # Canada
+    "CYVR", "CYYJ", "CYYC", "CYEG", "CYXE", "CYQR", "CYWG", "CYQT", "CYYZ",
+    "CYOW", "CYUL", "CWQB", "CYHZ", "CYSJ", "CYYT", "CYXY", "CYZF",
+    # Mexico
+    "MMMX", "MMGL", "MMMY", "MMTJ", "MMHO", "MMCS", "MMCU", "MMMZ", "MMSD",
+    "MMPR", "MMLO", "MMAA", "MMVR", "MMMD", "MMUN", "MMOX", "MMVA", "MMTM",
+}
+
+
+def pick_station_rows(rows, icaos, start="20010909", end="20010912"):
+    """Per ICAO in `icaos`, the isd-history row covering [start, end] with the
+    latest END. ICAOs with no covering row are simply absent from the result."""
+    picked = {}
+    for row in rows:
+        icao = (row.get("ICAO") or "").strip()
+        if icao not in icaos:
+            continue
+        if not (row["BEGIN"] <= start and row["END"] >= end):
+            continue
+        if icao not in picked or row["END"] > picked[icao]["END"]:
+            picked[icao] = row
+    return picked
+
+
+def station_record(row, tz):
+    """One stations.csv record from an isd-history row."""
+    elev = row.get("ELEV(M)", "").strip()
+    return {
+        "station_id": row["ICAO"].strip(),
+        "name": row["STATION NAME"].strip(),
+        "lat": float(row["LAT"]),
+        "lon": float(row["LON"]),
+        "elevation_m": float(elev) if elev else None,
+        "country": row["CTRY"].strip(),
+        "tz": tz,
+        "isd_id": f'{row["USAF"]}-{row["WBAN"]}',
+    }
+
+
+def main(refresh=False):
+    import httpx
+    from timezonefinder import TimezoneFinder
+
+    if refresh or not CACHE.is_file():
+        print(f"downloading {ISD_HISTORY_URL} ...")
+        CACHE.parent.mkdir(parents=True, exist_ok=True)
+        CACHE.write_bytes(httpx.get(ISD_HISTORY_URL, timeout=120,
+                                    follow_redirects=True).raise_for_status().content)
+    with CACHE.open(newline="", encoding="utf-8") as f:
+        picked = pick_station_rows(csv.DictReader(f), CURATED_ICAOS)
+
+    missing = sorted(CURATED_ICAOS - set(picked))
+    tf = TimezoneFinder()
+    records = []
+    for icao in sorted(picked):
+        row = picked[icao]
+        tz = tf.timezone_at(lng=float(row["LON"]), lat=float(row["LAT"]))
+        records.append(station_record(row, tz))
+
+    with OUT.open("w", newline="", encoding="utf-8") as f:
+        w = csv.DictWriter(f, fieldnames=FIELDS)
+        w.writeheader()
+        w.writerows(records)
+    print(f"wrote {len(records)} stations to {OUT}")
+    if missing:
+        print(f"WARNING: {len(missing)} curated ICAOs lack 2001-09 ISD coverage: "
+              f"{', '.join(missing)}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--refresh", action="store_true")
+    sys.exit(main(refresh=ap.parse_args().refresh))

packages/tools/weather-recon/tests/test_build_stations.py

@@ -0,0 +1,45 @@
+# tests/test_build_stations.py
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+from build_stations import pick_station_rows, station_record  # noqa: E402
+
+
+def _row(icao, begin, end, usaf="725300", wban="94846", lat="41.995",
+         lon="-87.934", elev="200.6", ctry="US", name="CHICAGO O'HARE"):
+    return {"USAF": usaf, "WBAN": wban, "STATION NAME": name, "CTRY": ctry,
+            "STATE": "IL", "ICAO": icao, "LAT": lat, "LON": lon,
+            "ELEV(M)": elev, "BEGIN": begin, "END": end}
+
+
+def test_pick_keeps_only_rows_covering_the_window():
+    rows = [_row("KORD", "19730101", "19991231"),      # ends before window
+            _row("KORD", "20000101", "20051231"),      # covers window
+            _row("KMSY", "20011001", "20051231")]      # begins after window
+    picked = pick_station_rows(rows, {"KORD", "KMSY"})
+    assert set(picked) == {"KORD"}
+    assert picked["KORD"]["BEGIN"] == "20000101"
+
+
+def test_pick_prefers_latest_end_when_multiple_rows_cover():
+    rows = [_row("KORD", "19730101", "20020101"),
+            _row("KORD", "20000101", "20251231")]
+    assert pick_station_rows(rows, {"KORD"})["KORD"]["END"] == "20251231"
+
+
+def test_pick_ignores_icaos_not_in_curated_set():
+    assert pick_station_rows([_row("KJFK", "19700101", "20251231")], {"KORD"}) == {}
+
+
+def test_station_record_shapes_and_types():
+    rec = station_record(_row("KORD", "20000101", "20251231"), tz="America/Chicago")
+    assert rec == {"station_id": "KORD", "name": "CHICAGO O'HARE",
+                   "lat": 41.995, "lon": -87.934, "elevation_m": 200.6,
+                   "country": "US", "tz": "America/Chicago", "isd_id": "725300-94846"}
+
+
+def test_station_record_missing_elevation_becomes_none():
+    rec = station_record(_row("KORD", "20000101", "20251231", elev=""),
+                         tz="America/Chicago")
+    assert rec["elevation_m"] is None

packages/tools/weather-recon/tests/test_schema_fields.py

@@ -0,0 +1,53 @@
+from weather_recon.directus import COLLECTIONS
+
+
+def test_expected_collections():
+    assert set(COLLECTIONS) == {"weather_stations", "weather_observations",
+                                "weather_forecasts"}
+
+
+def test_every_collection_has_exactly_one_primary_key():
+    for name, spec in COLLECTIONS.items():
+        pks = [f["field"] for f in spec["fields"]
+               if f.get("schema", {}).get("is_primary_key")]
+        assert len(pks) == 1, f"{name}: PKs {pks}"
+
+
+def test_no_duplicate_field_names():
+    for name, spec in COLLECTIONS.items():
+        fields = [f["field"] for f in spec["fields"]]
+        assert len(fields) == len(set(fields)), f"{name}"
+
+
+def test_json_fields_carry_cast_json_special():
+    # Repo law (flight_recon/directus.py:37): json fields without cast-json
+    # fail collection creation with an opaque 400.
+    for name, spec in COLLECTIONS.items():
+        for f in spec["fields"]:
+            if f["type"] == "json":
+                assert "cast-json" in f.get("meta", {}).get("special", []), \
+                    f"{name}.{f['field']}"
+
+
+def test_stations_pk_is_the_icao_string():
+    pk = next(f for f in COLLECTIONS["weather_stations"]["fields"]
+              if f.get("schema", {}).get("is_primary_key"))
+    assert pk["field"] == "station_id" and pk["type"] == "string"
+
+
+def test_delete_all_pk_filter_derivable_for_every_collection():
+    # delete_all filters on the collection PK; every collection must yield one
+    from weather_recon.directus import COLLECTIONS
+    for name, spec in COLLECTIONS.items():
+        pk = next(f["field"] for f in spec["fields"]
+                  if f.get("schema", {}).get("is_primary_key"))
+        assert pk
+
+
+def test_observation_and_forecast_time_fields_are_timestamps():
+    obs = {f["field"]: f for f in COLLECTIONS["weather_observations"]["fields"]}
+    fc = {f["field"]: f for f in COLLECTIONS["weather_forecasts"]["fields"]}
+    assert obs["observed_at"]["type"] == "timestamp"
+    assert obs["observed_at"]["schema"] == {"is_nullable": False}
+    assert fc["issued_at"]["type"] == "timestamp"
+    assert fc["issued_at"]["schema"] == {"is_nullable": False}

packages/tools/weather-recon/tests/test_stations.py

@@ -0,0 +1,52 @@
+import pytest
+
+from weather_recon.stations import load_stations
+
+HEADER = "station_id,name,lat,lon,elevation_m,country,tz,isd_id\n"
+GOOD = 'KORD,CHICAGO OHARE,41.995,-87.934,200.6,US,America/Chicago,725300-94846\n'
+
+
+def _write(tmp_path, body):
+    p = tmp_path / "stations.csv"
+    p.write_text(HEADER + body, encoding="utf-8")
+    return p
+
+
+def test_loads_and_types_a_valid_row(tmp_path):
+    rows = load_stations(_write(tmp_path, GOOD))
+    assert rows == [{"station_id": "KORD", "name": "CHICAGO OHARE", "lat": 41.995,
+                     "lon": -87.934, "elevation_m": 200.6, "country": "US",
+                     "tz": "America/Chicago", "isd_id": "725300-94846"}]
+
+
+def test_empty_elevation_is_none(tmp_path):
+    rows = load_stations(_write(
+        tmp_path, 'CYYZ,TORONTO PEARSON,43.68,-79.63,,CA,America/Toronto,712650-99999\n'))
+    assert rows[0]["elevation_m"] is None
+
+
+@pytest.mark.parametrize("bad,msg", [
+    ('KORD,X,91.0,-87.9,1,US,America/Chicago,1-2\n', "lat"),
+    ('KORD,X,41.9,-187.9,1,US,America/Chicago,1-2\n', "lon"),
+    ('KORD,X,41.9,-87.9,1,FR,Europe/Paris,1-2\n', "country"),
+    ('KORD,X,41.9,-87.9,1,US,,1-2\n', "tz"),
+    ('KORD,X,41.9,-87.9,1,US,America/Chicago,\n', "isd_id"),
+    (GOOD + GOOD, "duplicate"),
+])
+def test_rejects_invalid_rows(tmp_path, bad, msg):
+    with pytest.raises(ValueError, match=msg):
+        load_stations(_write(tmp_path, bad))
+
+
+def test_rejects_missing_columns(tmp_path):
+    p = tmp_path / "stations.csv"
+    p.write_text("station_id,name\nKORD,X\n", encoding="utf-8")
+    with pytest.raises(ValueError, match="column"):
+        load_stations(p)
+
+
+def test_committed_csv_is_valid():
+    from pathlib import Path
+    rows = load_stations(Path(__file__).resolve().parents[1] / "data" / "stations.csv")
+    assert len(rows) >= 150
+    assert {r["country"] for r in rows} == {"US", "CA", "MX"}