Ship the data-pipeline chain: local geocoding, Meteostat gusts, NASA POWER primary, ERA5 seed #10

Closed
admin_emi wants to merge 9 commits from ship-data-pipeline into dev
6 changed files with 138 additions and 49 deletions
Showing only changes of commit 6eff3ba3a7 - Show all commits

View file

@ -44,7 +44,9 @@ _PHASE_SOURCE = {
"history_fetch": "open-meteo-archive", "history_fetch": "open-meteo-archive",
"history_topup": "open-meteo-archive", "history_topup": "open-meteo-archive",
"recent_forecast_fetch": "open-meteo-forecast", "recent_forecast_fetch": "open-meteo-forecast",
"geocode": "open-meteo-geocoding", # Forward geocoding moved off Open-Meteo to Nominatim (same host as reverse) —
# /geocode now backstops the local GeoNames index on a miss.
"geocode": "nominatim",
"history_nasa": "nasa-power", "history_nasa": "nasa-power",
"forecast_metno": "met-norway", "forecast_metno": "met-norway",
"reverse_geocode": "nominatim", "reverse_geocode": "nominatim",

View file

@ -1058,24 +1058,54 @@ def reverse_geocode(lat: float, lon: float) -> str | None:
return None return None
def geocode(name: str, count: int = 5) -> list[dict]: def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
"""Look up places by name worldwide via Open-Meteo's geocoder.""" """Forward place-name lookup via OpenStreetMap Nominatim (keyless).
r = _request(
"https://geocoding-api.open-meteo.com/v1/search", Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the
{"name": name, "count": count, "language": "en", "format": "json"}, local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population
30, villages, and alternate/native-language spellings so /geocode falls back to
phase="geocode", it on a local miss. It carries no population, so results keep Nominatim's own
) relevance order; name/admin/country map straight across.
results = r.json().get("results", []) or []
return [ Shares the reverse geocoder's lock and ~1/sec pacing: both hit the same host,
{ so serializing them together keeps total Nominatim traffic under the usage
"name": g.get("name"), policy. Only the low-volume /geocode miss path reaches here autocomplete
"admin1": g.get("admin1"), (/suggest) is served purely from the local index and never calls out.
"country": g.get("country"), """
"country_code": g.get("country_code"), global _revgeo_last
"lat": g.get("latitude"), with _REVGEO_LOCK:
"lon": g.get("longitude"), wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
"population": g.get("population"), if wait > 0:
} time.sleep(wait)
for g in results try:
] r = _request(
"https://nominatim.openstreetmap.org/search",
{"q": name, "format": "jsonv2", "addressdetails": 1,
"limit": count, "accept-language": "en"},
30,
phase="geocode",
headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
)
rows = r.json() or []
finally:
_revgeo_last = time.monotonic()
out = []
for g in rows:
a = g.get("address", {}) or {}
# jsonv2 gives a `name` for named features; for addresses/areas fall back to
# the most specific place component, then the head of display_name.
label = (g.get("name")
or a.get("city") or a.get("town") or a.get("village")
or a.get("hamlet") or a.get("suburb") or a.get("neighbourhood")
or (g.get("display_name") or "").split(",")[0].strip() or None)
cc = (a.get("country_code") or "").upper() or None
out.append({
"name": label,
"admin1": a.get("state") or a.get("province") or a.get("region"),
"country": a.get("country"),
"country_code": cc,
"lat": float(g["lat"]) if g.get("lat") is not None else None,
"lon": float(g["lon"]) if g.get("lon") is not None else None,
"population": None, # Nominatim has no population; order is relevance-based
})
return out

View file

