Spaces:
Running
Running
File size: 3,545 Bytes
8aa7d2f ebeb860 8aa7d2f ebeb860 8aa7d2f ebeb860 8aa7d2f ebeb860 8aa7d2f ebeb860 8aa7d2f |
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 |
import json
from datetime import datetime, timedelta
import requests
def _extract_hourly_data(data: dict) -> list[dict]:
hourly_data = data["hourly"]
result = [
{k: v for k, v in zip(hourly_data.keys(), values)}
for values in zip(*hourly_data.values())
]
return result
def _filter_by_date(
date: datetime, hourly_data: list[dict], timedelta: timedelta = timedelta(hours=1)
):
start_date = date - timedelta
end_date = date + timedelta
return [
item
for item in hourly_data
if start_date <= datetime.fromisoformat(item["time"]) <= end_date
]
def get_wave_forecast(lat: float, lon: float, date: str | None = None) -> list[dict]:
"""Get wave forecast for given location.
Forecast will include:
- wave_direction (degrees)
- wave_height (meters)
- wave_period (seconds)
- sea_level_height_msl (meters)
Args:
lat: Latitude of the location.
lon: Longitude of the location.
date: Date to filter by in any valid ISO 8601 format.
If not provided, all data (default to 6 days forecast) will be returned.
Returns:
Hourly data for wave forecast.
Example output:
```json
[
{'time': '2025-03-19T09:00', 'winddirection_10m': 140, 'windspeed_10m': 24.5}, {'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1},
{'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1}, {'time': '2025-03-19T11:00', 'winddirection_10m': 141, 'windspeed_10m': 29.2}
]
```
"""
url = "https://marine-api.open-meteo.com/v1/marine"
params = {
"latitude": lat,
"longitude": lon,
"hourly": [
"wave_direction",
"wave_height",
"wave_period",
"sea_level_height_msl",
],
}
response = requests.get(url, params=params)
response.raise_for_status()
data = json.loads(response.content.decode())
hourly_data = _extract_hourly_data(data)
if date is not None:
date = datetime.fromisoformat(date)
hourly_data = _filter_by_date(date, hourly_data)
return hourly_data
def get_wind_forecast(lat: float, lon: float, date: str | None = None) -> list[dict]:
"""Get wind forecast for given location.
Forecast will include:
- wind_direction (degrees)
- wind_speed (meters per second)
Args:
lat: Latitude of the location.
lon: Longitude of the location.
date: Date to filter by in any valid ISO 8601 format.
If not provided, all data (default to 6 days forecast) will be returned.
Returns:
Hourly data for wind forecast.
Example output:
```json
[
{"time": "2025-03-18T22:00", "wind_direction": 196, "wind_speed": 9.6},
{"time": "2025-03-18T23:00", "wind_direction": 183, "wind_speed": 7.9},
]
```
"""
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": lat,
"longitude": lon,
"hourly": ["winddirection_10m", "windspeed_10m"],
}
response = requests.get(url, params=params)
response.raise_for_status()
data = json.loads(response.content.decode())
hourly_data = _extract_hourly_data(data)
if date is not None:
date = datetime.fromisoformat(date)
hourly_data = _filter_by_date(date, hourly_data)
return hourly_data
|