capturegraph.scheduling.forecast.weather

Weather Forecasting - Open-Meteo Integration

Fetches hourly weather forecasts from the Open-Meteo API (free, no API key required).

hourly_weather(location, days=3)

Fetch hourly weather forecast from Open-Meteo.

Parameters:

Name	Type	Description	Default
location	Location	Location with latitude and longitude attributes.	required
days	int	Number of days to forecast (1-16). Defaults to 3.	3

Returns:

Type	Description
List[Weather]	List of Weather objects for each hour.

Source code in capturegraph-lib/capturegraph/scheduling/forecast/weather.py

def hourly_weather(
    location: cg.Location,
    days: int = 3,
) -> cg.List[cg.Weather]:
    """Fetch hourly weather forecast from Open-Meteo.

    Args:
        location: Location with latitude and longitude attributes.
        days: Number of days to forecast (1-16). Defaults to 3.

    Returns:
        List of Weather objects for each hour.
    """
    import openmeteo_requests
    import requests_cache
    from retry_requests import retry

    # Setup cached session with retry
    cache_session = requests_cache.CachedSession(".cache", expire_after=3600)
    retry_session = retry(cache_session, retries=5, backoff_factor=0.2)
    openmeteo = openmeteo_requests.Client(session=retry_session)

    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": location.latitude,
        "longitude": location.longitude,
        "forecast_days": min(days, 16),
        "hourly": [
            "temperature_2m",
            "apparent_temperature",
            "dew_point_2m",
            "relative_humidity_2m",
            "pressure_msl",
            "wind_speed_10m",
            "wind_gusts_10m",
            "wind_direction_10m",
            "cloud_cover",
            "precipitation",
            "visibility",
            "uv_index",
            "is_day",
            "weather_code",
        ],
    }

    responses = openmeteo.weather_api(url, params=params)
    response = responses[0]

    hourly = response.Hourly()

    # Extract variables in the same order as requested
    temp = hourly.Variables(0).ValuesAsNumpy()
    apparent_temp = hourly.Variables(1).ValuesAsNumpy()
    dew_point = hourly.Variables(2).ValuesAsNumpy()
    humidity = hourly.Variables(3).ValuesAsNumpy()
    pressure = hourly.Variables(4).ValuesAsNumpy()
    wind_speed = hourly.Variables(5).ValuesAsNumpy()
    wind_gust = hourly.Variables(6).ValuesAsNumpy()
    wind_dir = hourly.Variables(7).ValuesAsNumpy()
    cloud_cover = hourly.Variables(8).ValuesAsNumpy()
    precip = hourly.Variables(9).ValuesAsNumpy()
    visibility = hourly.Variables(10).ValuesAsNumpy()
    uv_index = hourly.Variables(11).ValuesAsNumpy()
    is_day = hourly.Variables(12).ValuesAsNumpy()
    weather_code = hourly.Variables(13).ValuesAsNumpy()

    # Build datetime index
    dates = pd.date_range(
        start=pd.to_datetime(hourly.Time(), unit="s", utc=True),
        end=pd.to_datetime(hourly.TimeEnd(), unit="s", utc=True),
        freq=pd.Timedelta(seconds=hourly.Interval()),
        inclusive="left",
    )

    forecasts = []
    for i, dt in enumerate(dates):
        weather = cg.Weather(
            temperature_celsius=float(temp[i]),
            apparent_temperature_celsius=float(apparent_temp[i]),
            dew_point_celsius=float(dew_point[i]),
            humidity_ratio=float(humidity[i]) / 100.0,
            pressure_hpa=float(pressure[i]),
            pressure_trend="",
            wind_speed_mps=float(wind_speed[i]) / 3.6,  # km/h to m/s
            wind_gust_mps=float(wind_gust[i]) / 3.6,
            wind_direction_degrees=float(wind_dir[i]),
            condition=_weather_code_to_condition(int(weather_code[i])),
            symbol_name=_weather_code_to_symbol(int(weather_code[i])),
            cloud_cover_ratio=float(cloud_cover[i]) / 100.0,
            precipitation_intensity_mmph=float(precip[i]),
            visibility_meters=float(visibility[i]),
            uv_index=float(uv_index[i]),
            is_daylight=bool(is_day[i]),
            date=dt.to_pydatetime().replace(tzinfo=None),
        )
        forecasts.append(weather)

    return cg.List(forecasts)

nearest_weather(weather, times)

Find the nearest weather entry for each timestamp.

Sorts the weather list by date and uses binary search to find the closest weather entry for each query time.

Parameters:

Name	Type	Description	Default
weather	List[Weather]	List of Weather objects with date attributes.	required
times	List[datetime]	List of datetimes to find weather for.	required

Returns:

Type	Description
List[Weather]	List of Weather objects, one for each query time.

Source code in capturegraph-lib/capturegraph/scheduling/forecast/weather.py

def nearest_weather(
    weather: cg.List[cg.Weather],
    times: cg.List[datetime],
) -> cg.List[cg.Weather]:
    """Find the nearest weather entry for each timestamp.

    Sorts the weather list by date and uses binary search to find the
    closest weather entry for each query time.

    Args:
        weather: List of Weather objects with date attributes.
        times: List of datetimes to find weather for.

    Returns:
        List of Weather objects, one for each query time.
    """
    import bisect

    # Sort weather by timestamp
    sorted_weather = sorted(weather, key=lambda w: w.date)
    timestamps = [w.date for w in sorted_weather]

    results = []
    for t in times:
        # Find insertion point
        idx = bisect.bisect_left(timestamps, t)

        # Handle edge cases
        if idx == 0:
            nearest = sorted_weather[0]
        elif idx == len(sorted_weather):
            nearest = sorted_weather[-1]
        else:
            # Compare distances to left and right neighbors
            left = sorted_weather[idx - 1]
            right = sorted_weather[idx]
            if abs(t - left.date) <= abs(right.date - t):
                nearest = left
            else:
                nearest = right

        results.append(nearest)

    return cg.List(results)