@ -1,22 +1,25 @@
"""Typo-tolerant place-name index for search suggestions. """Typo-tolerant place-name index for search suggestions.
Open-Meteo's geocoder (climate.geocode) only matches exact spellings — one This module keeps a local index of world places (a GeoNames cities dump,
mistyped letter and the query returns nothing. This module keeps a local index downloaded once into data/geonames/ and reused across restarts) so /suggest can
of world places (a GeoNames cities dump, downloaded once into data/geonames/ answer instantly, offline, and tolerate a single-letter typo a substituted,
and reused across restarts) so /suggest can answer instantly and tolerate a missing, or extra letter, or two adjacent letters swapped anywhere in the
single-letter typo a substituted, missing, or extra letter, or two adjacent query, including the first character.
letters swapped anywhere in the query, including the first character.
It also builds a vocabulary of place-name tokens so a typo'd word in a It also builds a vocabulary of place-name tokens so a typo'd word in a
multi-word query can be respelled against words the index knows ("pest multi-word query can be respelled against words the index knows ("pest
seattle""west seattle"); the endpoint verifies those candidates against the seattle""west seattle").
upstream geocoder, which covers neighbourhood-level places the cities dump
lacks. `suggest()` still takes an injected ``upstream`` callable so the module stays
free of the fetch layer, but /suggest now passes a no-op: autocomplete is served
purely from this index and makes no per-keystroke network call. (/geocode, the
explicit search, is the path that falls back to Nominatim on a local miss for
neighbourhoods/postcodes/tiny places the cities dump lacks see web/app.py.)
The index loads in a background thread (start_loading()). Until it's ready — The index loads in a background thread (start_loading()). Until it's ready —
or forever, if the download fails search()/corrections() return None/empty or forever, if the download fails search()/corrections() return None/empty
and /suggest degrades to the plain upstream geocoder, so the app never needs and /suggest simply returns no results, so the app never needs this data to
this data to boot or serve. boot or serve.
""" """
import bisect import bisect
import heapq import heapq

View file

@ -14,7 +14,7 @@ def test_phase_source_map_covers_every_source():
assert m.source_for_phase("history_fetch") == "open-meteo-archive" assert m.source_for_phase("history_fetch") == "open-meteo-archive"
assert m.source_for_phase("history_topup") == "open-meteo-archive" assert m.source_for_phase("history_topup") == "open-meteo-archive"
assert m.source_for_phase("recent_forecast_fetch") == "open-meteo-forecast" assert m.source_for_phase("recent_forecast_fetch") == "open-meteo-forecast"
assert m.source_for_phase("geocode") == "open-meteo-geocoding" assert m.source_for_phase("geocode") == "nominatim" # forward geocoding moved off Open-Meteo
assert m.source_for_phase("history_nasa") == "nasa-power" assert m.source_for_phase("history_nasa") == "nasa-power"
assert m.source_for_phase("forecast_metno") == "met-norway" assert m.source_for_phase("forecast_metno") == "met-norway"
assert m.source_for_phase("reverse_geocode") == "nominatim" assert m.source_for_phase("reverse_geocode") == "nominatim"

View file

@ -6,6 +6,7 @@ from fastapi.testclient import TestClient
from web import app as appmod from web import app as appmod
from data import climate from data import climate
from data import places
@pytest.fixture @pytest.fixture
@ -182,11 +183,53 @@ def test_cell_bundle_matches_per_view_payloads(client):
assert r304.status_code == 304 assert r304.status_code == 304
def test_suggest_falls_back_to_upstream_geocoder(client, monkeypatch): _LOCAL_HIT = [{"name": "Seattle", "admin1": "Washington", "country": "United States",
upstream = [{"name": "Seattle", "admin1": "Washington", "country": "United States", "country_code": "US", "lat": 47.6, "lon": -122.33, "population": 737015,
"country_code": "US", "lat": 47.6, "lon": -122.33, "population": 737015}] "match": "prefix"}]
monkeypatch.setattr(climate, "geocode", lambda q, count=5: list(upstream))
r = client.get("/thermograph/api/v2/suggest", params={"q": "seattle-fallback-probe"})
def test_geocode_local_first_skips_network(client, monkeypatch):
"""A local GeoNames hit answers /geocode with no outbound Nominatim call, and
the internal prefix/fuzzy tag is stripped from the wire shape."""
monkeypatch.setattr(places, "search", lambda q, limit=5: list(_LOCAL_HIT))
def _boom(*a, **k):
raise AssertionError("Nominatim must not be called when the local index hits")
monkeypatch.setattr(climate, "geocode_nominatim", _boom)
r = client.get("/thermograph/api/v2/geocode", params={"q": "seattle"})
assert r.status_code == 200
res = r.json()["results"]
assert res[0]["name"] == "Seattle"
assert "match" not in res[0]
def test_geocode_falls_back_to_nominatim_on_local_miss(client, monkeypatch):
"""A local miss (or a still-loading index) falls back to Nominatim /search —
the path that covers neighbourhoods/postcodes the cities dump lacks."""
monkeypatch.setattr(places, "search", lambda q, limit=5: []) # genuine miss
nom = [{"name": "West Seattle", "admin1": "Washington", "country": "United States",
"country_code": "US", "lat": 47.57, "lon": -122.38, "population": None}]
called = []
monkeypatch.setattr(climate, "geocode_nominatim",
lambda q, count=5: (called.append(q), nom)[1])
r = client.get("/thermograph/api/v2/geocode", params={"q": "west seattle"})
assert r.status_code == 200
assert r.json()["results"][0]["name"] == "West Seattle"
assert called == ["west seattle"]
def test_suggest_is_local_only_no_network(client, monkeypatch):
"""Autocomplete is served purely from the local index — even a reachable
Nominatim must never be hit per keystroke."""
monkeypatch.setattr(places, "search", lambda q, limit=5: list(_LOCAL_HIT))
def _boom(*a, **k):
raise AssertionError("/suggest must not make an outbound geocoder call")
monkeypatch.setattr(climate, "geocode_nominatim", _boom)
r = client.get("/thermograph/api/v2/suggest", params={"q": "seattle"})
assert r.status_code == 200 assert r.status_code == 200
body = r.json() body = r.json()
assert body["results"][0]["name"] == "Seattle" assert body["results"][0]["name"] == "Seattle"

