jpmschweitzer/library-desk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

release: v1.6.1 - Weather forecasts, sun times, air quality
Build and Push / build (release) Successful in 52s
Details

Browse Source 
- Weather fetch now returns 7-day forecasts with UV index
- New /volatile/fetch/sun/{city} endpoint for sunrise/sunset
- New /volatile/fetch/air_quality/{city} endpoint for AQI and pollutants
- OpenMeteoProvider now implements AirQualityProvider interface

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-30 10:13:06 +01:00
co-authored by Claude Opus 4.5
parent 6b0530ed79
commit d6b30570a0
9 changed files with 561 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/apis/openmeteo.py
+138 -8
View File
@@ -14,11 +14,14 @@ from typing import Optional
from .base import (
WeatherProvider,
AirQualityProvider,
WeatherCondition,
CurrentWeather,
DayForecast,
WeatherForecast,
GeoLocation,
SunTimes,
AirQuality,
)
logger = logging.getLogger(__name__)
@@ -89,11 +92,12 @@ WMO_DESCRIPTIONS: dict[int, str] = {
}
class OpenMeteoProvider(WeatherProvider):
"""Open-Meteo weather API implementation."""
class OpenMeteoProvider(WeatherProvider, AirQualityProvider):
"""Open-Meteo weather and air quality API implementation."""
GEOCODING_URL = "https://geocoding-api.open-meteo.com/v1/search"
WEATHER_URL = "https://api.open-meteo.com/v1/forecast"
AIR_QUALITY_URL = "https://air-quality-api.open-meteo.com/v1/air-quality"
def __init__(
self,
@@ -194,9 +198,11 @@ class OpenMeteoProvider(WeatherProvider):
"wind_speed_10m",
"wind_direction_10m"
],
"daily": ["uv_index_max"],
"timezone": self.timezone,
"temperature_unit": "celsius",
"wind_speed_unit": "kmh"
"wind_speed_unit": "kmh",
"forecast_days": 1
}
)
response.raise_for_status()
@@ -205,6 +211,12 @@ class OpenMeteoProvider(WeatherProvider):
current = data.get("current", {})
weather_code = current.get("weather_code", 0)
# Get today's UV index from daily data
daily = data.get("daily", {})
uv_index = None
if daily.get("uv_index_max"):
uv_index = daily["uv_index_max"][0]
return CurrentWeather(
temperature=current.get("temperature_2m", 0.0),
feels_like=current.get("apparent_temperature"),
@@ -214,7 +226,8 @@ class OpenMeteoProvider(WeatherProvider):
condition=WMO_CODE_MAP.get(weather_code, WeatherCondition.UNKNOWN),
condition_text=WMO_DESCRIPTIONS.get(weather_code, "Unknown"),
timestamp=datetime.now(),
location=location.name
location=location.name,
uv_index=uv_index
)
except httpx.HTTPError as e:
logger.error(f"Weather request failed for {location.name}: {e}")
@@ -259,7 +272,8 @@ class OpenMeteoProvider(WeatherProvider):
"temperature_2m_max",
"temperature_2m_min",
"precipitation_sum",
"precipitation_probability_max"
"precipitation_probability_max",
"uv_index_max"
],
"timezone": self.timezone,
"temperature_unit": "celsius",
@@ -272,7 +286,14 @@ class OpenMeteoProvider(WeatherProvider):
# Parse current weather
current_data = data.get("current", {})
daily_data = data.get("daily", {})
weather_code = current_data.get("weather_code", 0)
# Get today's UV from daily data
uv_index = None
if daily_data.get("uv_index_max"):
uv_index = daily_data["uv_index_max"][0]
current = CurrentWeather(
temperature=current_data.get("temperature_2m", 0.0),
feels_like=current_data.get("apparent_temperature"),
@@ -282,15 +303,16 @@ class OpenMeteoProvider(WeatherProvider):
condition=WMO_CODE_MAP.get(weather_code, WeatherCondition.UNKNOWN),
condition_text=WMO_DESCRIPTIONS.get(weather_code, "Unknown"),
timestamp=datetime.now(),
location=location.name
location=location.name,
uv_index=uv_index
)
# Parse daily forecast
daily_data = data.get("daily", {})
daily = []
dates = daily_data.get("time", [])
for i, date_str in enumerate(dates):
code = daily_data.get("weather_code", [])[i] if i < len(daily_data.get("weather_code", [])) else 0
uv_max = daily_data.get("uv_index_max", [])[i] if i < len(daily_data.get("uv_index_max", [])) else None
daily.append(DayForecast(
date=datetime.fromisoformat(date_str),
temp_high=daily_data.get("temperature_2m_max", [])[i] if i < len(daily_data.get("temperature_2m_max", [])) else 0.0,
@@ -298,7 +320,8 @@ class OpenMeteoProvider(WeatherProvider):
condition=WMO_CODE_MAP.get(code, WeatherCondition.UNKNOWN),
condition_text=WMO_DESCRIPTIONS.get(code, "Unknown"),
precipitation_chance=daily_data.get("precipitation_probability_max", [])[i] if i < len(daily_data.get("precipitation_probability_max", [])) else None,
precipitation_mm=daily_data.get("precipitation_sum", [])[i] if i < len(daily_data.get("precipitation_sum", [])) else None
precipitation_mm=daily_data.get("precipitation_sum", [])[i] if i < len(daily_data.get("precipitation_sum", [])) else None,
uv_index_max=uv_max
))
return WeatherForecast(
@@ -309,3 +332,110 @@ class OpenMeteoProvider(WeatherProvider):
except httpx.HTTPError as e:
logger.error(f"Forecast request failed for {location.name}: {e}")
raise ValueError(f"Failed to get forecast: {e}")
async def get_sun_times(self, location: GeoLocation) -> SunTimes:
"""
Get sunrise/sunset times for today.
Args:
location: GeoLocation with lat/long
Returns:
SunTimes with sunrise, sunset, and daylight duration
Raises:
ValueError: If API request fails
"""
try:
response = await self.client.get(
self.WEATHER_URL,
params={
"latitude": location.latitude,
"longitude": location.longitude,
"daily": [
"sunrise",
"sunset",
"daylight_duration"
],
"timezone": self.timezone,
"forecast_days": 1
}
)
response.raise_for_status()
data = response.json()
daily = data.get("daily", {})
date_str = daily.get("time", [""])[0]
sunrise_str = daily.get("sunrise", [""])[0]
sunset_str = daily.get("sunset", [""])[0]
daylight = daily.get("daylight_duration", [0])[0]
return SunTimes(
location=location.name,
date=datetime.fromisoformat(date_str) if date_str else datetime.now(),
sunrise=datetime.fromisoformat(sunrise_str) if sunrise_str else datetime.now(),
sunset=datetime.fromisoformat(sunset_str) if sunset_str else datetime.now(),
daylight_duration=int(daylight) if daylight else 0
)
except httpx.HTTPError as e:
logger.error(f"Sun times request failed for {location.name}: {e}")
raise ValueError(f"Failed to get sun times: {e}")
async def get_air_quality(self, location: GeoLocation) -> AirQuality:
"""
Get current air quality for a location.
Args:
location: GeoLocation with lat/long
Returns:
AirQuality with pollutant measurements and AQI
Raises:
ValueError: If API request fails
"""
try:
response = await self.client.get(
self.AIR_QUALITY_URL,
params={
"latitude": location.latitude,
"longitude": location.longitude,
"current": [
"european_aqi",
"us_aqi",
"pm2_5",
"pm10",
"ozone",
"nitrogen_dioxide",
"sulphur_dioxide",
"carbon_monoxide",
"grass_pollen",
"birch_pollen",
"alder_pollen"
],
"timezone": self.timezone
}
)
response.raise_for_status()
data = response.json()
current = data.get("current", {})
return AirQuality(
location=location.name,
timestamp=datetime.now(),
aqi_european=current.get("european_aqi"),
aqi_us=current.get("us_aqi"),
pm2_5=current.get("pm2_5"),
pm10=current.get("pm10"),
ozone=current.get("ozone"),
nitrogen_dioxide=current.get("nitrogen_dioxide"),
sulphur_dioxide=current.get("sulphur_dioxide"),
carbon_monoxide=current.get("carbon_monoxide"),
pollen_grass=current.get("grass_pollen"),
pollen_birch=current.get("birch_pollen"),
pollen_alder=current.get("alder_pollen")
)
except httpx.HTTPError as e:
logger.error(f"Air quality request failed for {location.name}: {e}")
raise ValueError(f"Failed to get air quality: {e}")