Spaces:
Sleeping
Sleeping
File size: 6,547 Bytes
7b6b271 23a9367 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 |
"""
Open-Meteo Marine Data Tool - Free Reliable Fallback.
This module provides marine and weather data from Open-Meteo APIs
as a free, reliable fallback when premium services are unavailable.
It's designed to be compatible with the WaveDataOutput schema.
Data Sources:
- Marine API: Wave height, direction, period, swell data
- Weather API: Wind speed, direction, gusts
Features:
- No API key required
- High availability and reliability
- Compatible output format with premium services
- Supports both current conditions and forecasts
Limitations:
- Lower resolution than premium APIs
- No water temperature data
- Reduced forecast accuracy at longer ranges
Example:
>>> api = OpenMeteoMarineAPI()
>>> conditions = api.get_current_conditions(36.72, -4.42)
>>> print(f"Wave height: {conditions['wave_height']}m")
Author: Surf Spot Finder Team
License: MIT
"""
import requests
from datetime import datetime
from typing import Dict, Optional
class OpenMeteoMarineAPI:
"""Open-Meteo marine data client with fallback capabilities.
Provides marine and weather data from Open-Meteo's free APIs.
Designed as a reliable fallback for premium marine data services.
Attributes:
BASE_MARINE: Open-Meteo marine API endpoint.
BASE_WEATHER: Open-Meteo weather API endpoint.
Example:
>>> api = OpenMeteoMarineAPI()
>>> data = api.get_current_conditions(lat=36.72, lon=-4.42)
>>> forecast = api.get_forecast_for_hour(lat, lon, datetime_obj)
"""
BASE_MARINE = "https://marine-api.open-meteo.com/v1/marine"
BASE_WEATHER = "https://api.open-meteo.com/v1/forecast"
def get_current_conditions(self, lat: float, lon: float) -> Dict:
"""Get current marine and wind conditions.
Fetches current wave, wind, and swell data from Open-Meteo APIs.
Combines marine and weather data into unified response.
Args:
lat: Latitude in decimal degrees.
lon: Longitude in decimal degrees.
Returns:
Dict containing wave_height, wind_speed, directions, etc.
Compatible with WaveDataOutput schema.
"""
marine = self._fetch_marine(lat, lon)
wind = self._fetch_wind(lat, lon)
return {
"wave_height": marine.get("wave_height", 0),
"wave_direction": marine.get("wave_direction", 0),
"wave_period": marine.get("wave_period", 0),
"swell_direction": marine.get("swell_wave_direction", 0),
"wind_speed": wind.get("wind_speed", 0),
"wind_direction": wind.get("wind_direction", 0),
"wind_gust": wind.get("wind_gusts", 0),
"water_temperature": None,
"timestamp": marine.get("timestamp"),
"forecast_target": None,
"data_source": "open-meteo",
}
def get_forecast_for_hour(self, lat: float, lon: float, target_datetime: datetime) -> Dict:
"""Get marine forecast for specific datetime.
Retrieves forecast data for the specified hour from Open-Meteo.
Automatically finds the closest available forecast time.
Args:
lat: Latitude in decimal degrees.
lon: Longitude in decimal degrees.
target_datetime: Target datetime for forecast data.
Returns:
Dict with forecast conditions for the specified time.
Returns current conditions if forecast unavailable.
"""
marine = self._fetch_marine(lat, lon, forecast_days=3)
wind = self._fetch_wind(lat, lon)
# find the closest hour
target_hour_str = target_datetime.replace(minute=0, second=0, microsecond=0).isoformat()
timestamps = marine["timestamps"]
if target_hour_str in timestamps:
idx = timestamps.index(target_hour_str)
else:
idx = 0 # fallback
return {
"wave_height": marine["wave_height"][idx],
"wave_direction": marine["wave_direction"][idx],
"wave_period": marine["wave_period"][idx],
"swell_direction": marine["swell_wave_direction"][idx],
"wind_speed": wind.get("wind_speed", 0),
"wind_direction": wind.get("wind_direction", 0),
"wind_gust": wind.get("wind_gusts", 0),
"water_temperature": None,
"timestamp": timestamps[idx],
"forecast_target": target_hour_str,
"data_source": "open-meteo",
}
# --- Internal helpers ---------------------------------------------------
def _fetch_marine(self, lat: float, lon: float, forecast_days: int = 1) -> Dict:
params = {
"latitude": lat,
"longitude": lon,
"hourly": [
"wave_height",
"wave_direction",
"wave_period",
"swell_wave_height",
"swell_wave_direction",
"swell_wave_period",
],
"forecast_days": forecast_days,
"timezone": "auto",
}
resp = requests.get(self.BASE_MARINE, params=params)
resp.raise_for_status()
data = resp.json()
hourly = data.get("hourly", {})
return {
"timestamps": hourly.get("time", []),
"wave_height": hourly.get("wave_height", []),
"wave_direction": hourly.get("wave_direction", []),
"wave_period": hourly.get("wave_period", []),
"swell_wave_height": hourly.get("swell_wave_height", []),
"swell_wave_direction": hourly.get("swell_wave_direction", []),
"swell_wave_period": hourly.get("swell_wave_period", []),
"timestamp": hourly.get("time", [None])[0],
}
def _fetch_wind(self, lat: float, lon: float) -> Dict:
params = {
"latitude": lat,
"longitude": lon,
"hourly": ["wind_speed_10m", "wind_direction_10m", "wind_gusts_10m"],
"forecast_days": 1,
"timezone": "auto",
}
resp = requests.get(self.BASE_WEATHER, params=params)
resp.raise_for_status()
hourly = resp.json().get("hourly", {})
return {
"wind_speed": hourly.get("wind_speed_10m", [0])[0],
"wind_direction": hourly.get("wind_direction_10m", [0])[0],
"wind_gusts": hourly.get("wind_gusts_10m", [0])[0],
}
|