View file

@ -2,7 +2,6 @@
import asyncio import asyncio
import contextlib import contextlib
import datetime import datetime
import functools
import hashlib import hashlib
import hmac import hmac
import json import json
@ -414,9 +413,22 @@ def _cached_response(request, run, kind, cell, key, token, build, cache_placeles
# --- endpoints ---------------------------------------------------------------- # --- endpoints ----------------------------------------------------------------
_GEOCODE_LIMIT = 5
def api_geocode(q: str = Query(..., min_length=1)): def api_geocode(q: str = Query(..., min_length=1)):
"""Forward place search. The local GeoNames index answers first (no network);
only a genuine miss or a query the index can't hold (neighbourhoods,
postcodes, tiny villages, native-language names) falls back to Nominatim.
`search` returns None while the index is still loading, which also falls
through to the network so search keeps working during warmup."""
local = places.search(q, _GEOCODE_LIMIT)
if local:
# Strip the internal prefix/fuzzy tag so the wire shape matches the
# Nominatim branch (and the former Open-Meteo contract) exactly.
return {"results": [{k: v for k, v in r.items() if k != "match"} for r in local]}
try: try:
return {"results": climate.geocode(q)} return {"results": climate.geocode_nominatim(q, _GEOCODE_LIMIT)}
except Exception as e: # noqa: BLE001 - surface upstream failures to the client except Exception as e: # noqa: BLE001 - surface upstream failures to the client
raise HTTPException(status_code=502, detail=f"geocoding failed: {e}") raise HTTPException(status_code=502, detail=f"geocoding failed: {e}")
@ -424,21 +436,20 @@ def api_geocode(q: str = Query(..., min_length=1)):
_SUGGEST_LIMIT = 5 _SUGGEST_LIMIT = 5
@functools.lru_cache(maxsize=1024) def _no_upstream(q: str) -> tuple:
def _suggest_upstream(q: str) -> tuple: """Autocomplete is served purely from the local GeoNames index — no per-keystroke
"""Open-Meteo lookup for /suggest, memoized per query string — type-ahead call to any external geocoder. Typo-correction still runs against the local
re-asks the same prefixes constantly (backspacing, retyping). Failures index, so respellings ("pest seattle" "west seattle") keep working offline."""
raise and are not cached, so a transient upstream error doesn't stick.""" return ()
return tuple(climate.geocode(q, count=_SUGGEST_LIMIT))
def api_suggest(q: str = Query(..., min_length=1)): def api_suggest(q: str = Query(..., min_length=1)):
"""Type-ahead location suggestions: the top 5 places for a (possibly """Type-ahead location suggestions: the top 5 places for a (possibly
typo'd) query prefix; `corrected` reports the respelling that produced the typo'd) query prefix; `corrected` reports the respelling that produced the
results ("pest seattle" "west seattle"), or null. The ranking/typo policy results ("pest seattle" "west seattle"), or null. The ranking/typo policy
lives in places.suggest.""" lives in places.suggest, now driven entirely by the local index."""
try: try:
results, corrected = places.suggest(q, _suggest_upstream, _SUGGEST_LIMIT) results, corrected = places.suggest(q, _no_upstream, _SUGGEST_LIMIT)
except Exception as e: # noqa: BLE001 - nothing at all to serve except Exception as e: # noqa: BLE001 - nothing at all to serve
raise HTTPException(status_code=502, detail=f"geocoding failed: {e}") raise HTTPException(status_code=502, detail=f"geocoding failed: {e}")
return {"results": results, "corrected": corrected} return {"results": results, "corrected": corrected}