Spaces:
Sleeping
Sleeping
File size: 8,276 Bytes
7b6b271 23a9367 7b6b271 23a9367 7b6b271 23a9367 7b6b271 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 |
"""
Stormglass Marine Data Tool - Premium Wave and Wind Conditions.
This module provides access to high-quality marine data from the Stormglass API,
with intelligent fallback to Open-Meteo for reliability. It fetches real-time
and forecast data including waves, wind, swell, and water temperature.
Data Sources (Priority Order):
1. Stormglass API: Premium accuracy, requires API key
2. Open-Meteo Marine: Free fallback, always available
Supported Parameters:
- Wave height, direction, and period
- Wind speed, direction, and gusts
- Swell direction and characteristics
- Water temperature (Stormglass only)
- Forecast data up to 10 days
The tool automatically handles API failures, rate limits, and
missing API keys by falling back to the free Open-Meteo service.
Example:
>>> tool = create_stormglass_tool()["function"]
>>> result = tool(WaveDataInput(lat=36.72, lon=-4.42))
>>> print(f"Wave height: {result.wave_height}m")
Author: Surf Spot Finder Team
License: MIT
"""
import os
from datetime import datetime, timedelta
from typing import Optional
from .marine_data_tool import OpenMeteoMarineAPI
import requests
from pydantic import BaseModel, Field
# -------------------------------------------------------
# Input / Output Schemas
# -------------------------------------------------------
class WaveDataInput(BaseModel):
"""Input schema for wave data requests.
Attributes:
lat: Latitude in decimal degrees (-90 to 90).
lon: Longitude in decimal degrees (-180 to 180).
target_datetime: Optional future datetime for forecast data.
If None, returns current conditions.
"""
lat: float = Field(description="Latitude of the surf spot")
lon: float = Field(description="Longitude of the surf spot")
target_datetime: Optional[datetime] = Field(
None,
description="Datetime for forecast. If none, returns current conditions."
)
class WaveDataOutput(BaseModel):
"""Output schema for marine condition data.
Attributes:
wave_height: Significant wave height in meters.
wave_direction: Wave direction in degrees (0-360).
wave_period: Wave period in seconds.
wind_speed: Wind speed in meters per second.
wind_direction: Wind direction in degrees (0-360).
swell_direction: Primary swell direction in degrees.
wind_gust: Maximum wind gust speed in m/s.
water_temperature: Water temperature in Celsius.
timestamp: Data timestamp in ISO format.
forecast_target: Target forecast time if applicable.
"""
wave_height: float
wave_direction: float
wave_period: float
wind_speed: float
wind_direction: float
swell_direction: float
wind_gust: float
water_temperature: float
timestamp: Optional[str]
forecast_target: Optional[str]
data_source: str
# -------------------------------------------------------
# Stormglass Tool
# -------------------------------------------------------
class StormglassTool:
name = "fetch_wave_data"
description = "Fetch wave, wind, swell, and marine conditions using Stormglass API"
BASE_URL = "https://api.stormglass.io/v2/weather/point"
PARAMS = "waveHeight,waterTemperature,windSpeed,windDirection,windWaveDirection,swellDirection,gust,wavePeriod"
def __init__(self):
self.api_key = os.getenv("STORMGLASS_API_KEY")
def run(self, input_data: WaveDataInput) -> WaveDataOutput:
if not self.api_key:
# No API key? fallback directly
return self._fallback(input_data)
payload = {
"lat": input_data.lat,
"lng": input_data.lon,
"params": self.PARAMS,
}
# If forecast requested: add start/end timestamps
if input_data.target_datetime:
start_time = input_data.target_datetime.isoformat()
end_time = (input_data.target_datetime + timedelta(hours=1)).isoformat()
payload["start"] = start_time
payload["end"] = end_time
try:
response = requests.get(
self.BASE_URL,
params=payload,
headers={
"Authorization": self.api_key,
"Accept": "application/json"
}
)
if response.status_code != 200:
return self._fallback(input_data)
data = response.json()
hours = data.get("hours", [])
if not hours:
return self._fallback(input_data)
# Choose closest hour to target, or first
if input_data.target_datetime and len(hours) > 1:
target_hour = input_data.target_datetime.hour
hour = min(
hours,
key=lambda h: abs(
datetime.fromisoformat(
h["time"].replace("Z", "+00:00")
).hour - target_hour
)
)
else:
hour = hours[0]
return WaveDataOutput(
wave_height=round(hour["waveHeight"]["sg"], 2),
wave_direction=round(hour["windWaveDirection"]["sg"], 2),
wave_period=round(hour.get("wavePeriod", {}).get("sg", 0), 2),
wind_speed=round(hour["windSpeed"]["sg"], 2),
wind_direction=round(hour["windDirection"]["sg"], 2),
swell_direction=round(hour.get("swellDirection", {}).get("sg", 0), 2),
wind_gust=round(hour.get("gust", {}).get("sg", 0), 2),
water_temperature=round(hour["waterTemperature"]["sg"], 2),
timestamp=hour.get("time"),
forecast_target=input_data.target_datetime.isoformat() if input_data.target_datetime else None,
data_source="stormglass"
)
except Exception:
return self._fallback(input_data)
# -------------------------------------------------------
# Fallback (Open-Meteo)
# -------------------------------------------------------
def _fallback(self, input_data: WaveDataInput) -> WaveDataOutput:
"""
Calls the marine_data_tool fallback — converts any lists to single values for Pydantic.
"""
from .marine_data_tool import OpenMeteoMarineAPI
api = OpenMeteoMarineAPI()
if input_data.target_datetime:
forecast_data = api.get_forecast_for_hour(
lat=input_data.lat,
lon=input_data.lon,
target_datetime=input_data.target_datetime
)
else:
forecast_data = api.get_current_conditions(
lat=input_data.lat,
lon=input_data.lon
)
# Helper to pick first element if list, fallback to 0 if None
def to_float(val):
if isinstance(val, list):
return float(val[0]) if val else 0.0
if val is None:
return 0.0
return float(val)
return WaveDataOutput(
wave_height=to_float(forecast_data.get("wave_height")),
wave_direction=to_float(forecast_data.get("wave_direction")),
wave_period=to_float(forecast_data.get("wave_period")),
wind_speed=to_float(forecast_data.get("wind_speed")),
wind_direction=to_float(forecast_data.get("wind_direction")),
swell_direction=to_float(forecast_data.get("swell_direction")),
wind_gust=to_float(forecast_data.get("wind_gust")),
water_temperature=to_float(forecast_data.get("water_temperature")),
timestamp=forecast_data.get("timestamp"),
forecast_target=forecast_data.get("forecast_target"),
data_source="open-meteo"
)
# -------------------------------------------------------
# Registration
# -------------------------------------------------------
def create_stormglass_tool():
tool = StormglassTool()
return {
"name": tool.name,
"description": tool.description,
"input_schema": WaveDataInput.schema(),
"function": tool.run,
}
|