Spaces:
Sleeping
Sleeping
| """ | |
| 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], | |
| } | |