536 lines
16 KiB
Python
536 lines
16 KiB
Python
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from typing import TypedDict
|
|
import functools
|
|
import logging
|
|
import os
|
|
import zoneinfo
|
|
from urllib.parse import urljoin, urlencode
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from cachetools import TTLCache
|
|
import httpx
|
|
|
|
|
|
logger = logging.getLogger("uvicorn.error")
|
|
|
|
EASTERN_TZ = zoneinfo.ZoneInfo("America/New_York")
|
|
CACHE_TTL_SECONDS = 900 # 15 minutes
|
|
CACHE_MAX_SIZE = 100
|
|
POLLEN_MAX_INDEX = 12.0
|
|
HTTP_TIMEOUT = 10.0
|
|
DEFAULT_ZIP_CODE = "01970" # salem, ma
|
|
DEFAULT_LATITUDE = "42.3554334"
|
|
DEFAULT_LONGITUDE = "-71.060511"
|
|
|
|
|
|
def get_required_env(key: str) -> str:
|
|
"""Get required environment variable or raise error."""
|
|
value = os.environ.get(key)
|
|
if not value:
|
|
raise ValueError(f"Required environment variable {key} is not set")
|
|
return value
|
|
|
|
|
|
class WeatherCondition(TypedDict):
|
|
description: str
|
|
main: str
|
|
id: int
|
|
icon: str
|
|
|
|
|
|
class CurrentWeather(TypedDict):
|
|
dt: int
|
|
sunrise: int
|
|
sunset: int
|
|
temp: float
|
|
feels_like: float
|
|
pressure: int
|
|
humidity: int
|
|
weather: list[WeatherCondition]
|
|
|
|
|
|
class DailyTemperature(TypedDict):
|
|
min: float
|
|
max: float
|
|
|
|
|
|
class DailyWeather(TypedDict):
|
|
dt: int
|
|
sunrise: int
|
|
sunset: int
|
|
temp: DailyTemperature
|
|
humidity: int
|
|
pressure: int
|
|
weather: list[WeatherCondition]
|
|
|
|
|
|
class WeatherApiResponse(TypedDict):
|
|
current: CurrentWeather
|
|
daily: list[DailyWeather]
|
|
|
|
|
|
class PollenLocation(TypedDict):
|
|
periods: list[dict[str, str | int | float]]
|
|
|
|
|
|
class PollenApiResponse(TypedDict):
|
|
ForecastDate: str
|
|
Location: PollenLocation
|
|
|
|
|
|
class WeatherPeriod(TypedDict):
|
|
low: int
|
|
high: int
|
|
desc: str
|
|
humidity: int
|
|
sunrise: str
|
|
sunset: str
|
|
pressure: int
|
|
period: str
|
|
|
|
|
|
class PollenPeriod(TypedDict):
|
|
index: int
|
|
period: str
|
|
|
|
|
|
class WeatherReport(TypedDict):
|
|
forecast_date: str
|
|
current_temp: int
|
|
current_feels_like: int
|
|
current_humidity: int
|
|
sunrise: str
|
|
sunset: str
|
|
current_pressure: int
|
|
current_desc: str
|
|
periods: list[WeatherPeriod]
|
|
|
|
|
|
class PollenReport(TypedDict):
|
|
forecast_date: str
|
|
periods: list[PollenPeriod]
|
|
|
|
|
|
class CurrentWeatherData(TypedDict):
|
|
temp: int
|
|
feels_like: int
|
|
desc: str
|
|
humidity: int
|
|
pressure: int
|
|
|
|
|
|
class DailyData(TypedDict, total=False):
|
|
pollen: int
|
|
low: int
|
|
high: int
|
|
desc: str
|
|
humidity: int
|
|
sunrise: str
|
|
sunset: str
|
|
pressure: int
|
|
|
|
|
|
class FinalReport(TypedDict):
|
|
fetched_at: str
|
|
current: CurrentWeatherData
|
|
today: DailyData
|
|
tomorrow: DailyData
|
|
|
|
|
|
app = FastAPI(title="trmnl weather & pollen report")
|
|
|
|
weather_cache: TTLCache = TTLCache(maxsize=CACHE_MAX_SIZE, ttl=CACHE_TTL_SECONDS)
|
|
pollen_cache: TTLCache = TTLCache(maxsize=CACHE_MAX_SIZE, ttl=CACHE_TTL_SECONDS)
|
|
|
|
|
|
class Config:
|
|
"""Application configuration."""
|
|
|
|
def __init__(
|
|
self,
|
|
zip_code: str,
|
|
latitude: str,
|
|
longitude: str,
|
|
weather_api_key: str,
|
|
auth_token: str,
|
|
):
|
|
self.zip_code = zip_code
|
|
self.latitude = latitude
|
|
self.longitude = longitude
|
|
self.weather_api_key = weather_api_key
|
|
self.auth_token = auth_token
|
|
|
|
|
|
config = Config(
|
|
zip_code=DEFAULT_ZIP_CODE,
|
|
latitude=DEFAULT_LATITUDE,
|
|
longitude=DEFAULT_LONGITUDE,
|
|
weather_api_key=get_required_env("WEATHER_API_KEY"),
|
|
auth_token=get_required_env("AUTH_TOKEN"),
|
|
)
|
|
|
|
|
|
class DateTimeFormatter:
|
|
"""Utility class for datetime formatting."""
|
|
|
|
@staticmethod
|
|
def format_date(dt: datetime) -> str:
|
|
"""Format datetime as abbreviated date string (e.g., 'mon 15')."""
|
|
dt = dt.astimezone(EASTERN_TZ)
|
|
return dt.strftime("%a %d").lower()
|
|
|
|
@staticmethod
|
|
def format_time(dt: datetime) -> str:
|
|
"""Format datetime as time string (e.g., '02:30pm')."""
|
|
dt = dt.astimezone(EASTERN_TZ)
|
|
return dt.strftime("%I:%M%p").lower()
|
|
|
|
@staticmethod
|
|
def format_datetime(dt: datetime) -> str:
|
|
"""Format datetime as date and time string."""
|
|
return f"{DateTimeFormatter.format_date(dt)} {DateTimeFormatter.format_time(dt)}"
|
|
|
|
@staticmethod
|
|
def relative_day_to_date(relative_day: str) -> str:
|
|
"""Convert relative day string to formatted date string."""
|
|
now = datetime.now()
|
|
day_delta = timedelta(days=1)
|
|
|
|
relative_day = relative_day.lower().strip()
|
|
|
|
if relative_day == "yesterday":
|
|
return DateTimeFormatter.format_date(now - day_delta)
|
|
elif relative_day == "today":
|
|
return DateTimeFormatter.format_date(now)
|
|
elif relative_day == "tomorrow":
|
|
return DateTimeFormatter.format_date(now + day_delta)
|
|
else:
|
|
return relative_day
|
|
|
|
|
|
class WeatherData:
|
|
"""Data model for weather information."""
|
|
|
|
def __init__(self, raw_data: WeatherApiResponse):
|
|
self.raw_data = raw_data
|
|
self._validate_data()
|
|
|
|
def _validate_data(self) -> None:
|
|
"""Validate required fields in weather data."""
|
|
required_fields = ["current", "daily"]
|
|
for field in required_fields:
|
|
if field not in self.raw_data:
|
|
raise ValueError(f"Missing required field in weather data: {field}")
|
|
|
|
@property
|
|
def current(self) -> CurrentWeather:
|
|
"""Get current weather data."""
|
|
return self.raw_data["current"]
|
|
|
|
@property
|
|
def daily_periods(self) -> list[DailyWeather]:
|
|
"""Get daily forecast periods (limited to next 2 days)."""
|
|
return self.raw_data["daily"][:2]
|
|
|
|
|
|
class PollenData:
|
|
"""Data model for pollen information."""
|
|
|
|
def __init__(self, raw_data: PollenApiResponse):
|
|
self.raw_data = raw_data
|
|
self._validate_data()
|
|
|
|
def _validate_data(self) -> None:
|
|
"""Validate required fields in pollen data."""
|
|
required_fields = ["ForecastDate", "Location"]
|
|
for field in required_fields:
|
|
if field not in self.raw_data:
|
|
raise ValueError(f"Missing required field in pollen data: {field}")
|
|
|
|
@property
|
|
def forecast_date(self) -> str:
|
|
"""Get formatted forecast date."""
|
|
dt = datetime.fromisoformat(self.raw_data["ForecastDate"])
|
|
return DateTimeFormatter.format_datetime(dt)
|
|
|
|
@property
|
|
def periods(self) -> list[PollenPeriod]:
|
|
"""Get pollen periods for today and tomorrow."""
|
|
periods = self.raw_data["Location"].get("periods", [])
|
|
valid_periods: list[PollenPeriod] = []
|
|
|
|
for period in periods:
|
|
period_type = str(period.get("Type", "")).lower().strip()
|
|
if period_type in ["today", "tomorrow"]:
|
|
index_value = float(period.get("Index", 0))
|
|
# convert pollen index to percentage scale
|
|
pollen_percentage = int(index_value / POLLEN_MAX_INDEX * 100)
|
|
|
|
valid_periods.append(
|
|
PollenPeriod(
|
|
{
|
|
"index": pollen_percentage,
|
|
"period": DateTimeFormatter.relative_day_to_date(period_type),
|
|
}
|
|
)
|
|
)
|
|
|
|
return valid_periods
|
|
|
|
|
|
def error_handler(func):
|
|
"""Decorator to handle exceptions in async functions."""
|
|
|
|
@functools.wraps(func)
|
|
async def wrapper(*args, **kwargs):
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except Exception as e:
|
|
logger.exception(f"Error in {func.__name__}: {e}")
|
|
return []
|
|
|
|
return wrapper
|
|
|
|
|
|
class WeatherService:
|
|
"""Service for fetching weather data."""
|
|
|
|
@staticmethod
|
|
@error_handler
|
|
async def fetch_weather(latitude: str, longitude: str, api_key: str) -> list[WeatherReport]:
|
|
"""Fetch weather data from OpenWeatherMap API."""
|
|
cache_key = (latitude, longitude)
|
|
|
|
if cache_key in weather_cache:
|
|
return weather_cache[cache_key]
|
|
|
|
base_url = "https://api.openweathermap.org/data/3.0/onecall"
|
|
params = {
|
|
"lat": latitude,
|
|
"lon": longitude,
|
|
"appid": api_key,
|
|
"units": "imperial",
|
|
}
|
|
url = f"{base_url}?{urlencode(params)}"
|
|
|
|
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
raw_data = response.json()
|
|
|
|
weather_data = WeatherData(raw_data)
|
|
result = WeatherService._format_weather_data(weather_data)
|
|
|
|
weather_cache[cache_key] = result
|
|
return result
|
|
|
|
@staticmethod
|
|
def _format_weather_data(weather_data: WeatherData) -> list[WeatherReport]:
|
|
"""Format weather data for API response."""
|
|
current = weather_data.current
|
|
daily_periods = weather_data.daily_periods
|
|
|
|
current_dt = datetime.fromtimestamp(current["dt"])
|
|
|
|
return [
|
|
WeatherReport(
|
|
{
|
|
"forecast_date": DateTimeFormatter.format_datetime(current_dt),
|
|
"current_temp": int(round(current["temp"])),
|
|
"current_feels_like": int(round(current["feels_like"])),
|
|
"current_humidity": current["humidity"],
|
|
"sunrise": DateTimeFormatter.format_time(
|
|
datetime.fromtimestamp(current["sunrise"])
|
|
),
|
|
"sunset": DateTimeFormatter.format_time(
|
|
datetime.fromtimestamp(current["sunset"])
|
|
),
|
|
"current_pressure": current["pressure"],
|
|
"current_desc": current["weather"][0]["description"],
|
|
"periods": [
|
|
WeatherService._format_daily_period(period) for period in daily_periods
|
|
],
|
|
}
|
|
)
|
|
]
|
|
|
|
@staticmethod
|
|
def _format_daily_period(period: DailyWeather) -> WeatherPeriod:
|
|
"""Format a single daily weather period."""
|
|
period_dt = datetime.fromtimestamp(period["dt"])
|
|
|
|
return WeatherPeriod(
|
|
{
|
|
"low": int(round(period["temp"]["min"])),
|
|
"high": int(round(period["temp"]["max"])),
|
|
"desc": period["weather"][0]["description"],
|
|
"humidity": period["humidity"],
|
|
"sunrise": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunrise"])),
|
|
"sunset": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunset"])),
|
|
"pressure": period["pressure"],
|
|
"period": DateTimeFormatter.format_date(period_dt),
|
|
}
|
|
)
|
|
|
|
|
|
class PollenService:
|
|
"""Service for fetching pollen data."""
|
|
|
|
@staticmethod
|
|
@error_handler
|
|
async def fetch_pollen(zip_code: str) -> list[PollenReport]:
|
|
"""Fetch pollen data from pollen.com API."""
|
|
if zip_code in pollen_cache:
|
|
return pollen_cache[zip_code]
|
|
|
|
base_url = "https://www.pollen.com/api/forecast/current/pollen/"
|
|
url = urljoin(base_url, zip_code)
|
|
headers = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/123.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Referer": url,
|
|
"Cookie": f"geo={zip_code}",
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
|
|
response = await client.get(url, headers=headers)
|
|
response.raise_for_status()
|
|
raw_data = response.json()
|
|
|
|
pollen_data = PollenData(raw_data)
|
|
result = [
|
|
PollenReport(
|
|
{
|
|
"forecast_date": pollen_data.forecast_date,
|
|
"periods": pollen_data.periods,
|
|
}
|
|
)
|
|
]
|
|
|
|
pollen_cache[zip_code] = result
|
|
return result
|
|
|
|
|
|
class DataAggregator:
|
|
"""Service for aggregating weather and pollen data."""
|
|
|
|
@staticmethod
|
|
def build_daily_data(
|
|
date: str,
|
|
pollen_periods: dict[str, PollenPeriod],
|
|
weather_periods: dict[str, WeatherPeriod],
|
|
) -> DailyData:
|
|
"""Build daily data combining weather and pollen information."""
|
|
daily_data: DailyData = {}
|
|
|
|
if date in pollen_periods:
|
|
daily_data["pollen"] = pollen_periods[date]["index"]
|
|
|
|
if date in weather_periods:
|
|
weather_data = weather_periods[date]
|
|
daily_data.update(
|
|
{
|
|
"low": weather_data["low"],
|
|
"high": weather_data["high"],
|
|
"desc": weather_data["desc"],
|
|
"humidity": weather_data["humidity"],
|
|
"sunrise": weather_data["sunrise"],
|
|
"sunset": weather_data["sunset"],
|
|
"pressure": weather_data["pressure"],
|
|
}
|
|
)
|
|
|
|
return daily_data
|
|
|
|
@staticmethod
|
|
def create_periods_lookup(
|
|
pollen_data: list[PollenReport], weather_data: list[WeatherReport]
|
|
) -> tuple[dict[str, PollenPeriod], dict[str, WeatherPeriod], CurrentWeatherData]:
|
|
"""Create lookup dictionaries for pollen and weather periods."""
|
|
pollen_periods: dict[str, PollenPeriod] = {}
|
|
weather_periods: dict[str, WeatherPeriod] = {}
|
|
|
|
if pollen_data and pollen_data[0].get("periods"):
|
|
pollen_periods = {p["period"]: p for p in pollen_data[0]["periods"]}
|
|
|
|
if weather_data and weather_data[0].get("periods"):
|
|
weather_periods = {p["period"]: p for p in weather_data[0]["periods"]}
|
|
|
|
current_weather_info = CurrentWeatherData(
|
|
{
|
|
"temp": 0,
|
|
"feels_like": 0,
|
|
"desc": "",
|
|
"humidity": 0,
|
|
"pressure": 0,
|
|
}
|
|
)
|
|
|
|
if weather_data and weather_data[0]:
|
|
data = weather_data[0]
|
|
current_weather_info = CurrentWeatherData(
|
|
{
|
|
"temp": data["current_temp"],
|
|
"feels_like": data["current_feels_like"],
|
|
"desc": data["current_desc"],
|
|
"humidity": data["current_humidity"],
|
|
"pressure": data["current_pressure"],
|
|
}
|
|
)
|
|
|
|
return pollen_periods, weather_periods, current_weather_info
|
|
|
|
|
|
@app.get("/")
|
|
async def get_weather_pollen_report(token: str) -> FinalReport:
|
|
"""
|
|
Get weather and pollen report.
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
Dictionary containing current, today, and tomorrow weather/pollen data
|
|
|
|
Raises:
|
|
HTTPException: If authentication fails
|
|
"""
|
|
if token != config.auth_token:
|
|
raise HTTPException(status_code=403, detail="Unauthorized")
|
|
|
|
pollen_data, weather_data = await asyncio.gather(
|
|
PollenService.fetch_pollen(config.zip_code),
|
|
WeatherService.fetch_weather(config.latitude, config.longitude, config.weather_api_key),
|
|
)
|
|
|
|
pollen_periods, weather_periods, current_weather_info = DataAggregator.create_periods_lookup(
|
|
pollen_data, weather_data
|
|
)
|
|
|
|
now = datetime.now()
|
|
today_date = DateTimeFormatter.format_date(now)
|
|
tomorrow_date = DateTimeFormatter.format_date(now + timedelta(days=1))
|
|
|
|
result: FinalReport = FinalReport(
|
|
{
|
|
"fetched_at": DateTimeFormatter.format_datetime(now),
|
|
"current": current_weather_info,
|
|
"today": DailyData({}),
|
|
"tomorrow": DailyData({}),
|
|
}
|
|
)
|
|
|
|
today_data = DataAggregator.build_daily_data(today_date, pollen_periods, weather_periods)
|
|
if today_data:
|
|
result["today"] = today_data
|
|
|
|
tomorrow_data = DataAggregator.build_daily_data(tomorrow_date, pollen_periods, weather_periods)
|
|
if tomorrow_data:
|
|
result["tomorrow"] = tomorrow_data
|
|
|
|
return result
|