ERA5 lake: bucket-hosted history primary + SQL indexer service #15
15 changed files with 934 additions and 25 deletions
|
|
@ -13,6 +13,9 @@ RUN apt-get update \
|
|||
# Install deps first so this layer caches across code-only changes.
|
||||
COPY requirements.txt /tmp/requirements.txt
|
||||
RUN pip install --no-cache-dir -r /tmp/requirements.txt
|
||||
# Bake DuckDB's httpfs extension so the lake role's S3 reads need no runtime
|
||||
# extension download (tasks may roll while egress is degraded).
|
||||
RUN python -c "import duckdb; duckdb.connect().execute('INSTALL httpfs')"
|
||||
|
||||
COPY . /app/
|
||||
RUN chmod +x /app/deploy/entrypoint.sh
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from core import audit
|
|||
from core import metrics
|
||||
import paths
|
||||
from data import climate_store
|
||||
from data import era5lake
|
||||
from data import meteostat
|
||||
from data import store
|
||||
|
||||
|
|
@ -789,21 +790,32 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]:
|
|||
def _serve_stale():
|
||||
return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None}
|
||||
|
||||
# Fetch. NASA POWER is the primary source (keyless, independent of Open-Meteo);
|
||||
# the gusts it lacks are filled from Meteostat inside _fetch_history_nasa.
|
||||
# Open-Meteo is the fallback when NASA is unavailable — still guarded by its
|
||||
# rate-limit cooldown. Both sources must return a plausibly-full span before
|
||||
# being accepted and cached as a complete record (a short/partial response is
|
||||
# rejected rather than held indefinitely).
|
||||
# Fetch. The ERA5 lake is first when configured (true ERA5 with measured
|
||||
# gusts, served from our own bucket — no third-party API in the path);
|
||||
# unconfigured or missing-point cells fall through at zero cost. NASA
|
||||
# POWER is next (keyless, independent of Open-Meteo; its missing gusts
|
||||
# are filled from Meteostat inside _fetch_history_nasa), then Open-Meteo
|
||||
# — still guarded by its rate-limit cooldown. Every source must return a
|
||||
# plausibly-full span before being accepted and cached as a complete
|
||||
# record (a short/partial response is rejected rather than held
|
||||
# indefinitely).
|
||||
df = None
|
||||
source = None
|
||||
try:
|
||||
fetched = _fetch_history_nasa(cell)
|
||||
fetched = era5lake.fetch_history_lake(cell, start=START_DATE)
|
||||
if fetched.height >= MIN_ARCHIVE_DAYS:
|
||||
df = fetched
|
||||
source = "nasa-power"
|
||||
except Exception: # noqa: BLE001 - primary unavailable; fall through to Open-Meteo
|
||||
source = "era5-lake"
|
||||
except Exception: # noqa: BLE001 - lake unconfigured/miss; fall through to NASA
|
||||
pass
|
||||
if df is None:
|
||||
try:
|
||||
fetched = _fetch_history_nasa(cell)
|
||||
if fetched.height >= MIN_ARCHIVE_DAYS:
|
||||
df = fetched
|
||||
source = "nasa-power"
|
||||
except Exception: # noqa: BLE001 - primary unavailable; fall through to Open-Meteo
|
||||
pass
|
||||
if df is None and time.time() >= _archive_cooldown_until:
|
||||
try:
|
||||
fetched = _fetch_history(cell)
|
||||
|
|
|
|||
173
backend/data/era5lake.py
Normal file
173
backend/data/era5lake.py
Normal file
|
|
@ -0,0 +1,173 @@
|
|||
"""The ERA5 lake: layout, grid math, and read clients for the bucket-hosted archive.
|
||||
|
||||
The lake is a set of parquet files in an S3-compatible bucket (Contabo Object
|
||||
Storage, ``era5-thermograph``), extracted once from the public Earthmover ERA5
|
||||
Icechunk archive by ``gen_era5_lake.py`` and read three ways:
|
||||
|
||||
- the **lake service** (``lake_app.py``, prod-only Swarm service) serves point
|
||||
histories and SQL-style queries with a local disk cache in front;
|
||||
- the **web/worker backend** asks the lake service first
|
||||
(``THERMOGRAPH_LAKE_URL``), or reads the bucket directly when only S3
|
||||
credentials are configured (beta/LAN, no lake service);
|
||||
- ad-hoc analysis hits the hive table straight from the bucket.
|
||||
|
||||
Two layouts, one manifest (all under ``era5/``):
|
||||
|
||||
- ``daily/tile=<ti>_<tj>/year=<Y>/month=<M>/part.parquet`` — the analytical
|
||||
table, hive-partitioned by location/year/month. The location grain is a
|
||||
12x12-point tile (~3°x3°), aligned to the Earthmover store's chunk grid; a
|
||||
per-point x per-month grain would be ~170M objects, which no object store
|
||||
enjoys.
|
||||
- ``points/lat=<i>/lon=<j>.parquet`` — the serving projection: one file per
|
||||
ERA5 grid point holding its full daily history, so the hot path (one cell's
|
||||
history) is a single GET.
|
||||
- ``manifest.parquet`` — Iceberg-manifest-style index of every point file:
|
||||
grid indices, coordinates, row count, date span. Readers prune with it and
|
||||
the extractor resumes from it.
|
||||
|
||||
Columns everywhere: the finalized history-store schema (date, tmax, tmin,
|
||||
precip, wind, gust, humid, fmax, fmin, feels, doy — °F/in/mph/%, as
|
||||
``climate._finalize_approximated`` emits it), so a lake frame drops straight
|
||||
into ``climate._write_history_backed`` exactly like a NASA fetch does.
|
||||
"""
|
||||
import io
|
||||
import os
|
||||
|
||||
import polars as pl
|
||||
|
||||
# --- ERA5 native grid (0.25°, 721 x 1440, lat 90..-90, lon 0..359.75) ---------
|
||||
LAT_N = 721
|
||||
LON_N = 1440
|
||||
STEP = 0.25
|
||||
TILE = 12 # matches the Earthmover chunk grid (8736h x 12 x 12)
|
||||
|
||||
PREFIX = "era5"
|
||||
MANIFEST_KEY = f"{PREFIX}/manifest.parquet"
|
||||
|
||||
MIN_LAKE_DAYS = 10_000 # a plausibly-full 1980+ record, same spirit as
|
||||
# climate.MIN_ARCHIVE_DAYS (rejects partial files)
|
||||
|
||||
|
||||
def to_idx(lat: float, lon: float) -> tuple[int, int]:
|
||||
"""Nearest ERA5 grid indices for a lat/lon. lat_idx runs north->south."""
|
||||
i = round((90.0 - lat) / STEP)
|
||||
j = round((lon % 360.0) / STEP) % LON_N
|
||||
return min(max(i, 0), LAT_N - 1), j
|
||||
|
||||
|
||||
def to_coords(i: int, j: int) -> tuple[float, float]:
|
||||
"""Center lat/lon ([-180, 180) longitude) of grid indices."""
|
||||
lon = j * STEP
|
||||
return 90.0 - i * STEP, lon - 360.0 if lon >= 180.0 else lon
|
||||
|
||||
|
||||
def tile_of(i: int, j: int) -> tuple[int, int]:
|
||||
return i // TILE, j // TILE
|
||||
|
||||
|
||||
def point_key(i: int, j: int) -> str:
|
||||
return f"{PREFIX}/points/lat={i}/lon={j}.parquet"
|
||||
|
||||
|
||||
def daily_part_key(ti: int, tj: int, year: int, month: int) -> str:
|
||||
return f"{PREFIX}/daily/tile={ti}_{tj}/year={year}/month={month:02d}/part.parquet"
|
||||
|
||||
|
||||
# --- Config (all optional: unset means "no lake configured") ------------------
|
||||
|
||||
def lake_url() -> str:
|
||||
"""Lake service base URL (prod: the Swarm VIP), '' when not deployed."""
|
||||
return os.environ.get("THERMOGRAPH_LAKE_URL", "").rstrip("/")
|
||||
|
||||
|
||||
def s3_config() -> "dict | None":
|
||||
"""Bucket access for direct reads/writes, None unless fully configured.
|
||||
Contabo is path-style; region is a signing formality there ('default')."""
|
||||
key = os.environ.get("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "")
|
||||
secret = os.environ.get("THERMOGRAPH_LAKE_S3_SECRET_KEY", "")
|
||||
if not (key and secret):
|
||||
return None
|
||||
return {
|
||||
"endpoint": os.environ.get("THERMOGRAPH_LAKE_S3_ENDPOINT",
|
||||
"https://eu2.contabostorage.com"),
|
||||
"bucket": os.environ.get("THERMOGRAPH_LAKE_S3_BUCKET", "era5-thermograph"),
|
||||
"region": os.environ.get("THERMOGRAPH_LAKE_S3_REGION", "default"),
|
||||
"access_key": key,
|
||||
"secret_key": secret,
|
||||
}
|
||||
|
||||
|
||||
def local_dir() -> str:
|
||||
"""Filesystem mirror of the lake layout — tests and offline dev reads."""
|
||||
return os.environ.get("THERMOGRAPH_LAKE_LOCAL_DIR", "")
|
||||
|
||||
|
||||
def storage_options(cfg: dict) -> dict:
|
||||
"""polars/object_store options for scan_parquet against the bucket."""
|
||||
return {
|
||||
"aws_access_key_id": cfg["access_key"],
|
||||
"aws_secret_access_key": cfg["secret_key"],
|
||||
"aws_region": cfg["region"],
|
||||
"endpoint_url": cfg["endpoint"],
|
||||
# Contabo serves buckets on the path, not as a subdomain.
|
||||
"aws_virtual_hosted_style_request": "false",
|
||||
}
|
||||
|
||||
|
||||
# --- Point-history reads (the app's hot path) ----------------------------------
|
||||
|
||||
class LakeUnavailable(Exception):
|
||||
"""No lake configured, point not in the lake, or the read failed."""
|
||||
|
||||
|
||||
def _read_local(i: int, j: int) -> pl.DataFrame:
|
||||
path = os.path.join(local_dir(), point_key(i, j))
|
||||
if not os.path.exists(path):
|
||||
raise LakeUnavailable(f"no local lake file for point ({i}, {j})")
|
||||
return pl.read_parquet(path)
|
||||
|
||||
|
||||
def _read_service(i: int, j: int) -> pl.DataFrame:
|
||||
import httpx # noqa: PLC0415 - keep module import light for the extractor venv
|
||||
|
||||
lat, lon = to_coords(i, j)
|
||||
r = httpx.get(f"{lake_url()}/history", params={"lat": lat, "lon": lon},
|
||||
timeout=httpx.Timeout(20.0, connect=3.0))
|
||||
if r.status_code == 404:
|
||||
raise LakeUnavailable(f"point ({i}, {j}) not in the lake")
|
||||
r.raise_for_status()
|
||||
return pl.read_parquet(io.BytesIO(r.content))
|
||||
|
||||
|
||||
def _read_bucket(i: int, j: int, cfg: dict) -> pl.DataFrame:
|
||||
try:
|
||||
return pl.read_parquet(f"s3://{cfg['bucket']}/{point_key(i, j)}",
|
||||
storage_options=storage_options(cfg))
|
||||
except Exception as e: # noqa: BLE001 - object_store errors are not one type
|
||||
raise LakeUnavailable(f"bucket read failed for point ({i}, {j}): {e}") from e
|
||||
|
||||
|
||||
def fetch_history_lake(cell: dict, start: "str | None" = None) -> pl.DataFrame:
|
||||
"""Finalized daily history for the ERA5 point nearest a grid cell, from the
|
||||
first configured source: local dir -> lake service -> bucket. The lake
|
||||
stores each point's whole 1940+ record; pass ``start`` (the caller's
|
||||
grading window, e.g. climate.START_DATE) to slice — grading semantics stay
|
||||
the caller's, the deep record stays in the lake. Raises LakeUnavailable
|
||||
when unconfigured, missing, or implausibly short."""
|
||||
import datetime # noqa: PLC0415
|
||||
|
||||
i, j = to_idx(cell["center_lat"], cell["center_lon"])
|
||||
if local_dir():
|
||||
df = _read_local(i, j)
|
||||
elif lake_url():
|
||||
df = _read_service(i, j)
|
||||
else:
|
||||
cfg = s3_config()
|
||||
if cfg is None:
|
||||
raise LakeUnavailable("no lake configured")
|
||||
df = _read_bucket(i, j, cfg)
|
||||
if start is not None:
|
||||
df = df.filter(pl.col("date") >= datetime.date.fromisoformat(start))
|
||||
if df.height < MIN_LAKE_DAYS:
|
||||
raise LakeUnavailable(f"lake point ({i}, {j}) is short: {df.height} rows")
|
||||
return df.drop([c for c in ("lat_idx", "lon_idx") if c in df.columns])
|
||||
|
|
@ -62,6 +62,13 @@ run_migrations() {
|
|||
done
|
||||
}
|
||||
|
||||
# The lake role serves the ERA5 bucket only — no database, so no migrations: a
|
||||
# lake replica must come up (and stay up) even when Postgres is down.
|
||||
if [ "${THERMOGRAPH_ROLE:-}" = "lake" ]; then
|
||||
echo "==> Starting lake service on 0.0.0.0:${PORT:-8141} with ${WORKERS:-1} worker(s)"
|
||||
exec uvicorn lake_app:app --host 0.0.0.0 --port "${PORT:-8141}" --workers "${WORKERS:-1}"
|
||||
fi
|
||||
|
||||
# One-shot migrate mode (`entrypoint.sh migrate`, or THERMOGRAPH_MIGRATE_ONLY=1): run
|
||||
# the migration and exit — a one-shot task, so it runs once against the DB rather
|
||||
# than racing per web replica.
|
||||
|
|
|
|||
218
backend/gen_era5_lake.py
Normal file
218
backend/gen_era5_lake.py
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
"""Extract the ERA5 lake: Earthmover Icechunk archive -> the Thermograph bucket.
|
||||
|
||||
Builds the lake that ``data/era5lake.py`` reads (layout documented there):
|
||||
per-point serving files, the tile/year/month hive table, and the manifest.
|
||||
Pulls are tile-aligned to the source's chunk grid (8736h x 12 x 12) — one
|
||||
46-year pull (~19s measured) decodes all 144 points of a tile, so per-point
|
||||
pulls would waste 144x the wire. Roughly: ~600 tiles cover the curated
|
||||
cities (hours), ~3000 cover land (a day or two); both resume from the
|
||||
manifest, so ctrl-C and re-run is always safe.
|
||||
|
||||
pip install -r requirements-seed.txt 'numcodecs[pcodec]' boto3
|
||||
python gen_era5_lake.py --cities --dry-run # one tile, no writes
|
||||
python gen_era5_lake.py --cities # tiles covering cities.json
|
||||
python gen_era5_lake.py --tiles 20:30,40:60 # explicit tile-index window
|
||||
python gen_era5_lake.py --land # every land-containing tile
|
||||
|
||||
Destination: the bucket from THERMOGRAPH_LAKE_S3_* (see era5lake.s3_config),
|
||||
or a local mirror when THERMOGRAPH_LAKE_LOCAL_DIR is set (tests/rehearsal).
|
||||
Seed-only deps (icechunk/xarray/zarr) stay out of the app image, same policy
|
||||
as seed_era5.py.
|
||||
"""
|
||||
import argparse
|
||||
import datetime
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import polars as pl
|
||||
|
||||
from data import cities
|
||||
from data import climate
|
||||
from data import era5lake
|
||||
from seed_era5 import ERA5_BUCKET, ERA5_GROUP, ERA5_PREFIX, ERA5_REGION, ERA5_VARS, \
|
||||
hourly_to_daily
|
||||
|
||||
# The lake keeps each location's WHOLE available record (the store reaches
|
||||
# back to 1940); the app slices to its own grading window (climate.START_DATE)
|
||||
# at read time, so the extra decades are analytical surface, not a grading
|
||||
# change.
|
||||
START_DATE = "1940-01-01"
|
||||
|
||||
|
||||
def _open_store():
|
||||
import icechunk # noqa: PLC0415 - seed-only dep
|
||||
import xarray as xr # noqa: PLC0415
|
||||
|
||||
storage = icechunk.s3_storage(
|
||||
bucket=ERA5_BUCKET, prefix=ERA5_PREFIX, region=ERA5_REGION, anonymous=True)
|
||||
repo = icechunk.Repository.open(storage)
|
||||
return xr.open_zarr(repo.readonly_session("main").store,
|
||||
group=ERA5_GROUP, consolidated=False)
|
||||
|
||||
|
||||
class _Sink:
|
||||
"""Writes lake objects to the bucket (boto3) or a local mirror dir."""
|
||||
|
||||
def __init__(self):
|
||||
self.local = era5lake.local_dir()
|
||||
self.cfg = era5lake.s3_config()
|
||||
if not self.local and self.cfg is None:
|
||||
sys.exit("no destination: set THERMOGRAPH_LAKE_S3_* or "
|
||||
"THERMOGRAPH_LAKE_LOCAL_DIR")
|
||||
self._s3 = None
|
||||
if not self.local:
|
||||
import boto3 # noqa: PLC0415 - extractor-only dep
|
||||
self._s3 = boto3.client(
|
||||
"s3", endpoint_url=self.cfg["endpoint"],
|
||||
region_name=self.cfg["region"],
|
||||
aws_access_key_id=self.cfg["access_key"],
|
||||
aws_secret_access_key=self.cfg["secret_key"])
|
||||
|
||||
def put(self, key: str, df: pl.DataFrame) -> None:
|
||||
if self.local:
|
||||
path = os.path.join(self.local, key)
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
df.write_parquet(path)
|
||||
return
|
||||
buf = io.BytesIO()
|
||||
df.write_parquet(buf)
|
||||
self._s3.put_object(Bucket=self.cfg["bucket"], Key=key,
|
||||
Body=buf.getvalue())
|
||||
|
||||
def read_manifest(self) -> "pl.DataFrame | None":
|
||||
try:
|
||||
if self.local:
|
||||
return pl.read_parquet(
|
||||
os.path.join(self.local, era5lake.MANIFEST_KEY))
|
||||
obj = self._s3.get_object(Bucket=self.cfg["bucket"],
|
||||
Key=era5lake.MANIFEST_KEY)
|
||||
return pl.read_parquet(io.BytesIO(obj["Body"].read()))
|
||||
except Exception: # noqa: BLE001 - a fresh lake has no manifest yet
|
||||
return None
|
||||
|
||||
|
||||
def _tile_frame(ds, ti: int, tj: int, end: str) -> "pl.DataFrame | None":
|
||||
"""Pull one 12x12 tile's hourly series and return the finalized daily frame
|
||||
for all its points (lat_idx, lon_idx, date, ...). None for all-sea tiles."""
|
||||
t = era5lake.TILE
|
||||
i0, j0 = ti * t, tj * t
|
||||
lat_sl = slice(i0, min(i0 + t, era5lake.LAT_N))
|
||||
lon_sl = slice(j0, j0 + t)
|
||||
if "lsm" in ds: # land-sea mask: skip all-sea tiles outright
|
||||
lsm = ds["lsm"].isel(latitude=lat_sl, longitude=lon_sl)
|
||||
if "valid_time" in lsm.dims: # static field, stored with a time axis
|
||||
lsm = lsm.isel(valid_time=-1)
|
||||
if float(lsm.max()) < 0.05:
|
||||
return None
|
||||
sub = (ds[list(ERA5_VARS)]
|
||||
.isel(latitude=lat_sl, longitude=lon_sl)
|
||||
.sel(valid_time=slice(START_DATE, end))
|
||||
.load())
|
||||
|
||||
frames = []
|
||||
for a, lat in enumerate(sub.latitude.values):
|
||||
for b, lon in enumerate(sub.longitude.values):
|
||||
cols = {"time": sub.valid_time.values}
|
||||
for src, dst in ERA5_VARS.items():
|
||||
cols[dst] = sub[src].isel(latitude=a, longitude=b).values
|
||||
daily = climate._finalize_approximated(hourly_to_daily(pl.DataFrame(cols)))
|
||||
i, j = era5lake.to_idx(float(lat), float(lon))
|
||||
frames.append(daily.with_columns(
|
||||
pl.lit(i, dtype=pl.Int32).alias("lat_idx"),
|
||||
pl.lit(j, dtype=pl.Int32).alias("lon_idx")))
|
||||
return pl.concat(frames)
|
||||
|
||||
|
||||
def _write_tile(sink: _Sink, ti: int, tj: int, tile_df: pl.DataFrame) -> pl.DataFrame:
|
||||
"""Write one tile's point files + hive partitions; return its manifest rows."""
|
||||
rows = []
|
||||
for (i, j), pdf in tile_df.group_by(["lat_idx", "lon_idx"]):
|
||||
pdf = pdf.sort("date")
|
||||
sink.put(era5lake.point_key(i, j), pdf.drop(["lat_idx", "lon_idx"]))
|
||||
lat, lon = era5lake.to_coords(i, j)
|
||||
rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon,
|
||||
"tile": f"{ti}_{tj}", "rows": pdf.height,
|
||||
"date_min": str(pdf["date"].min()),
|
||||
"date_max": str(pdf["date"].max())})
|
||||
parts = tile_df.with_columns(
|
||||
pl.col("date").dt.year().alias("year"),
|
||||
pl.col("date").dt.month().alias("month"))
|
||||
for (year, month), mdf in parts.group_by(["year", "month"]):
|
||||
sink.put(era5lake.daily_part_key(ti, tj, year, month),
|
||||
mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"]))
|
||||
return pl.DataFrame(rows)
|
||||
|
||||
|
||||
def _city_tiles() -> list:
|
||||
from data import grid # noqa: PLC0415
|
||||
|
||||
tiles = set()
|
||||
for c in cities.all_cities():
|
||||
cell = grid.snap(c["lat"], c["lon"])
|
||||
tiles.add(era5lake.tile_of(
|
||||
*era5lake.to_idx(cell["center_lat"], cell["center_lon"])))
|
||||
return sorted(tiles)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--cities", action="store_true")
|
||||
ap.add_argument("--land", action="store_true")
|
||||
ap.add_argument("--tiles", help="ti0:ti1,tj0:tj1 tile-index window")
|
||||
ap.add_argument("--limit", type=int)
|
||||
ap.add_argument("--overwrite", action="store_true")
|
||||
ap.add_argument("--dry-run", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.cities:
|
||||
todo = _city_tiles()
|
||||
elif args.tiles:
|
||||
(a, b), (c, d) = (p.split(":") for p in args.tiles.split(","))
|
||||
todo = [(ti, tj) for ti in range(int(a), int(b))
|
||||
for tj in range(int(c), int(d))]
|
||||
elif args.land:
|
||||
todo = [(ti, tj) for ti in range(era5lake.LAT_N // era5lake.TILE + 1)
|
||||
for tj in range(era5lake.LON_N // era5lake.TILE)]
|
||||
else:
|
||||
ap.error("pick one of --cities / --tiles / --land")
|
||||
if args.limit:
|
||||
todo = todo[:args.limit]
|
||||
|
||||
sink = _Sink()
|
||||
ds = _open_store()
|
||||
end = (datetime.date.today()
|
||||
- datetime.timedelta(days=climate.ARCHIVE_LATENCY_DAYS)).isoformat()
|
||||
|
||||
manifest = sink.read_manifest()
|
||||
done = set() if manifest is None or args.overwrite else \
|
||||
set(manifest["tile"].unique().to_list())
|
||||
|
||||
written = skipped = empty = 0
|
||||
for n, (ti, tj) in enumerate(todo, 1):
|
||||
if f"{ti}_{tj}" in done:
|
||||
skipped += 1
|
||||
continue
|
||||
t0 = time.time()
|
||||
tile_df = _tile_frame(ds, ti, tj, end)
|
||||
if tile_df is None:
|
||||
empty += 1
|
||||
continue
|
||||
if args.dry_run:
|
||||
print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows "
|
||||
f"({time.time() - t0:.1f}s); nothing written")
|
||||
print(tile_df.head())
|
||||
return
|
||||
rows = _write_tile(sink, ti, tj, tile_df)
|
||||
manifest = rows if manifest is None else pl.concat(
|
||||
[manifest.filter(pl.col("tile") != f"{ti}_{tj}"), rows])
|
||||
sink.put(era5lake.MANIFEST_KEY, manifest) # after every tile: resumable
|
||||
written += 1
|
||||
print(f"[{n}/{len(todo)}] tile {ti}_{tj}: {rows.height} points "
|
||||
f"in {time.time() - t0:.1f}s")
|
||||
print(f"done: tiles written={written} skipped(done)={skipped} sea={empty}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
176
backend/lake_app.py
Normal file
176
backend/lake_app.py
Normal file
|
|
@ -0,0 +1,176 @@
|
|||
"""The lake service: low-latency SQL-ish reads over the ERA5 bucket (prod-only).
|
||||
|
||||
A small FastAPI app run as the Swarm ``lake`` service (same backend image,
|
||||
``THERMOGRAPH_ROLE=lake`` — the entrypoint launches this module instead of the
|
||||
web app; no database, no migrations). Web replicas call it instead of touching
|
||||
the bucket themselves, so a cell's history costs one LAN round-trip against a
|
||||
warm disk cache instead of an S3 fetch, and analytical queries get
|
||||
partition-pruned scans without shipping S3 credentials to every web task.
|
||||
|
||||
Endpoints:
|
||||
- ``GET /healthz`` liveness (also reports manifest row count)
|
||||
- ``GET /history?lat=&lon=`` the hot path: nearest ERA5 point's full daily
|
||||
history as parquet bytes, disk-cached per point
|
||||
(a plain single-file read — no engine needed)
|
||||
- ``POST /query {"sql": ..}`` SELECT-only SQL over two views: ``era5_daily``
|
||||
(the hive table, partition-pruned on
|
||||
tile/year/month predicates) and ``manifest``.
|
||||
Runs on DuckDB — a real lake engine (full SQL,
|
||||
parallel scans, hive + row-group pushdown); the
|
||||
httpfs extension is baked into the image so S3
|
||||
needs no runtime extension fetch.
|
||||
|
||||
Scale-out safe: replicas share nothing — each keeps its own cache under
|
||||
THERMOGRAPH_LAKE_CACHE (a per-task local volume), so 1..2 replicas behind the
|
||||
Swarm VIP need no coordination.
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
|
||||
import polars as pl
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.responses import FileResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from data import era5lake
|
||||
|
||||
CACHE_DIR = os.environ.get("THERMOGRAPH_LAKE_CACHE", "/state/lake-cache")
|
||||
MANIFEST_TTL = 6 * 3600 # re-pull the manifest at most this often
|
||||
QUERY_ROW_CAP = 10_000 # /query responses are JSON; keep them bounded
|
||||
|
||||
app = FastAPI(title="thermograph-lake")
|
||||
|
||||
_manifest_lock = threading.Lock()
|
||||
_manifest: "pl.DataFrame | None" = None
|
||||
_manifest_at = 0.0
|
||||
|
||||
|
||||
def _source():
|
||||
"""(mode, cfg) for this process: a local lake dir (tests/dev) or the bucket."""
|
||||
if era5lake.local_dir():
|
||||
return "local", None
|
||||
cfg = era5lake.s3_config()
|
||||
return ("bucket", cfg) if cfg else ("none", None)
|
||||
|
||||
|
||||
def _manifest_frame() -> "pl.DataFrame | None":
|
||||
global _manifest, _manifest_at
|
||||
with _manifest_lock:
|
||||
if _manifest is not None and time.time() - _manifest_at < MANIFEST_TTL:
|
||||
return _manifest
|
||||
mode, cfg = _source()
|
||||
try:
|
||||
if mode == "local":
|
||||
df = pl.read_parquet(
|
||||
os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY))
|
||||
elif mode == "bucket":
|
||||
df = pl.read_parquet(
|
||||
f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}",
|
||||
storage_options=era5lake.storage_options(cfg))
|
||||
else:
|
||||
return None
|
||||
except Exception: # noqa: BLE001 - empty/unreachable lake: serve health, 404 reads
|
||||
return None
|
||||
_manifest, _manifest_at = df, time.time()
|
||||
return df
|
||||
|
||||
|
||||
@app.get("/healthz")
|
||||
def healthz():
|
||||
mode, _ = _source()
|
||||
m = _manifest_frame()
|
||||
return {"status": "ok", "source": mode,
|
||||
"points": None if m is None else m.height}
|
||||
|
||||
|
||||
@app.get("/history")
|
||||
def history(lat: float, lon: float):
|
||||
"""One ERA5 point's full daily history, as parquet bytes (cached on disk —
|
||||
the cache file IS the response body, so a warm hit is a sendfile)."""
|
||||
i, j = era5lake.to_idx(lat, lon)
|
||||
path = os.path.join(CACHE_DIR, f"p{i}_{j}.parquet")
|
||||
if not os.path.exists(path):
|
||||
mode, cfg = _source()
|
||||
if mode == "none":
|
||||
raise HTTPException(status_code=503, detail="lake not configured")
|
||||
m = _manifest_frame()
|
||||
if m is not None and m.filter(
|
||||
(pl.col("lat_idx") == i) & (pl.col("lon_idx") == j)).is_empty():
|
||||
raise HTTPException(status_code=404, detail="point not in the lake")
|
||||
try:
|
||||
if mode == "local":
|
||||
df = era5lake._read_local(i, j)
|
||||
else:
|
||||
df = era5lake._read_bucket(i, j, cfg)
|
||||
except era5lake.LakeUnavailable as e:
|
||||
raise HTTPException(status_code=404, detail=str(e)) from e
|
||||
os.makedirs(CACHE_DIR, exist_ok=True)
|
||||
tmp = f"{path}.{os.getpid()}.partial"
|
||||
df.write_parquet(tmp)
|
||||
os.replace(tmp, path) # atomic: concurrent requests see whole files only
|
||||
return FileResponse(path, media_type="application/vnd.apache.parquet",
|
||||
headers={"X-Lake-Point": f"{i},{j}"})
|
||||
|
||||
|
||||
class Query(BaseModel):
|
||||
sql: str
|
||||
|
||||
|
||||
_FORBIDDEN = re.compile(
|
||||
r"\b(insert|update|delete|drop|create|alter|attach|copy|truncate|install|load"
|
||||
r"|export|import|pragma|set|call)\b", re.I)
|
||||
|
||||
|
||||
def _connect():
|
||||
"""A fresh in-memory DuckDB with the lake views registered. Per-request:
|
||||
connections are cheap, and a throwaway one means a query can't leave state
|
||||
behind for the next. Returns None when no lake is configured."""
|
||||
import duckdb # noqa: PLC0415 - only the lake role pays the import
|
||||
|
||||
mode, cfg = _source()
|
||||
if mode == "local":
|
||||
daily_glob = os.path.join(era5lake.local_dir(), era5lake.PREFIX,
|
||||
"daily/*/*/*/*.parquet")
|
||||
manifest_src = os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY)
|
||||
elif mode == "bucket":
|
||||
daily_glob = f"s3://{cfg['bucket']}/{era5lake.PREFIX}/daily/*/*/*/*.parquet"
|
||||
manifest_src = f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}"
|
||||
else:
|
||||
return None
|
||||
con = duckdb.connect()
|
||||
if mode == "bucket":
|
||||
con.execute("LOAD httpfs") # baked into the image at build time
|
||||
host = cfg["endpoint"].split("://", 1)[-1]
|
||||
con.execute(f"SET s3_endpoint='{host}'")
|
||||
con.execute("SET s3_url_style='path'")
|
||||
con.execute(f"SET s3_region='{cfg['region']}'")
|
||||
con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
|
||||
con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
|
||||
con.execute(f"""
|
||||
CREATE VIEW era5_daily AS
|
||||
SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""")
|
||||
con.execute(f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')")
|
||||
return con
|
||||
|
||||
|
||||
@app.post("/query")
|
||||
def query(q: Query):
|
||||
sql = q.sql.strip().rstrip(";").strip()
|
||||
if not sql.lower().startswith(("select", "with")) or ";" in sql \
|
||||
or _FORBIDDEN.search(sql):
|
||||
raise HTTPException(status_code=400,
|
||||
detail="only a single SELECT statement is allowed")
|
||||
con = _connect()
|
||||
if con is None:
|
||||
raise HTTPException(status_code=503, detail="lake not configured")
|
||||
try:
|
||||
out = con.execute(
|
||||
f"SELECT * FROM ({sql}) LIMIT {QUERY_ROW_CAP}").fetchall()
|
||||
cols = [d[0] for d in con.description]
|
||||
except Exception as e: # noqa: BLE001 - surface the engine's message to the caller
|
||||
raise HTTPException(status_code=400, detail=f"query failed: {e}") from e
|
||||
finally:
|
||||
con.close()
|
||||
return {"columns": cols, "rows": out, "row_cap": QUERY_ROW_CAP}
|
||||
|
|
@ -7,3 +7,10 @@
|
|||
icechunk
|
||||
xarray
|
||||
zarr
|
||||
# The live store compresses with PCodec; without this extra every read fails
|
||||
# with "codec not available: 'pcodec'".
|
||||
numcodecs[pcodec]
|
||||
# gen_era5_lake.py's bucket writer.
|
||||
boto3
|
||||
numcodecs[pcodec]
|
||||
boto3
|
||||
|
|
|
|||
|
|
@ -25,3 +25,6 @@ websockets==16.0
|
|||
# Recurring worker-tier jobs — city warming, IndexNow pings (see
|
||||
# notifications/scheduler.py). In-process; no broker/queue needed at this scale.
|
||||
apscheduler==3.11.3
|
||||
# The lake role's SQL engine (lake_app.py /query): embedded, parallel
|
||||
# partition-pruned scans over the ERA5 hive table. Only lake_app imports it.
|
||||
duckdb==1.4.2
|
||||
|
|
|
|||
|
|
@ -16,10 +16,10 @@ Idempotent by default: a cell that already has a cached history is skipped (pass
|
|||
--overwrite to reseed). Not run in CI or the app image — the icechunk/xarray/zarr
|
||||
stack lives only in requirements-seed.txt.
|
||||
|
||||
IMPORTANT — the S3/Icechunk *access* layer (_open_store / fetch_point) targets a
|
||||
young, fast-moving API and the store's exact variable/coordinate names; it is NOT
|
||||
exercised by the test suite. VERIFY it with `--dry-run` on the seed box before a
|
||||
full run. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the
|
||||
The access layer was VERIFIED against the live store 2026-07-23 (prefix
|
||||
`icechunkV2`, group `single/temporal`, coord `valid_time`, ECMWF short variable
|
||||
names, and the pcodec codec — install `numcodecs[pcodec]`). Re-verify with
|
||||
`--dry-run` if the store revs. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the
|
||||
part that must be numerically correct.
|
||||
|
||||
Source: https://registry.opendata.aws/earthmover-era5/ (group="temporal", anonymous)
|
||||
|
|
@ -38,19 +38,19 @@ START_DATE = climate.START_DATE # "1980-01-01"
|
|||
|
||||
# Anonymous AWS Open Data Icechunk store; overridable in case the registry moves.
|
||||
ERA5_BUCKET = os.environ.get("THERMOGRAPH_ERA5_BUCKET", "earthmover-icechunk-era5")
|
||||
ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "") # verify on the seed box
|
||||
ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "icechunkV2")
|
||||
ERA5_REGION = os.environ.get("THERMOGRAPH_ERA5_REGION", "us-east-1")
|
||||
ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "temporal") # time-contiguous layout
|
||||
ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "single/temporal") # time-contiguous layout
|
||||
|
||||
# ERA5 store variable name -> our short name. ERA5 gusts (i10fg) are real here, so the
|
||||
# seed keeps genuine measured gusts rather than the Meteostat/estimate fallback.
|
||||
ERA5_VARS = {
|
||||
"2m_temperature": "t2m", # K
|
||||
"2m_dewpoint_temperature": "d2m", # K (→ RH with t2m)
|
||||
"total_precipitation": "tp", # m, hourly accumulation
|
||||
"10m_u_component_of_wind": "u10", # m/s
|
||||
"10m_v_component_of_wind": "v10", # m/s
|
||||
"instantaneous_10m_wind_gust": "i10fg", # m/s
|
||||
"t2m": "t2m", # 2m temperature, K
|
||||
"d2m": "d2m", # 2m dewpoint, K (→ RH with t2m)
|
||||
"tp": "tp", # total precipitation, m, hourly accumulation
|
||||
"u10": "u10", # 10m wind u, m/s
|
||||
"v10": "v10", # 10m wind v, m/s
|
||||
"fg10": "i10fg", # 10m wind gust, m/s (store short name; transform expects i10fg)
|
||||
}
|
||||
|
||||
_M_TO_IN = 39.37007874
|
||||
|
|
@ -117,9 +117,9 @@ def fetch_point(ds, lat: float, lon: float, start: str, end: str) -> pl.DataFram
|
|||
may differ per store revision — verify with --dry-run."""
|
||||
sub = (ds[list(ERA5_VARS)]
|
||||
.sel(latitude=lat, longitude=lon % 360.0, method="nearest")
|
||||
.sel(time=slice(start, end))
|
||||
.sel(valid_time=slice(start, end))
|
||||
.load())
|
||||
data = {"time": sub["time"].values}
|
||||
data = {"time": sub["valid_time"].values}
|
||||
for src, dst in ERA5_VARS.items():
|
||||
data[dst] = sub[src].values
|
||||
return pl.DataFrame(data)
|
||||
|
|
|
|||
|
|
@ -311,6 +311,44 @@ def _full_history_frame(n=4000):
|
|||
}))
|
||||
|
||||
|
||||
def test_lake_is_first_history_source(monkeypatch, tmp_path):
|
||||
"""A configured ERA5 lake short-circuits the whole third-party chain: neither
|
||||
NASA POWER nor Open-Meteo is consulted."""
|
||||
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
|
||||
monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0)
|
||||
cell = {"id": "lake_primary", "center_lat": 47.6, "center_lon": -122.3}
|
||||
|
||||
full = _full_history_frame()
|
||||
monkeypatch.setattr(climate.era5lake, "fetch_history_lake",
|
||||
lambda c, start=None: full)
|
||||
def _nasa_should_not_run(c): raise AssertionError("NASA should not be called")
|
||||
def _om_should_not_run(c): raise AssertionError("Open-Meteo should not be called")
|
||||
monkeypatch.setattr(climate, "_fetch_history_nasa", _nasa_should_not_run)
|
||||
monkeypatch.setattr(climate, "_fetch_history", _om_should_not_run)
|
||||
|
||||
df, meta = climate._load_history(cell)
|
||||
assert meta["source"] == "era5-lake"
|
||||
assert df.height == full.height
|
||||
|
||||
|
||||
def test_lake_miss_falls_through_to_nasa(monkeypatch, tmp_path):
|
||||
"""An unconfigured lake (or a point outside it) costs nothing: NASA serves
|
||||
exactly as before the lake existed."""
|
||||
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
|
||||
monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0)
|
||||
cell = {"id": "lake_miss", "center_lat": 47.6, "center_lon": -122.3}
|
||||
|
||||
def _no_lake(c, start=None):
|
||||
raise climate.era5lake.LakeUnavailable("no lake configured")
|
||||
monkeypatch.setattr(climate.era5lake, "fetch_history_lake", _no_lake)
|
||||
full = _full_history_frame()
|
||||
monkeypatch.setattr(climate, "_fetch_history_nasa", lambda c: full)
|
||||
|
||||
df, meta = climate._load_history(cell)
|
||||
assert meta["source"] == "nasa-power"
|
||||
assert df.height == full.height
|
||||
|
||||
|
||||
def test_nasa_is_primary_history_source(monkeypatch, tmp_path):
|
||||
"""NASA POWER is the primary history source: a full NASA frame is accepted and
|
||||
Open-Meteo is never consulted."""
|
||||
|
|
|
|||
90
backend/tests/data/test_era5lake.py
Normal file
90
backend/tests/data/test_era5lake.py
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
"""The ERA5 lake: grid math, layout keys, and the read/slice client (no network
|
||||
— the local-dir source stands in for the bucket)."""
|
||||
import datetime
|
||||
|
||||
import polars as pl
|
||||
import pytest
|
||||
|
||||
from data import era5lake
|
||||
|
||||
|
||||
def _daily_frame(start="1940-01-01", days=32000):
|
||||
d0 = datetime.date.fromisoformat(start)
|
||||
dates = [d0 + datetime.timedelta(days=k) for k in range(days)]
|
||||
n = len(dates)
|
||||
return pl.DataFrame({
|
||||
"date": dates,
|
||||
"tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n,
|
||||
"wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n,
|
||||
"fmax": [71.0] * n, "fmin": [48.0] * n,
|
||||
})
|
||||
|
||||
|
||||
def test_grid_roundtrip_and_bounds():
|
||||
# Nearest-index snap, poles clamped, longitude wrapped to [0, 360).
|
||||
assert era5lake.to_idx(90.0, 0.0) == (0, 0)
|
||||
assert era5lake.to_idx(-90.0, 0.0) == (720, 0)
|
||||
assert era5lake.to_idx(51.5074, -0.1278) == (154, 1439)
|
||||
i, j = era5lake.to_idx(35.6762, 139.6503)
|
||||
lat, lon = era5lake.to_coords(i, j)
|
||||
assert abs(lat - 35.6762) <= era5lake.STEP / 2 + 1e-9
|
||||
assert abs(lon - 139.6503) <= era5lake.STEP / 2 + 1e-9
|
||||
|
||||
|
||||
def test_tile_and_keys_align_with_hive_layout():
|
||||
assert era5lake.tile_of(154, 1439) == (12, 119)
|
||||
assert era5lake.point_key(154, 1439) == "era5/points/lat=154/lon=1439.parquet"
|
||||
assert era5lake.daily_part_key(12, 119, 1994, 3) == \
|
||||
"era5/daily/tile=12_119/year=1994/month=03/part.parquet"
|
||||
|
||||
|
||||
def test_s3_config_requires_both_keys(monkeypatch):
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False)
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False)
|
||||
assert era5lake.s3_config() is None
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "k")
|
||||
assert era5lake.s3_config() is None
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", "s")
|
||||
cfg = era5lake.s3_config()
|
||||
assert cfg["bucket"] == "era5-thermograph"
|
||||
opts = era5lake.storage_options(cfg)
|
||||
assert opts["aws_virtual_hosted_style_request"] == "false"
|
||||
|
||||
|
||||
def _write_point(tmp_path, cell, df):
|
||||
i, j = era5lake.to_idx(cell["center_lat"], cell["center_lon"])
|
||||
path = tmp_path / era5lake.point_key(i, j)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
df.write_parquet(path)
|
||||
|
||||
|
||||
CELL = {"center_lat": 51.5, "center_lon": -0.125}
|
||||
|
||||
|
||||
def test_fetch_slices_to_start_and_keeps_schema(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
|
||||
_write_point(tmp_path, CELL, _daily_frame())
|
||||
df = era5lake.fetch_history_lake(CELL, start="1980-01-01")
|
||||
assert df["date"].min() == datetime.date(1980, 1, 1)
|
||||
assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind",
|
||||
"gust", "humid", "fmax", "fmin"}
|
||||
# Without a start, the whole 1940+ record comes back.
|
||||
assert era5lake.fetch_history_lake(CELL).height == 32000
|
||||
|
||||
|
||||
def test_fetch_rejects_short_and_missing(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
|
||||
with pytest.raises(era5lake.LakeUnavailable):
|
||||
era5lake.fetch_history_lake(CELL) # nothing written
|
||||
_write_point(tmp_path, CELL, _daily_frame(days=500))
|
||||
with pytest.raises(era5lake.LakeUnavailable): # implausibly short record
|
||||
era5lake.fetch_history_lake(CELL)
|
||||
|
||||
|
||||
def test_fetch_unconfigured_raises(monkeypatch):
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_LOCAL_DIR", raising=False)
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_URL", raising=False)
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False)
|
||||
monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False)
|
||||
with pytest.raises(era5lake.LakeUnavailable):
|
||||
era5lake.fetch_history_lake(CELL)
|
||||
100
backend/tests/web/test_lake_app.py
Normal file
100
backend/tests/web/test_lake_app.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
"""The lake service over a local-dir lake fixture: health, the point hot path
|
||||
(+ its disk cache), and the SELECT-only SQL surface."""
|
||||
import datetime
|
||||
import io
|
||||
|
||||
import polars as pl
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lake(tmp_path, monkeypatch):
|
||||
"""A tiny two-point lake (points, hive partitions, manifest) on disk."""
|
||||
from data import era5lake
|
||||
|
||||
points = [(154, 1439), (154, 0)] # around London
|
||||
d0 = datetime.date(1940, 1, 1)
|
||||
dates = [d0 + datetime.timedelta(days=k) for k in range(12000)]
|
||||
n = len(dates)
|
||||
rows = []
|
||||
for i, j in points:
|
||||
df = pl.DataFrame({
|
||||
"date": dates,
|
||||
"tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n,
|
||||
"wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n,
|
||||
"fmax": [71.0] * n, "fmin": [48.0] * n,
|
||||
})
|
||||
p = tmp_path / era5lake.point_key(i, j)
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
df.write_parquet(p)
|
||||
ti, tj = era5lake.tile_of(i, j)
|
||||
part = df.with_columns(pl.lit(i, dtype=pl.Int32).alias("lat_idx"),
|
||||
pl.lit(j, dtype=pl.Int32).alias("lon_idx"))
|
||||
hp = tmp_path / era5lake.daily_part_key(ti, tj, 1940, 1)
|
||||
hp.parent.mkdir(parents=True, exist_ok=True)
|
||||
part.filter((pl.col("date").dt.year() == 1940)
|
||||
& (pl.col("date").dt.month() == 1)).write_parquet(hp)
|
||||
lat, lon = era5lake.to_coords(i, j)
|
||||
rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon,
|
||||
"tile": f"{ti}_{tj}", "rows": n,
|
||||
"date_min": "1940-01-01", "date_max": str(dates[-1])})
|
||||
(tmp_path / "era5").mkdir(exist_ok=True)
|
||||
pl.DataFrame(rows).write_parquet(tmp_path / era5lake.MANIFEST_KEY)
|
||||
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path))
|
||||
monkeypatch.setenv("THERMOGRAPH_LAKE_CACHE", str(tmp_path / "cache"))
|
||||
import lake_app
|
||||
monkeypatch.setattr(lake_app, "CACHE_DIR", str(tmp_path / "cache"))
|
||||
monkeypatch.setattr(lake_app, "_manifest", None)
|
||||
return TestClient(lake_app.app)
|
||||
|
||||
|
||||
def test_healthz_reports_points(lake):
|
||||
body = lake.get("/healthz").json()
|
||||
assert body["status"] == "ok"
|
||||
assert body["source"] == "local"
|
||||
assert body["points"] == 2
|
||||
|
||||
|
||||
def test_history_serves_parquet_and_caches(lake, tmp_path):
|
||||
r = lake.get("/history", params={"lat": 51.5074, "lon": -0.1278})
|
||||
assert r.status_code == 200
|
||||
assert r.headers["x-lake-point"] == "154,1439"
|
||||
df = pl.read_parquet(io.BytesIO(r.content))
|
||||
assert df.height == 12000
|
||||
assert (tmp_path / "cache" / "p154_1439.parquet").exists()
|
||||
# Second hit comes off the cache file and is byte-identical.
|
||||
assert lake.get("/history",
|
||||
params={"lat": 51.5074, "lon": -0.1278}).content == r.content
|
||||
|
||||
|
||||
def test_history_404_for_point_outside_lake(lake):
|
||||
r = lake.get("/history", params={"lat": -45.0, "lon": 170.0})
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_query_selects_with_partition_predicates(lake):
|
||||
r = lake.post("/query", json={"sql":
|
||||
"SELECT year, month, count(*) AS n FROM era5_daily "
|
||||
"WHERE year = 1940 AND month = 1 GROUP BY year, month"})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["columns"] == ["year", "month", "n"]
|
||||
assert body["rows"][0][2] == 2 * 31 # two points x January 1940
|
||||
|
||||
|
||||
def test_query_manifest_view(lake):
|
||||
r = lake.post("/query", json={"sql": "SELECT count(*) FROM manifest"})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["rows"][0][0] == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sql", [
|
||||
"DROP TABLE era5_daily",
|
||||
"SELECT 1; SELECT 2",
|
||||
"INSERT INTO era5_daily VALUES (1)",
|
||||
"COPY era5_daily TO 'x'",
|
||||
])
|
||||
def test_query_rejects_non_select(lake, sql):
|
||||
assert lake.post("/query", json={"sql": sql}).status_code == 400
|
||||
|
|
@ -15,7 +15,7 @@ no per-host duplication.
|
|||
| File | Holds | Encrypted? |
|
||||
|------|-------|------------|
|
||||
| `common.yaml` | Secrets shared by every host — Discord tokens/secret, VAPID keypair, `THERMOGRAPH_AUTH_SECRET`, metrics token, indexnow key, SMTP creds | ✅ |
|
||||
| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN` | ✅ |
|
||||
| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN`, ERA5 lake bucket creds (`THERMOGRAPH_LAKE_S3_ACCESS_KEY` / `THERMOGRAPH_LAKE_S3_SECRET_KEY`, consumed by the `lake` Swarm service and `gen_era5_lake.py`) | ✅ |
|
||||
| `beta.yaml` | Beta-only overrides | ✅ |
|
||||
| `example.yaml` | Format reference / CI fixture (fake values) | ✅ |
|
||||
| `../../.sops.yaml` | Which age recipient files are encrypted to (plaintext config) | — |
|
||||
|
|
|
|||
|
|
@ -21,7 +21,9 @@
|
|||
set -eu
|
||||
|
||||
STACK_NAME="${STACK_NAME:-thermograph}"
|
||||
SERVICE="${STACK_NAME}_web"
|
||||
# Which stack service to scale (the stack runs one autoscaler instance per
|
||||
# scaled service — web and lake today; same loop, different bounds).
|
||||
SERVICE="${STACK_NAME}_${TARGET_SERVICE:-web}"
|
||||
MIN="${MIN_REPLICAS:-1}"
|
||||
MAX="${MAX_REPLICAS:-3}"
|
||||
UP_AT="${SCALE_UP_CPU:-220}"
|
||||
|
|
|
|||
|
|
@ -82,6 +82,10 @@ services:
|
|||
# not the compose bridge's 172.19.0.1 (an overlay has no host gateway).
|
||||
# provision-mail.sh's DOCKER_MAIL_GATEWAY/SUBNET cover this listener.
|
||||
THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1}
|
||||
# History reads ask the lake service (Swarm VIP over 1..2 replicas)
|
||||
# before any third-party source; unset on beta/LAN, where compose has no
|
||||
# lake service and climate.py falls straight through to NASA.
|
||||
THERMOGRAPH_LAKE_URL: http://lake:8141
|
||||
volumes:
|
||||
- appdata:/state
|
||||
- applogs:/app/logs
|
||||
|
|
@ -121,6 +125,7 @@ services:
|
|||
THERMOGRAPH_SINGLETON_PG: "1"
|
||||
RUN_MIGRATIONS: "0"
|
||||
THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1}
|
||||
THERMOGRAPH_LAKE_URL: http://lake:8141
|
||||
volumes:
|
||||
- appdata:/state
|
||||
- applogs:/app/logs
|
||||
|
|
@ -138,6 +143,45 @@ services:
|
|||
restart_policy:
|
||||
condition: on-failure
|
||||
|
||||
# The ERA5 lake indexer: SQL-style, partition-pruned reads over the
|
||||
# era5-thermograph bucket with a node-local parquet cache, so a cell's
|
||||
# history is one LAN hop for web instead of an S3 fetch (or a NASA call).
|
||||
# Same backend image; THERMOGRAPH_ROLE=lake makes the entrypoint start
|
||||
# lake_app (no database, no migrations). Prod-only by construction: it only
|
||||
# exists in this stack file, and its S3 credentials
|
||||
# (THERMOGRAPH_LAKE_S3_ACCESS_KEY/_SECRET_KEY) arrive via the same rendered
|
||||
# /host/thermograph.env as every other secret. Without them it stays
|
||||
# healthy and answers 503/404, and web falls through to NASA — the lake is
|
||||
# an accelerator, never a point of failure.
|
||||
lake:
|
||||
image: ${REGISTRY_HOST:-git.thermograph.org}/${BACKEND_IMAGE_PATH:-emi/thermograph/backend}:${BACKEND_IMAGE_TAG:?required}
|
||||
entrypoint: ["/host/env-entrypoint.sh"]
|
||||
environment:
|
||||
THERMOGRAPH_ROLE: lake
|
||||
PORT: 8141
|
||||
THERMOGRAPH_SERVICE_ROLE: backend
|
||||
WORKERS: "1"
|
||||
THERMOGRAPH_LAKE_CACHE: /state/lake-cache
|
||||
volumes:
|
||||
# Replicas share the cache; lake_app's tmp+rename writes make that safe.
|
||||
- lakecache:/state
|
||||
- /opt/thermograph/infra/deploy/stack/env-entrypoint.sh:/host/env-entrypoint.sh:ro
|
||||
- /etc/thermograph/stack.env:/host/thermograph.env:ro
|
||||
networks:
|
||||
- internal
|
||||
deploy:
|
||||
replicas: ${LAKE_MIN_REPLICAS:-1}
|
||||
placement:
|
||||
constraints: ["node.role == manager"]
|
||||
resources:
|
||||
limits:
|
||||
cpus: "${LAKE_CPUS:-2}"
|
||||
restart_policy:
|
||||
condition: on-failure
|
||||
update_config:
|
||||
order: start-first
|
||||
failure_action: rollback
|
||||
|
||||
frontend:
|
||||
image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required}
|
||||
entrypoint: ["/host/env-entrypoint.sh"]
|
||||
|
|
@ -200,6 +244,39 @@ services:
|
|||
restart_policy:
|
||||
condition: any
|
||||
|
||||
# Second autoscaler instance, watching `lake` (1..2 replicas). The lake is
|
||||
# read-only and cache-heavy, so it saturates on decode CPU well below web's
|
||||
# thresholds — scale up sooner, still down slowly.
|
||||
autoscaler-lake:
|
||||
image: docker:27-cli
|
||||
entrypoint: ["/bin/sh", "/host/autoscale.sh"]
|
||||
environment:
|
||||
STACK_NAME: ${STACK_NAME:-thermograph}
|
||||
TARGET_SERVICE: lake
|
||||
MIN_REPLICAS: ${LAKE_MIN_REPLICAS:-1}
|
||||
MAX_REPLICAS: ${LAKE_MAX_REPLICAS:-2}
|
||||
SCALE_UP_CPU: ${LAKE_SCALE_UP_CPU:-120}
|
||||
SCALE_DOWN_CPU: ${LAKE_SCALE_DOWN_CPU:-30}
|
||||
POLL_SECONDS: ${POLL_SECONDS:-15}
|
||||
UP_SAMPLES: ${UP_SAMPLES:-3}
|
||||
DOWN_SAMPLES: ${DOWN_SAMPLES:-20}
|
||||
COOLDOWN_SECONDS: ${COOLDOWN_SECONDS:-180}
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
- /opt/thermograph/infra/deploy/stack/autoscale.sh:/host/autoscale.sh:ro
|
||||
networks:
|
||||
- internal
|
||||
deploy:
|
||||
replicas: 1
|
||||
placement:
|
||||
constraints: ["node.role == manager"]
|
||||
resources:
|
||||
limits:
|
||||
cpus: "0.2"
|
||||
memory: 64m
|
||||
restart_policy:
|
||||
condition: any
|
||||
|
||||
networks:
|
||||
internal:
|
||||
driver: overlay
|
||||
|
|
@ -221,3 +298,6 @@ volumes:
|
|||
applogs:
|
||||
external: true
|
||||
name: ${APPLOGS_VOLUME:-thermograph_applogs}
|
||||
# Lake parquet cache: node-local and disposable (a cold cache just refills
|
||||
# from the bucket), so NOT external — the stack owns its lifecycle.
|
||||
lakecache: {}
|
||||
|
|
|
|||
Loading…
Reference in a new issue