import os
import json
from datetime import datetime
import math
from functools import lru_cache
from typing import Dict, Any, List
import numpy as np
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse, HTMLResponse
import gradio as gr
# Local import: vendored from working project
from backend.grib_wave_puller import GRIBWavePuller
app = FastAPI(title="Wave Visualizer API")
def _compute_uv_from_wave(height: np.ndarray, direction_deg: np.ndarray, scale: float = 0.1):
"""Compute U/V components from wave height and meteorological 'from' direction.
- height: significant wave height array (m)
- direction_deg: wave direction (deg, meteorological, coming from)
- scale: visualization scaling factor
"""
dir_rad = np.deg2rad(direction_deg)
mag = np.clip(height, 0, np.nanmax(height)) * scale
# Eastward (u) and northward (v) components; negative on v because 'from'
u = mag * np.sin(dir_rad)
v = -mag * np.cos(dir_rad)
return u, v
def _build_velocity_grib_json(lats: np.ndarray, lons: np.ndarray, u: np.ndarray, v: np.ndarray, ref_time: str) -> List[Dict[str, Any]]:
"""Build leaflet-velocity compatible JSON (Wind/Earth GRIB-like format).
Data must be provided on a regular lat-lon grid. Arrays are 2D with shape (ny, nx)
where ny=len(lats), nx=len(lons). Latitude should be provided in descending order
(north to south) to match common GRIB conventions; reorder if needed.
"""
# Ensure 1D coordinate arrays
lats_1d = lats if lats.ndim == 1 else lats[:, 0]
lons_1d = lons if lons.ndim == 1 else lons[0, :]
ny = int(len(lats_1d))
nx = int(len(lons_1d))
# If latitude increases northward, reverse to north->south
if ny > 1 and lats_1d[0] < lats_1d[-1]:
lats_1d = lats_1d[::-1]
u = np.flipud(u)
v = np.flipud(v)
# Normalize longitudes to [-180, 180) to avoid 0..360 grids causing clipping
# Then ensure they ascend west->east and reorder u/v columns accordingly.
if nx > 1:
lons_wrapped = ((np.asarray(lons_1d, dtype=float) + 180.0) % 360.0) - 180.0
order = np.argsort(lons_wrapped)
lons_1d = lons_wrapped[order]
if u.ndim == 2 and v.ndim == 2 and u.shape[1] == nx and v.shape[1] == nx:
u = u[:, order]
v = v[:, order]
la1 = float(lats_1d[0])
la2 = float(lats_1d[-1])
lo1 = float(lons_1d[0])
lo2 = float(lons_1d[-1])
# Grid spacing (approx)
dy = float(abs(lats_1d[1] - lats_1d[0])) if ny > 1 else 0.0
dx = float(abs(lons_1d[1] - lons_1d[0])) if nx > 1 else 0.0
# Sanitize arrays: replace NaN/Inf with zeros for JSON compliance
u = np.nan_to_num(np.asarray(u, dtype=float), nan=0.0, posinf=0.0, neginf=0.0)
v = np.nan_to_num(np.asarray(v, dtype=float), nan=0.0, posinf=0.0, neginf=0.0)
# Optional clamp to reasonable range (avoid absurd values)
# Here, clamp to [-20, 20] m/s just for safety in visualization
u = np.clip(u, -20.0, 20.0)
v = np.clip(v, -20.0, 20.0)
# Flatten row-major (lat-major first, then lon) matching header
u_data = u.flatten().tolist()
v_data = v.flatten().tolist()
header_common = {
"lo1": lo1,
"la1": la1,
"lo2": lo2,
"la2": la2,
"nx": nx,
"ny": ny,
"dx": dx,
"dy": dy,
"refTime": ref_time,
}
u_record = {
"header": {
**header_common,
"parameterCategory": 2,
"parameterNumber": 2, # U component
"parameterUnit": "m/s",
},
"data": u_data,
}
v_record = {
"header": {
**header_common,
"parameterCategory": 2,
"parameterNumber": 3, # V component
"parameterUnit": "m/s",
},
"data": v_data,
}
return [u_record, v_record]
@lru_cache(maxsize=16)
def get_puller() -> GRIBWavePuller:
return GRIBWavePuller()
@app.get("/data/points")
def data_points(hour: int = Query(0, ge=0, le=240)):
puller = get_puller()
result = puller.fetch_global_wave_data(hour)
if not result:
return JSONResponse(status_code=503, content={"error": "No data available"})
def _sanitize(obj):
if isinstance(obj, dict):
return {k: _sanitize(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_sanitize(v) for v in obj]
if isinstance(obj, (np.floating,)):
v = float(obj)
return None if not math.isfinite(v) else v
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, float):
return None if not math.isfinite(obj) else obj
return obj
payload = {
"type": "points",
"refTime": result.get("timestamp"),
"points": result.get("sample_points", []),
}
return JSONResponse(content=_sanitize(payload))
@app.get("/data/velocity")
def data_velocity(hour: int = Query(0, ge=0, le=240), scale: float = Query(0.1)):
puller = get_puller()
result = puller.fetch_global_wave_data(hour)
if not result:
return JSONResponse(status_code=503, content={"error": "No data available"})
# If we have a downsampled UV grid, return leaflet-velocity JSON
grid_uv = result.get("grid_uv")
if grid_uv:
lats = np.array(grid_uv['lats'])
lons = np.array(grid_uv['lons'])
u = np.array(grid_uv['u'])
v = np.array(grid_uv['v'])
# Validate grid content; if empty or trivial, fall back to points
if (
u.size < 16 or v.size < 16 or
not np.isfinite(u).any() or not np.isfinite(v).any() or
(np.nanmax(np.abs(u)) < 1e-6 and np.nanmax(np.abs(v)) < 1e-6)
):
sample_points = result.get("sample_points", [])
return JSONResponse(content={"type": "points", "refTime": result.get("timestamp"), "points": sample_points})
payload = _build_velocity_grib_json(lats, lons, u, v, ref_time=result.get("timestamp", datetime.utcnow().isoformat()))
return JSONResponse(content=payload)
# Fallback to points if no grid is present
sample_points = result.get("sample_points", [])
payload = {"type": "points", "refTime": result.get("timestamp"), "points": sample_points}
# Sanitize for JSON compliance
def _san(obj):
if isinstance(obj, dict):
return {k: _san(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_san(v) for v in obj]
if isinstance(obj, (np.floating,)):
v = float(obj)
return None if not math.isfinite(v) else v
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, float):
return None if not math.isfinite(obj) else obj
return obj
return JSONResponse(content=_san(payload))
def leaflet_html() -> str:
return """
"""
@app.get("/map", response_class=HTMLResponse)
def map_page():
return leaflet_html()
@app.get("/", response_class=HTMLResponse)
def root_page():
return leaflet_html()
# Optional Gradio UI under /ui
with gr.Blocks(title="Wave Visualizer UI") as demo:
gr.Markdown("# Wave Visualizer\nUse the link below to open the map page.")
gr.HTML('