Spaces:
Running
Running
| """CREST_demo — FastAPI backend for the map-centric dashboard. | |
| Serves the Leaflet frontend + JSON/SSE APIs, reusing the hf_data layer + agents. | |
| POST /api/query NL query -> region + event pins + gauge pins (agent-driven) | |
| GET /api/gauges gauge pins for a map viewport (map-first, no AI needed) | |
| POST /api/simulate selected gauges -> sim job (cap 10; ~4 concurrent) | |
| GET /api/stream SSE live status + hydrograph + 2-D overlays | |
| POST /api/eventinfo LLM event brief (damage/fatalities/links) for the AI info feed | |
| POST /api/calibrate AI calibration job (NSE<0.3 flow); GET /api/calstream SSE | |
| /login /auth/callback /logout /api/me /api/profile HF OAuth + profile store | |
| GET/POST /api/favorites registered-user focused basins (up to 5 gauges) | |
| Run: python server.py (uvicorn on :7860, Docker Space entrypoint) | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import sys | |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) | |
| import asyncio | |
| import json | |
| import math | |
| import secrets | |
| import time | |
| from datetime import datetime, timedelta | |
| from fastapi import FastAPI, Request, Response | |
| from fastapi.responses import FileResponse, JSONResponse, RedirectResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel | |
| from sse_starlette.sse import EventSourceResponse | |
| from starlette.middleware.sessions import SessionMiddleware | |
| from hf_data import caljobs, crashlog, datamgr, gauges, persist, simjobs | |
| from hf_data.pipeline import parse_query | |
| from hf_data.statecache import CACHE_DIR | |
| crashlog.init() # optional SENTRY_DSN mirror (Sentry/GlitchTip/Bugsink) | |
| crashlog.install_thread_hook() # record any thread that dies unwrapped | |
| datamgr.start_janitor() # hourly cache cleanup + result compaction | |
| persist.start() # restore states/results/params/users from the | |
| # private dataset, then keep them synced | |
| HERE = os.path.dirname(os.path.abspath(__file__)) | |
| FRONTEND = os.path.join(HERE, "frontend") | |
| MAX_SIMS = 10 # demo cap (selection) | |
| app = FastAPI(title="CREST_demo") | |
| app.add_middleware(SessionMiddleware, | |
| secret_key=os.environ.get("SESSION_SECRET", secrets.token_hex(32)), | |
| max_age=14 * 24 * 3600, same_site="lax", https_only=False) | |
| # ---- HF OAuth (Space sets OAUTH_* env when README has hf_oauth: true) ------ | |
| OAUTH_CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID") | |
| _oauth = None | |
| if OAUTH_CLIENT_ID: | |
| from authlib.integrations.starlette_client import OAuth | |
| _oauth = OAuth() | |
| _oauth.register( | |
| name="huggingface", | |
| client_id=OAUTH_CLIENT_ID, | |
| client_secret=os.environ.get("OAUTH_CLIENT_SECRET"), | |
| server_metadata_url=(os.environ.get("OPENID_PROVIDER_URL", "https://huggingface.co") | |
| + "/.well-known/openid-configuration"), | |
| client_kwargs={"scope": os.environ.get("OAUTH_SCOPES", "openid profile")}, | |
| ) | |
| def _profile_path(username: str) -> str: | |
| d = os.path.join(CACHE_DIR, "users") | |
| os.makedirs(d, exist_ok=True) | |
| safe = "".join(c for c in username if c.isalnum() or c in "-_") | |
| return os.path.join(d, f"{safe}.json") | |
| def _load_profile(username: str) -> dict: | |
| try: | |
| with open(_profile_path(username), encoding="utf-8") as fh: | |
| return json.load(fh) | |
| except Exception: | |
| return {} | |
| # ---- admin role: HF usernames allowed to browse the FULL result archive ----- | |
| # (user directive 2026-09-08: the owner's Hugging Face account only, for now; | |
| # widen with the ADMIN_USERS Space variable, comma-separated usernames) | |
| ADMIN_USERS = {u.strip().lower() for u in | |
| os.environ.get("ADMIN_USERS", "vincewin").split(",") if u.strip()} | |
| def _sid(request: Request) -> str: | |
| """Per-BROWSER identity (UX isolation, user 2026-09-28): a random id kept | |
| in the session cookie, independent of the HF login — people on a shared | |
| account or a shared machine still get their own jobs, tabs and panels.""" | |
| sid = request.session.get("sid") | |
| if not sid: | |
| sid = secrets.token_hex(8) | |
| request.session["sid"] = sid | |
| return sid | |
| def _owner(request: Request) -> dict: | |
| """Owner record stamped on a job at start; the token goes back to the | |
| browser that started it (survives a session reset / a shared account).""" | |
| u = request.session.get("user") or {} | |
| return {"sid": _sid(request), "user": u.get("username"), | |
| "token": secrets.token_hex(12)} | |
| def _owns(job, request: Request, tok: str = "", strict: bool = False) -> bool: | |
| """May this browser act on the job? Its own job (same sid, or the token it | |
| was handed) always; the same signed-in account on another device only | |
| for non-strict reads (an explicit Open from the history list).""" | |
| o = getattr(job, "owner", None) or {} | |
| if not o: | |
| return True # legacy job, no owner | |
| if tok and tok == o.get("token"): | |
| return True | |
| if o.get("sid") and o.get("sid") == request.session.get("sid"): | |
| return True | |
| if strict: | |
| return False | |
| u = (request.session.get("user") or {}).get("username") | |
| return bool(u) and u == o.get("user") | |
| _FORBIDDEN = {"error": "this simulation belongs to another browser session"} | |
| def _is_admin(request: Request) -> bool: | |
| u = request.session.get("user") or {} | |
| return bool(u.get("username")) and u["username"].lower() in ADMIN_USERS | |
| async def _error_recorder(request: Request, call_next): | |
| """Error watchdog: every unhandled server error is recorded with context.""" | |
| try: | |
| return await call_next(request) | |
| except Exception as e: | |
| crashlog.capture(f"http:{request.url.path}", e, method=request.method) | |
| raise | |
| def api_errors(n: int = 50): | |
| """Recent recorded errors (the error-watchdog log).""" | |
| return {"errors": crashlog.recent(min(n, 200)), **crashlog.stats()} | |
| class ClientError(BaseModel): | |
| message: str = "" | |
| source: str | None = None | |
| line: int | None = None | |
| stack: str | None = None | |
| _client_err_ts: list[float] = [] # flood guard for the browser error beacon | |
| def api_clienterror(err: ClientError): | |
| """Browser-side error beacon (app.js window.onerror/unhandledrejection) — | |
| frontend crashes land in the same watchdog log as server errors.""" | |
| now = time.time() | |
| _client_err_ts[:] = [t for t in _client_err_ts if now - t < 60] | |
| if len(_client_err_ts) >= 30: # global: max 30/min | |
| return {"ok": False, "throttled": True} | |
| _client_err_ts.append(now) | |
| crashlog.capture("client:js", message=(err.message or "")[:400], | |
| source=err.source, line=err.line, stack=err.stack) | |
| return {"ok": True} | |
| def api_datastats(): | |
| """Data-manager view: per-category disk usage + caps.""" | |
| return datamgr.stats() | |
| def api_datacleanup(): | |
| """Run a janitor pass now (cleanup + result compaction).""" | |
| rep = datamgr.cleanup() | |
| rep["compact"] = datamgr.compact_results() | |
| return rep | |
| def api_report(sim_id: str, gauge_id: str, request: Request, tok: str = ""): | |
| """Downloadable simulation report (AQUAH report-writer agent -> PDF). | |
| Generated on first request (LLM + pandoc, ~30-90 s), then cached. Signed-in | |
| users get a persistent copy in their report library; anonymous reports stay | |
| in the ephemeral cache and are discarded when they close the app.""" | |
| job = simjobs.get_job(sim_id) | |
| if job is None: | |
| return JSONResponse({"error": "simulation not found (it may have been " | |
| "restarted away) — re-run it, then download " | |
| "the report"}, status_code=404) | |
| if not _owns(job, request, tok): | |
| return JSONResponse(_FORBIDDEN, status_code=403) | |
| if not (job.hydro.get(gauge_id) or []): | |
| return JSONResponse({"error": "no results for this gauge yet"}, status_code=409) | |
| from hf_data import report | |
| try: | |
| path = report.generate(job, gauge_id) | |
| except Exception as e: | |
| crashlog.capture("report", e, sim_id=sim_id, gauge=gauge_id) | |
| return JSONResponse({"error": f"report generation failed: {e}"}, status_code=500) | |
| saved = False | |
| u = request.session.get("user") | |
| if u: # registered-user benefit: keep the report | |
| try: | |
| report.save_for_user(u["username"], gauge_id, job, path) | |
| persist.poke() | |
| saved = True | |
| except Exception as e: | |
| crashlog.capture("report:save", e, user=u.get("username")) | |
| ext = os.path.splitext(path)[1] | |
| return FileResponse( | |
| path, | |
| media_type="application/pdf" if ext == ".pdf" else "text/markdown", | |
| filename=f"CREST_report_{gauge_id}_{job.t_start:%Y%m%d}{ext}", | |
| headers={"X-Report-Saved": "1" if saved else "0"}) | |
| def api_myreports(request: Request): | |
| """Signed-in user's saved report library.""" | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| from hf_data import report | |
| return {"reports": report.list_saved(u["username"]), | |
| "max": report.MAX_SAVED_PER_USER} | |
| def api_myreports_get(name: str, request: Request): | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| from hf_data import report | |
| p = report.saved_path(u["username"], name) | |
| if not p: | |
| return JSONResponse({"error": "report not found"}, status_code=404) | |
| return FileResponse(p, media_type="application/pdf" if p.endswith(".pdf") | |
| else "text/markdown", filename=os.path.basename(p)) | |
| def api_myreports_delete(name: str, request: Request): | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| from hf_data import report | |
| p = report.saved_path(u["username"], name) | |
| if p: | |
| os.remove(p) | |
| persist.poke() # mirror the deletion to the private dataset | |
| return {"ok": True, "reports": report.list_saved(u["username"])} | |
| def api_report_discard(sim_id: str): | |
| """App-close beacon from anonymous users: drop their ephemeral reports. | |
| Saved (registered) copies are untouched; everything here is regenerable.""" | |
| from hf_data import report | |
| return {"ok": True, "removed": report.discard_sim(sim_id)} | |
| def api_persist_status(): | |
| """Durable-cache sync status (private-dataset mirror).""" | |
| return persist.status() | |
| def api_persist_now(): | |
| """Push a sync pass right now (normally every 10 min / after each run).""" | |
| try: | |
| return {"ok": True, **persist.backup()} | |
| except Exception as e: | |
| return JSONResponse({"ok": False, "error": str(e)}, status_code=500) | |
| APP_VERSION = str(int(time.time())) # changes every deploy/restart | |
| def index(): | |
| """index.html is never browser-cached, and its asset URLs carry a version | |
| that changes on every deploy — users always get the latest frontend | |
| without needing a hard refresh.""" | |
| with open(os.path.join(FRONTEND, "index.html"), encoding="utf-8") as fh: | |
| html = fh.read().replace("__V__", APP_VERSION) | |
| return Response(html, media_type="text/html", | |
| headers={"Cache-Control": "no-cache"}) | |
| async def login(request: Request): | |
| if _oauth is None: # local dev fallback: fake session, no HF round-trip | |
| request.session["user"] = {"username": "dev-user", "name": "Dev User", | |
| "picture": None, "dev": True} | |
| return RedirectResponse("/") | |
| redirect_uri = str(request.url_for("auth_callback")).replace("http://", "https://") \ | |
| if os.environ.get("SPACE_HOST") else request.url_for("auth_callback") | |
| return await _oauth.huggingface.authorize_redirect(request, redirect_uri) | |
| async def auth_callback(request: Request): | |
| if _oauth is None: | |
| return RedirectResponse("/") | |
| try: | |
| token = await _oauth.huggingface.authorize_access_token(request) | |
| info = token.get("userinfo") or {} | |
| request.session["user"] = { | |
| "username": info.get("preferred_username") or info.get("sub", "user"), | |
| "name": info.get("name") or info.get("preferred_username", "user"), | |
| "picture": info.get("picture"), | |
| } | |
| except Exception as e: | |
| crashlog.capture("auth:callback", e) # user lands signed-out; know why | |
| return RedirectResponse("/") | |
| def logout(request: Request): | |
| request.session.clear() | |
| return RedirectResponse("/") | |
| def api_me(request: Request): | |
| sid = _sid(request) # per-browser id (set on first visit) | |
| u = request.session.get("user") | |
| if not u: | |
| return {"user": None, "oauth": _oauth is not None, "admin": False, "sid": sid} | |
| return {"user": u, "profile": _load_profile(u["username"]), "oauth": _oauth is not None, | |
| "admin": _is_admin(request), "sid": sid} | |
| class ProfileUpdate(BaseModel): | |
| display_name: str | None = None | |
| affiliation: str | None = None | |
| email: str | None = None | |
| bio: str | None = None | |
| def api_profile(request: Request, p: ProfileUpdate): | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| prof = _load_profile(u["username"]) | |
| prof.update({k: v for k, v in p.model_dump().items() if v is not None}) | |
| with open(_profile_path(u["username"]), "w", encoding="utf-8") as fh: | |
| json.dump(prof, fh, indent=1) | |
| persist.poke() | |
| return {"ok": True, "profile": prof} | |
| # ---- registered-user benefit: focused basins (favorite gauges) ------------- | |
| MAX_FAVORITES = 5 | |
| def _fav_pins(ids: list) -> list: | |
| """Favorite gauge ids -> map pins with catalog metadata (unknown ids skipped).""" | |
| cat = gauges.load_catalog() | |
| out = [] | |
| for gid in ids: | |
| row = cat.loc[cat.STAID == gid] | |
| if row.empty: | |
| continue | |
| r = row.iloc[0] | |
| out.append({"id": gid, "name": str(r.STANAME), "lat": float(r.LAT_GAGE), | |
| "lon": float(r.LNG_GAGE), "area_km2": float(r.DRAIN_SQKM)}) | |
| return out | |
| def api_favorites(request: Request): | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| ids = _load_profile(u["username"]).get("favorites", []) | |
| return {"favorites": _fav_pins(ids), "max": MAX_FAVORITES} | |
| class FavoriteReq(BaseModel): | |
| gauge_id: str | |
| action: str = "add" # add | remove | |
| def api_favorites_post(request: Request, f: FavoriteReq): | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "Sign in to save favorite gauges."}, status_code=401) | |
| digits = "".join(c for c in f.gauge_id if c.isdigit()) | |
| if not digits: | |
| return JSONResponse({"error": "Enter a USGS gauge number, e.g. 08167000."}, | |
| status_code=422) | |
| gid = digits.zfill(8) # catalog STAIDs are zero-padded to >= 8 digits | |
| prof = _load_profile(u["username"]) | |
| favs = [g for g in prof.get("favorites", []) if g] | |
| if f.action == "remove": | |
| favs = [g for g in favs if g != gid] | |
| elif gid not in favs: | |
| if gauges.get_gauge_coordinates(gid) is None: | |
| return JSONResponse( | |
| {"error": f"Gauge {gid} isn't in the GAGES-II catalog — " | |
| "double-check the USGS station number."}, status_code=404) | |
| if len(favs) >= MAX_FAVORITES: | |
| return JSONResponse( | |
| {"error": f"You already have {MAX_FAVORITES} focused basins — " | |
| "remove one first."}, status_code=409) | |
| favs.append(gid) | |
| prof["favorites"] = favs | |
| with open(_profile_path(u["username"]), "w", encoding="utf-8") as fh: | |
| json.dump(prof, fh, indent=1) | |
| persist.poke() # favorites are account data — sync soon | |
| return {"ok": True, "favorites": _fav_pins(favs), "max": MAX_FAVORITES} | |
| # ---- map-first gauge pins --------------------------------------------------- | |
| def _pins_for_bbox(w: float, s: float, e: float, n: float, limit: int = 300): | |
| cat = gauges.load_catalog((w, s, e, n)) | |
| if len(cat) > limit: # keep the map responsive | |
| cx, cy = (w + e) / 2, (s + n) / 2 | |
| cat = cat.assign(_d=((cat.LAT_GAGE - cy) ** 2 + (cat.LNG_GAGE - cx) ** 2)).nsmallest(limit, "_d") | |
| from hf_data import gaugetz | |
| return [{ | |
| "id": str(r.STAID).zfill(8), "name": str(r.STANAME), | |
| "lat": float(r.LAT_GAGE), "lon": float(r.LNG_GAGE), | |
| "area_km2": float(r.DRAIN_SQKM), | |
| "tz": gaugetz.tz_of(str(r.STAID).zfill(8)), | |
| } for r in cat.itertuples()] | |
| def api_gauges(w: float, s: float, e: float, n: float): | |
| """USGS gauge pins for the current viewport — works with zero AI interaction. | |
| vp_pins: ungauged virtual points (HydroBASINS pour points) that fill the | |
| coverage gaps — hindcast-simulatable via upstream-gauge injection.""" | |
| pins = _pins_for_bbox(w, s, e, n) | |
| try: | |
| from hf_data import virtualpoints | |
| vps = virtualpoints.for_bbox(w, s, e, n) | |
| except Exception: | |
| vps = [] | |
| return {"gauge_pins": pins, "vp_pins": vps, | |
| "n_gauges": len(pins), "max_sims": MAX_SIMS} | |
| class Query(BaseModel): | |
| query: str | |
| radius_deg: float = 1.0 | |
| def api_query(q: Query): | |
| """Parse the query into a map view + event pins + nearby USGS gauge pins.""" | |
| try: | |
| ctx = parse_query(q.query) | |
| except ValueError as e: | |
| return JSONResponse({"detail": str(e)}, status_code=422) | |
| lat, lon = ctx.anchor | |
| r = q.radius_deg | |
| bbox = [lon - r, lat - r, lon + r, lat + r] | |
| gauge_pins = _pins_for_bbox(*bbox) | |
| event_pins = [{"lat": lat, "lon": lon, "label": ctx.label}] | |
| return { | |
| "label": ctx.label, "center": [lat, lon], "bbox": bbox, | |
| "t_start": ctx.t_start.isoformat(), "t_end": ctx.t_end.isoformat(), | |
| "time_known": ctx.time_known, | |
| "event_pins": event_pins, "gauge_pins": gauge_pins, | |
| "n_gauges": len(gauge_pins), "max_sims": MAX_SIMS, | |
| } | |
| class FeedbackReq(BaseModel): | |
| text: str | |
| contact: str | None = None | |
| def api_feedback(req: FeedbackReq, request: Request): | |
| """Test-user improvement comments — recorded locally + persisted to the HF | |
| dataset for the daily review job.""" | |
| text = (req.text or "").strip() | |
| if not text: | |
| return JSONResponse({"error": "empty comment"}, status_code=422) | |
| from hf_data import feedback | |
| u = request.session.get("user") | |
| rec = feedback.record(text, user=u["username"] if u else None, | |
| contact=req.contact, | |
| context={"ua": request.headers.get("user-agent", "")[:120]}) | |
| return {"ok": True, "id": rec["id"]} | |
| def api_feedback_list(n: int = 100): | |
| from hf_data import feedback | |
| return {"feedback": feedback.recent(min(n, 500))} | |
| class ChatReq(BaseModel): | |
| message: str | |
| history: list = [] | |
| context: dict = {} | |
| def api_chat(req: ChatReq): | |
| """Conversational agent turn: guides users to an event, answers questions | |
| about the current simulation, and routes locate/set_time actions. | |
| {"action": "fallback"} tells the client to use its rule-based path.""" | |
| from hf_data import chatagent | |
| ctx = dict(req.context or {}) | |
| try: # live risk summary so the agent can route | |
| from hf_data import nowcaststore | |
| hs = nowcaststore.hotspots() # "hotspot" questions with no location | |
| if hs.get("ok"): | |
| ctx["nowcast_risk"] = { | |
| "t0": hs["t0"], | |
| "n_hotspots": hs["n_hotspots"], # true total (dynamic, may be 0) | |
| "hotspots": [{k: h[k] for k in ("center", "score", "n_gauges", | |
| "n_flood", "n_minor", "n_elevated")} | |
| for h in hs["hotspots"][:10]]} # top 10 for context | |
| except Exception: | |
| pass | |
| try: | |
| d = chatagent.respond(req.message, req.history, ctx) | |
| except Exception as e: | |
| crashlog.capture("chat", e, message_text=req.message[:200]) | |
| return {"action": "fallback", "error": str(e)} | |
| return d if d else {"action": "fallback"} | |
| class ExplainReq(BaseModel): | |
| error: str | |
| where: str | None = None # e.g. "simulation" | "calibration" | |
| context: dict = {} | |
| def api_explain(req: ExplainReq): | |
| """Translate a backend error into a human-readable chat explanation.""" | |
| from hf_data import llm | |
| if not llm.available(): | |
| return {"text": None} | |
| try: | |
| sys_p = ("You are the assistant in a flood-simulation dashboard (CREST/EF5 " | |
| "hydrologic model, data streamed from Hugging Face). A backend step " | |
| "failed. Explain to a NON-developer user in 2-4 short sentences: " | |
| "(1) what went wrong in plain words, (2) the most likely cause, " | |
| "(3) what to do — e.g. simply try again (transient network reads are " | |
| "common), pick a different time period/gauge, or use the 💡 Feedback " | |
| "button to report it. No stack traces, no jargon, no blame.") | |
| user_p = (f"Failed step: {req.where or 'simulation'}\n" | |
| f"Raw error: {req.error[:1500]}\n" | |
| f"Context: {str(req.context)[:500]}") | |
| text, provider = llm.chat([{"role": "system", "content": sys_p}, | |
| {"role": "user", "content": user_p}], | |
| temperature=0.2) | |
| return {"text": text, "provider": provider} | |
| except Exception as e: | |
| crashlog.capture("explain", e) | |
| return {"text": None} | |
| class EventInfoReq(BaseModel): | |
| label: str | |
| t_start: str | |
| t_end: str | |
| def api_eventinfo(req: EventInfoReq): | |
| """LLM brief about the event (damage, fatalities, links) for the AI info feed.""" | |
| from hf_data import llm | |
| if not llm.available(): | |
| return {"text": "_(no LLM configured — event background unavailable)_", "provider": None} | |
| try: | |
| text, provider = llm.event_brief(req.label, req.t_start[:10], req.t_end[:10]) | |
| return {"text": text, "provider": provider} | |
| except Exception as e: | |
| return {"text": f"_(event lookup failed: {e})_", "provider": None} | |
| class SimRequest(BaseModel): | |
| gauge_ids: list[str] | |
| t_start: str | None = None | |
| t_end: str | None = None | |
| hours: int = 48 | |
| model: str = "auto" | |
| snow: str = "auto" | |
| timestep: str = "1h" | |
| warmup_days: int = 90 | |
| overrides: dict | None = None | |
| label: str | None = None # event label for the history entry | |
| prev_sim_id: str | None = None # caller's previous job — superseded (cancelled) | |
| prev_tok: str | None = None # ... only with its token / from the same browser | |
| scheme: str = "full" # "full" = whole basin; "speed" = domain | |
| # truncated at boundary gauges (obs = inflow) | |
| MAX_HISTORY = 20 | |
| def _save_history(username: str, entry: dict): | |
| """Registered-user benefit: keep their simulation history server-side.""" | |
| prof = _load_profile(username) | |
| hist = [h for h in prof.get("history", []) if h.get("sim_id") != entry["sim_id"]] | |
| hist.insert(0, entry) | |
| prof["history"] = hist[:MAX_HISTORY] | |
| with open(_profile_path(username), "w", encoding="utf-8") as fh: | |
| json.dump(prof, fh, indent=1) | |
| # generous ceiling: a full water year fits; only guards the shared demo CPU | |
| # against extreme multi-year requests (0 disables the cap entirely) | |
| MAX_SIM_HOURS = int(os.environ.get("CREST_MAX_SIM_HOURS", "8784")) # 366 days | |
| def api_simulate(req: SimRequest, request: Request): | |
| warnings = [] | |
| if len(req.gauge_ids) > MAX_SIMS: | |
| warnings.append(f"This demo simulates at most {MAX_SIMS} gauges at once; " | |
| f"running the first {MAX_SIMS} of {len(req.gauge_ids)} selected.") | |
| t0 = datetime.fromisoformat(req.t_start) if req.t_start else datetime(2025, 7, 3) | |
| t1 = datetime.fromisoformat(req.t_end) if req.t_end else t0 + timedelta(hours=req.hours) | |
| if t1 <= t0: # nonsensical window | |
| t1 = t0 + timedelta(hours=req.hours) | |
| warnings.append(f"End time was not after the start — using {req.hours} h instead.") | |
| if MAX_SIM_HOURS > 0 and (t1 - t0) > timedelta(hours=MAX_SIM_HOURS): | |
| t1 = t0 + timedelta(hours=MAX_SIM_HOURS) # shared-CPU sanity ceiling | |
| warnings.append(f"Simulation window capped at {MAX_SIM_HOURS // 24} days " | |
| f"(now ends {t1:%Y-%m-%d %H:%M}).") | |
| # supersede the caller's previous run: a stale in-flight job would otherwise | |
| # hold the per-gauge lock and the new run would queue behind it indefinitely | |
| owner = _owner(request) | |
| if req.prev_sim_id: | |
| prev = simjobs.get_job(req.prev_sim_id) | |
| if prev and not prev.done.is_set() and _owns(prev, request, req.prev_tok or "", | |
| strict=True): | |
| prev.cancel.set() | |
| warnings.append("Your previous simulation was still running — " | |
| "it was stopped and replaced by this one.") | |
| opts = {"model": req.model, "hours": req.hours, "snow": req.snow, | |
| "timestep": req.timestep, "warmup_days": req.warmup_days, | |
| "overrides": req.overrides, | |
| "scheme": req.scheme if req.scheme in ("full", "speed") else "full"} | |
| job = simjobs.start_job(req.gauge_ids, t0, t1, opts, owner=owner) | |
| u = request.session.get("user") | |
| if u: # signed-in -> history entry | |
| try: | |
| _save_history(u["username"], { | |
| "sim_id": job.id, "gauge_ids": job.gauge_ids, | |
| "t_start": t0.isoformat(), "t_end": t1.isoformat(), | |
| "label": req.label, "model": req.model, "snow": req.snow, | |
| "when": datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")}) | |
| except Exception as e: | |
| crashlog.capture("history:save", e, user=u.get("username")) | |
| return {"sim_id": job.id, "token": owner["token"], "gauge_ids": job.gauge_ids, | |
| "t_start": t0.isoformat(), "t_end": t1.isoformat(), | |
| "warning": " ".join(warnings) or None, | |
| "max_concurrent": simjobs.MAX_CONCURRENT, | |
| "per_user": simjobs.PER_USER_CONCURRENT} | |
| def api_nowcast(sim_id: str, gauge_id: str, request: Request, tok: str = ""): | |
| """AI nowcast tail (+6 h) from the CREST_nowcast Space for a finished | |
| gauge. Always answers; {"ok": false} just means no tail is shown.""" | |
| job = simjobs.get_job(sim_id) | |
| if not job: | |
| return JSONResponse({"ok": False, "reason": "unknown job"}, status_code=404) | |
| if not _owns(job, request, tok): | |
| return JSONResponse({"ok": False, **_FORBIDDEN}, status_code=403) | |
| from hf_data import nowcast as _nc | |
| return _nc.for_job(job, gauge_id) | |
| def api_nowcast_risk(): | |
| """CONUS-wide flood-risk snapshot for Nowcast mode: which gauges' AI | |
| next-6-h peak exceeds their 10-yr return flood (red density layer / | |
| red pins).""" | |
| from hf_data import nowcaststore | |
| return nowcaststore.all_risk() | |
| def api_nowcast_now(w: float, s: float, e: float, n: float, limit: int = 100, | |
| obs_hours: int = 0, ids: str = ""): | |
| """Precomputed fleet nowcasts for every gauge in the viewport/rectangle | |
| (or an explicit `ids` comma list) — instant (the updater Space refreshes | |
| them hourly), no simulation needed. obs_hours>0 attaches recent observed | |
| series (<=25 gauges) for plotting.""" | |
| from hf_data import nowcaststore | |
| return nowcaststore.for_bbox(w, s, e, n, limit, obs_hours, ids) | |
| def api_ungauged_now(w: float, s: float, e: float, n: float, limit: int = 200, | |
| ids: str = ""): | |
| """Precomputed ungauged routed nowcasts for the viewport/rectangle (or an | |
| explicit `ids` comma list) — served instantly from the keep-warm Space's | |
| hourly parquet, no live EF5 run. Each point carries its ~7-day routed | |
| history + 12-h forecast so the panel plots the split hydrograph directly.""" | |
| from hf_data import ungaugednow_store | |
| return (ungaugednow_store.by_ids(ids) if ids | |
| else ungaugednow_store.for_bbox(w, s, e, n, limit)) | |
| def api_mrms_frames(): | |
| """Nowcast-mode rain-radar animation: which MRMS Pass1 PNG frames exist | |
| (rolling 7 days, rendered hourly by the updater) + the CDN base URL.""" | |
| from hf_data import mrmsframes | |
| return mrmsframes.frames() | |
| def api_nowcast_hotspots(): | |
| """Ranked spatial clusters of flagged flood-risk gauges — the chat agent's | |
| answer to "bring me to the flood risk hotspot" (no location needed).""" | |
| from hf_data import nowcaststore | |
| return nowcaststore.hotspots() | |
| def api_nowcast_brief(refresh: int = 0): | |
| """Nowcast-mode opening card: structured CONUS flood snapshot (flagged | |
| gauges by state, hotspots, rain leaders, 2-D events) + the LLM's national | |
| summary cross-checked against current news (web search when an OpenAI key | |
| is configured). Cached per hourly issue; `building` => poll again.""" | |
| from hf_data import floodbrief | |
| return floodbrief.brief(force=bool(refresh)) | |
| # V30 runner/dashboard split: when EVENT_RUNNER_URL points at the runner | |
| # Space, THIS Space is the pure-reader dashboard — event triggers proxy | |
| # there and the runner's live status is merged into /api/events. UI deploys | |
| # then never kill an in-flight solve or reset the tick clock. | |
| EVENT_RUNNER_URL = os.environ.get("EVENT_RUNNER_URL", "").rstrip("/") | |
| _runner_cache: dict = {"t": 0.0, "status": None} | |
| def _remote_runner_status(): | |
| import time as _time | |
| if not EVENT_RUNNER_URL: | |
| return None | |
| if _time.time() - _runner_cache["t"] < 20: | |
| return _runner_cache["status"] | |
| try: | |
| import requests | |
| r = requests.get(f"{EVENT_RUNNER_URL}/api/events?full=0", | |
| timeout=15).json() | |
| _runner_cache.update(t=_time.time(), status=r.get("runner")) | |
| except Exception: | |
| _runner_cache.update(t=_time.time(), status=None) | |
| return _runner_cache["status"] | |
| def api_events(full: int = 0): | |
| """V25: published 2-D inundation events (CREST-iMAP v2) + runner state. | |
| Frame/manifest files resolve at {base}/<event_id>/<file>.""" | |
| from hf_data import eventqueue, eventsim, eventstore | |
| try: | |
| q = eventqueue.queue_status() | |
| except Exception: | |
| q = [] | |
| runner = _remote_runner_status() or eventsim.status(bool(full)) | |
| try: # CREST misses: trigger fired, EF5 did not | |
| missed = eventstore.load_missed()[-30:] | |
| except Exception: | |
| missed = [] | |
| try: # auto-calibrations those misses triggered | |
| autocal = eventstore.load_autocal()[-10:] | |
| except Exception: | |
| autocal = [] | |
| return {"events": eventstore.load_index(), "queue": q, | |
| "runner": runner, "missed": missed, "autocal": autocal, | |
| "base": (f"https://huggingface.co/datasets/{eventstore.REPO}/" | |
| f"resolve/main/{eventstore.PREFIX}")} | |
| def api_admin_archive(request: Request, refresh: int = 0): | |
| """Admin only: every stored 2-D inundation result — the listed events plus | |
| those that rolled off the public list (results kept RESULTS_MAX_AGE_D).""" | |
| if not _is_admin(request): | |
| return JSONResponse({"error": "admin only"}, status_code=403) | |
| from hf_data import eventstore | |
| return eventstore.archive_index(force=bool(refresh)) | |
| def api_riskprobe(event_id: str, lat: float, lon: float, t: str = ""): | |
| """Risk-view lens probe: per-PIXEL depth/speed/flow-direction, DEFRA | |
| hazard class, WorldPop 2026-2030 population, and OSM infrastructure | |
| values for the solver cell at (lat, lon) of a published event. `t` | |
| (frame timestamp from the manifest) reads that frame's depth; default | |
| is the episode max. Returns the pixel's footprint for the highlight.""" | |
| from hf_data import riskprobe | |
| try: | |
| return riskprobe.probe(event_id, lat, lon, t) | |
| except Exception as e: | |
| crashlog.capture("riskprobe", e, event=event_id) | |
| return JSONResponse({"error": f"{type(e).__name__}: {e}"}, | |
| status_code=500) | |
| def api_health(): | |
| """V30 pipeline self-monitoring: nowcast/radar freshness, MRMS Pass2 / | |
| PET / TEMP archive freshness, event-tick liveness, queue depth + oldest | |
| unclaimed age, worker heartbeats, event throughput — each with an ok | |
| flag. The daily review routine leads its digest with this.""" | |
| from hf_data import health | |
| return health.snapshot() | |
| def api_event_tick(): | |
| """Hourly auto-trigger (pinged by the updater after each data refresh): | |
| re-simulates every ACTIVE inundation episode at the fresh t0 and starts | |
| newly flagged ones. An episode ends when observed flow recedes to | |
| normal, NOT when precipitation stops. On the split dashboard | |
| (EVENT_TICK=off) this forwards to the runner Space.""" | |
| if os.environ.get("EVENT_TICK", "1") != "1": | |
| if EVENT_RUNNER_URL: | |
| try: | |
| import requests | |
| return requests.post(f"{EVENT_RUNNER_URL}/api/event_tick", | |
| timeout=30).json() | |
| except Exception as e: | |
| return JSONResponse({"ok": False, "reason": | |
| f"runner unreachable: {type(e).__name__}"}, | |
| status_code=502) | |
| return {"ok": False, "reason": "event tick disabled on this Space"} | |
| try: | |
| import crestimap # noqa: F401 | |
| except ImportError: | |
| return JSONResponse({"ok": False, "reason": | |
| "crestimap not installed on this Space build"}, | |
| status_code=501) | |
| from hf_data import eventsim | |
| if eventsim.status()["running"]: | |
| # stamp liveness even here: the ping IS arriving, we are just busy. | |
| # Without this a wedged runner looked identical to a dead tick loop. | |
| eventsim.note_tick_attempt() | |
| return {"ok": False, "reason": "runner busy", | |
| "running": eventsim.status()["running"], | |
| "busy_h": eventsim._busy_h()} | |
| import threading | |
| threading.Thread(target=eventsim.hourly_tick, daemon=True).start() | |
| return {"ok": True, "started": "tick"} | |
| def api_event_run(gid: str = "", t0: str = "", defer: int = 0, | |
| episode: str = "", cold: int = 0): | |
| """V25 manual/agent trigger: 2-D inundation event for `gid`, or | |
| auto-detect from the nowcast flood tier when gid is empty. The run takes | |
| minutes (EF5 + hydrodynamic solver) and executes in a worker thread; | |
| poll /api/events for progress. | |
| t0: backdate the event window ("YYYY-MM-DDTHH:MM" UTC) — refill a flood | |
| the trigger pipeline missed; the window is [t0-backset, t0+horizon]. | |
| defer=1: enqueue for the GPU worker only (no CPU fallback) — the runner | |
| stays free and the result publishes whenever the worker gets to it. | |
| episode=<event id>: re-simulate an EXISTING listed episode under its own | |
| id (anchored at its recorded start; the folder is replaced in place) — | |
| e.g. re-running the whole list after an engine/domain change. | |
| cold=1 (with episode): no warm-start frame — the whole anchored span is | |
| re-solved and every published frame replaced.""" | |
| if os.environ.get("EVENT_TICK", "1") != "1" and EVENT_RUNNER_URL: | |
| try: # split dashboard: forward to the runner | |
| import requests | |
| return requests.post( | |
| f"{EVENT_RUNNER_URL}/api/event_run", | |
| params={"gid": gid, "t0": t0, "defer": defer, | |
| "episode": episode, "cold": cold}, | |
| timeout=30).json() | |
| except Exception as e: | |
| return JSONResponse({"ok": False, "reason": | |
| f"runner unreachable: {type(e).__name__}"}, | |
| status_code=502) | |
| try: | |
| import crestimap # noqa: F401 (installed from the CREST-iMAP fork, v2) | |
| except ImportError: | |
| return JSONResponse({"ok": False, "reason": | |
| "crestimap not installed on this Space build"}, | |
| status_code=501) | |
| from hf_data import eventsim | |
| st = eventsim.status() | |
| if st["running"]: | |
| return {"ok": False, "reason": "an event simulation is already running", | |
| **st} | |
| t0_dt = None | |
| if t0: | |
| import datetime as _dt | |
| try: | |
| t0_dt = _dt.datetime.strptime(t0, "%Y-%m-%dT%H:%M") | |
| except ValueError: | |
| return JSONResponse({"ok": False, | |
| "reason": "t0 must be YYYY-MM-DDTHH:MM (UTC)"}, | |
| status_code=422) | |
| import threading | |
| if episode and not gid: | |
| gid = episode.split("_", 1)[-1] # ids are <YYYYMMDDHH>_<gid> | |
| if gid: | |
| threading.Thread(target=eventsim.run_one, args=(gid, t0_dt), | |
| kwargs={"defer": bool(defer), | |
| "episode_id": episode or None, | |
| "cold": bool(cold and episode)}, | |
| daemon=True).start() | |
| else: | |
| threading.Thread(target=eventsim.run_detected, daemon=True).start() | |
| return {"ok": True, "started": gid or "auto-detect", "t0": t0 or "now", | |
| "defer": bool(defer), "episode": episode or None, | |
| "cold": bool(cold and episode)} | |
| def api_upstream(gid: str): | |
| """River network upstream of a gauge (HydroRIVERS topology walk) — drawn | |
| on the Nowcast map when the gauge is focused. Static data, LRU-cached. | |
| JSONResponse directly: jsonable_encoder walking ~50k coord floats costs | |
| ~2 s; plain json.dumps is ~50 ms.""" | |
| from hf_data import rivernet | |
| return JSONResponse(rivernet.upstream(gid)) | |
| def api_cancel(sim_id: str, request: Request, tok: str = ""): | |
| """Stop a running simulation job: the EF5 processes are killed and the | |
| per-gauge run locks released so a new run can start immediately. Only | |
| the browser that started it (or holds its token) may stop it.""" | |
| job = simjobs.get_job(sim_id) | |
| if not job: | |
| return JSONResponse({"error": "unknown job"}, status_code=404) | |
| if not _owns(job, request, tok, strict=True): | |
| return JSONResponse(_FORBIDDEN, status_code=403) | |
| if job.done.is_set(): | |
| return {"ok": True, "already_done": True} | |
| job.cancel.set() | |
| return {"ok": True} | |
| def api_history(request: Request): | |
| """Signed-in users: their saved simulations, with live job status.""" | |
| u = request.session.get("user") | |
| if not u: | |
| return JSONResponse({"error": "not signed in"}, status_code=401) | |
| hist = _load_profile(u["username"]).get("history", []) | |
| sid = _sid(request) | |
| for h in hist: | |
| job = simjobs.get_job(h.get("sim_id", "")) | |
| h["status"] = ("done" if job and job.done.is_set() else | |
| "running" if job else "expired") # expired -> cache re-run | |
| # started from THIS browser? (a shared account sees the others' runs | |
| # listed, but only its own browser follows them automatically) | |
| h["this_browser"] = bool(job) and (getattr(job, "owner", None) or {}).get("sid") == sid | |
| return {"history": hist} | |
| def _sse_json(ev) -> str: | |
| """json.dumps emits bare NaN/Infinity, which JSON.parse rejects — one such | |
| value breaks the whole event stream in the browser. Missing values travel as | |
| null, matching the convention the rest of the pipeline already uses.""" | |
| def clean(v): | |
| if isinstance(v, float) and (math.isnan(v) or math.isinf(v)): | |
| return None | |
| if isinstance(v, dict): | |
| return {k: clean(x) for k, x in v.items()} | |
| if isinstance(v, (list, tuple)): | |
| return [clean(x) for x in v] | |
| return v | |
| return json.dumps(clean(ev), allow_nan=False) | |
| async def _drain(job, cursor: int = 0): | |
| """Replay the job's event log from `cursor`, then follow it live. Multiple | |
| clients (or a reopened browser) can each attach with their own cursor.""" | |
| try: | |
| while True: | |
| if cursor < len(job.events): | |
| ev = job.events[cursor] | |
| cursor += 1 | |
| yield {"data": _sse_json(ev)} | |
| if ev.get("kind") in ("all_done", "cal_done"): | |
| break | |
| elif job.done.is_set(): | |
| break | |
| else: | |
| await asyncio.sleep(0.15) | |
| except asyncio.CancelledError: # client closed the tab — not an error | |
| raise | |
| except Exception as e: # SSE runs outside the HTTP middleware | |
| crashlog.capture("sse:drain", e, job=getattr(job, "id", "?"), cursor=cursor) | |
| async def api_stream(sim_id: str, request: Request, cursor: int = 0, tok: str = ""): | |
| job = simjobs.get_job(sim_id) | |
| if not job: | |
| return Response(status_code=404) | |
| if not _owns(job, request, tok): | |
| return Response(status_code=403) | |
| return EventSourceResponse(_drain(job, cursor)) | |
| def api_job(sim_id: str, request: Request, tok: str = ""): | |
| """Job descriptor for reattaching after the browser was closed. The run | |
| keeps going server-side; the client replays the event log via /api/stream. | |
| Another browser session gets 403 and forgets the id.""" | |
| job = simjobs.get_job(sim_id) | |
| if not job: | |
| return JSONResponse({"error": "unknown or expired job"}, status_code=404) | |
| if not _owns(job, request, tok): | |
| return JSONResponse(_FORBIDDEN, status_code=403) | |
| return {"sim_id": job.id, "gauge_ids": job.gauge_ids, | |
| "t_start": job.t_start.isoformat(), "t_end": job.t_end.isoformat(), | |
| "done": job.done.is_set(), "n_events": len(job.events), | |
| "age_s": int(time.time() - job.created)} | |
| class CalRequest(BaseModel): | |
| gauge_id: str | |
| t_start: str | |
| t_end: str | |
| model: str = "auto" | |
| snow: str = "auto" | |
| # budget = rounds*k candidate runs + 1 baseline (defaults: 4*3+1 = 13 EF5 | |
| # runs). If the best NSE is still < 0.3 afterwards, calibrate.py appends | |
| # one extended stage of 5 rounds x 4 candidates (+20 runs) automatically. | |
| # Space variables: CREST_CAL_ROUNDS / CREST_CAL_K (standard stage) and | |
| # CREST_CAL_EXT_ROUNDS / CREST_CAL_EXT_K / CREST_CAL_EXT_NSE (extension) — | |
| # no redeploy needed (candidates warm-start from the saved state, so each | |
| # extra run costs seconds-to-minutes, not a full warm-up). | |
| rounds: int = int(os.environ.get("CREST_CAL_ROUNDS", "4")) | |
| k: int = int(os.environ.get("CREST_CAL_K", "3")) | |
| def api_calibrate(req: CalRequest, request: Request): | |
| from hf_data import virtualpoints | |
| if virtualpoints.is_virtual(req.gauge_id): | |
| return JSONResponse({"error": "ungauged points have no observations — " | |
| "calibration is not possible"}, status_code=422) | |
| owner = _owner(request) | |
| job = caljobs.start_job(req.gauge_id, | |
| datetime.fromisoformat(req.t_start), | |
| datetime.fromisoformat(req.t_end), | |
| {"model": req.model, "snow": req.snow, | |
| "rounds": req.rounds, "k": req.k}, owner=owner) | |
| return {"cal_id": job.id, "token": owner["token"], "gauge_id": req.gauge_id} | |
| async def api_calstream(cal_id: str, request: Request, cursor: int = 0, tok: str = ""): | |
| job = caljobs.get_job(cal_id) | |
| if not job: | |
| return Response(status_code=404) | |
| if not _owns(job, request, tok): | |
| return Response(status_code=403) | |
| return EventSourceResponse(_drain(job, cursor)) | |
| def api_overlay(sim_id: str, gauge_id: str): | |
| job = simjobs.get_job(sim_id) | |
| png = job.overlay_png(gauge_id) if job else None | |
| if png is None: | |
| return Response(status_code=404) | |
| return Response(content=png, media_type="image/png", | |
| headers={"Cache-Control": "no-store"}) | |
| def api_frame(sim_id: str, gauge_id: str, idx: int): | |
| """A single pre-rendered animation frame (fixed color scale).""" | |
| job = simjobs.get_job(sim_id) | |
| png = job.frame_png(gauge_id, idx) if job else None | |
| if png is None: | |
| return Response(status_code=404) | |
| return Response(content=png, media_type="image/png", | |
| headers={"Cache-Control": "max-age=3600"}) | |
| if os.path.isdir(FRONTEND): | |
| app.mount("/static", StaticFiles(directory=FRONTEND), name="static") | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |