"""Parameter endpoints - Seven hydrodynamic parameters""" from fastapi import APIRouter, HTTPException, Query, Path from typing import Dict, List, Optional from datetime import datetime router = APIRouter() # Parameter definitions PARAMETERS = { "wcc": { "name": "Wave Front Celerity Coefficient", "description": "Measures departure from linear propagation speed", "unit": "dimensionless", "safe": "< 1.35", "monitor": "1.35 - 1.50", "alert": "1.50 - 1.58", "critical": "> 1.58" }, "kpr": { "name": "Kinetic-to-Potential Energy Transfer Ratio", "description": "Tracks energy partition and bore formation", "unit": "dimensionless", "safe": "< 1.2", "monitor": "1.2 - 1.6", "alert": "1.6 - 2.0", "critical": "> 2.0" }, "hfsi": { "name": "Hydrodynamic Front Stability Index", "description": "Quantifies wave front stability via Boussinesq parameter", "unit": "dimensionless", "safe": "> 0.80", "monitor": "0.60 - 0.80", "alert": "0.40 - 0.60", "critical": "< 0.40" }, "becf": { "name": "Bathymetric Energy Concentration Factor", "description": "Energy amplification by convergent bathymetry", "unit": "dimensionless", "safe": "< 2.0", "monitor": "2.0 - 4.0", "alert": "4.0 - 6.0", "critical": "> 6.0" }, "sdb": { "name": "Spectral Dispersion Bandwidth", "description": "Tracks spectral spreading and harmonic transfer", "unit": "dimensionless", "safe": "> 3.5", "monitor": "2.5 - 3.5", "alert": "1.0 - 2.5", "critical": "< 1.0" }, "sbsp": { "name": "Shoreline Boundary Stress Parameter", "description": "Estimates inundation momentum flux", "unit": "dimensionless", "safe": "< 0.3", "monitor": "0.3 - 0.7", "alert": "0.7 - 1.2", "critical": "> 1.2" }, "smvi": { "name": "Sub-Surface Micro-Vorticity Index", "description": "Detects vorticity at bathymetric slope breaks", "unit": "dimensionless", "safe": "< 0.2", "monitor": "0.2 - 0.4", "alert": "0.4 - 0.6", "critical": "> 0.6" } } @router.get("/", response_model=Dict) async def get_parameters_list(): """Get list of all seven parameters with descriptions""" return { "parameters": PARAMETERS, "count": len(PARAMETERS) } @router.get("/{param_name}", response_model=Dict) async def get_parameter_info( param_name: str = Path(..., description="Parameter name (wcc, kpr, hfsi, becf, sdb, sbsp, smvi)") ): """Get detailed information about a specific parameter""" if param_name not in PARAMETERS: raise HTTPException(status_code=404, detail="Parameter not found") return PARAMETERS[param_name] @router.get("/compute/{param_name}", response_model=Dict) async def compute_parameter( param_name: str = Path(...), H: float = Query(..., description="Water depth [m]"), eta: float = Query(..., description="Wave amplitude [m]"), wavelength: Optional[float] = Query(None, description="Wavelength [m]"), u: Optional[float] = Query(None, description="Velocity [m/s]") ): """Compute a specific parameter from inputs""" if param_name == "wcc": if H is None or eta is None or wavelength is None: raise HTTPException(status_code=400, detail="Need H, eta, and wavelength") g = 9.81 c0 = (g * H) ** 0.5 c_nl = c0 * (1 + (3*eta)/(4*H) - (3.1416**2 * H**2)/(6 * wavelength**2)) wcc = c_nl / c0 return { "parameter": "wcc", "value": wcc, "status": get_status("wcc", wcc), "inputs": {"H": H, "eta": eta, "wavelength": wavelength}, "c0": c0, "c_nl": c_nl } elif param_name == "kpr": if H is None or u is None or eta is None: raise HTTPException(status_code=400, detail="Need H, u, and eta") g = 9.81 kpr = (H * u**2) / (g * eta**2) return { "parameter": "kpr", "value": kpr, "status": get_status("kpr", kpr), "inputs": {"H": H, "u": u, "eta": eta} } elif param_name == "hfsi": if H is None or eta is None or wavelength is None: raise HTTPException(status_code=400, detail="Need H, eta, and wavelength") Bo = H**3 / (eta * wavelength**2) hfsi = np.tanh(Bo) return { "parameter": "hfsi", "value": hfsi, "status": get_status("hfsi", hfsi), "inputs": {"H": H, "eta": eta, "wavelength": wavelength}, "boussinesq": Bo } else: raise HTTPException(status_code=400, detail=f"Computation for {param_name} not implemented") @router.get("/zone/{zone_name}", response_model=Dict) async def get_zone_parameters( zone_name: str = Path(...), event_id: Optional[str] = Query(None, description="Event ID for context") ): """Get current parameters for a specific coastal zone""" # Placeholder for zone parameters zone_params = { "hilo_bay_hawaii": { "wcc": 1.31, "kpr": 1.44, "hfsi": 0.63, "becf": 4.80, "sdb": 1.20, "sbsp": 0.67, "smvi": 0.29 }, "miyako_japan": { "wcc": 1.56, "kpr": 1.89, "hfsi": 0.31, "becf": 7.30, "sdb": 0.80, "sbsp": 1.18, "smvi": 0.38 } } if zone_name not in zone_params: raise HTTPException(status_code=404, detail="Zone not found") params = zone_params[zone_name] return { "zone": zone_name, "timestamp": datetime.utcnow().isoformat(), "parameters": { name: { "value": value, "status": get_status(name, value) } for name, value in params.items() } } def get_status(param: str, value: float) -> str: """Get status for parameter value""" if param == "wcc": if value < 1.35: return "SAFE" elif value < 1.50: return "MONITOR" elif value < 1.58: return "ALERT" else: return "CRITICAL" elif param == "kpr": if value < 1.2: return "SAFE" elif value < 1.6: return "MONITOR" elif value < 2.0: return "ALERT" else: return "CRITICAL" elif param == "hfsi": if value > 0.8: return "SAFE" elif value > 0.6: return "MONITOR" elif value > 0.4: return "ALERT" else: return "CRITICAL" elif param == "becf": if value < 2.0: return "SAFE" elif value < 4.0: return "MONITOR" elif value < 6.0: return "ALERT" else: return "CRITICAL" elif param == "sdb": if value > 3.5: return "SAFE" elif value > 2.5: return "MONITOR" elif value > 1.0: return "ALERT" else: return "CRITICAL" elif param == "sbsp": if value < 0.3: return "SAFE" elif value < 0.7: return "MONITOR" elif value < 1.2: return "ALERT" else: return "CRITICAL" elif param == "smvi": if value < 0.2: return "SAFE" elif value < 0.4: return "MONITOR" elif value < 0.6: return "ALERT" else: return "CRITICAL" else: return "UNKNOWN"