tools/openmeteo.py

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, strict=False)}
        for values in zip(*hourly_data.values(), strict=False)
    ]
    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) -> 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.

    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)
    if len(hourly_data) == 0:
        raise ValueError("No data found for the given date")
    return hourly_data


def get_wind_forecast(lat: float, lon: float, date: str) -> 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.

    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)
    date = datetime.fromisoformat(date)
    hourly_data = _filter_by_date(date, hourly_data)
    if len(hourly_data) == 0:
        raise ValueError("No data found for the given date")
    return hourly_data