WIP: refactor

main.py

@@ -1,230 +1,535 @@
 import asyncio
 from datetime import datetime, timedelta
+from typing import TypedDict
 import functools
 import logging
 import os
-import pprint
 import zoneinfo
+from urllib.parse import urljoin, urlencode
 
 from fastapi import FastAPI, HTTPException
 from cachetools import TTLCache
 import httpx
 
 
-pp = pprint.PrettyPrinter()
 logger = logging.getLogger("uvicorn.error")
-eastern = zoneinfo.ZoneInfo("America/New_York")
 
-app = FastAPI()
+EASTERN_TZ = zoneinfo.ZoneInfo("America/New_York")
-
+CACHE_TTL_SECONDS = 900  # 15 minutes
-weather_cache = TTLCache(maxsize=100, ttl=900)  # 15 minutes
+CACHE_MAX_SIZE = 100
-pollen_cache = TTLCache(maxsize=100, ttl=900)  # 15 minutes
+POLLEN_MAX_INDEX = 12.0
-
+HTTP_TIMEOUT = 10.0
-CONFIG = {
+DEFAULT_ZIP_CODE = "01970"  # salem, ma
-    # salem, ma
+DEFAULT_LATITUDE = "42.3554334"
-    "zip": "01970",
+DEFAULT_LONGITUDE = "-71.060511"
-    "lat": "42.3554334",
-    "lon": "-71.060511",
-    "weather_api_key": os.environ.get("WEATHER_API_KEY", ""),
-    "auth_token": os.environ["AUTH_TOKEN"],
-}
 
 
-def format_date(dt):
+def get_required_env(key: str) -> str:
-    dt = dt.astimezone(eastern)
+    """Get required environment variable or raise error."""
-    return dt.strftime("%a %d").lower()
+    value = os.environ.get(key)
+    if not value:
+        raise ValueError(f"Required environment variable {key} is not set")
+    return value
 
 
-def format_time(dt):
+class WeatherCondition(TypedDict):
-    dt = dt.astimezone(eastern)
+    description: str
-    return dt.strftime("%I:%M%p").lower()
+    main: str
+    id: int
+    icon: str
 
 
-def format_datetime(dt):
+class CurrentWeather(TypedDict):
-    dt = dt.astimezone(eastern)
+    dt: int
-    return f"{format_date(dt)} {format_time(dt)}"
+    sunrise: int
+    sunset: int
+    temp: float
+    feels_like: float
+    pressure: int
+    humidity: int
+    weather: list[WeatherCondition]
 
 
-def relative_day_to_date(rel_dt):
+class DailyTemperature(TypedDict):
-    dt = datetime.now()
+    min: float
-    day = timedelta(days=1)
+    max: float
-    match rel_dt.lower().strip():
-        case "yesterday":
-            return format_date(dt - day)
-        case "today":
-            return format_date(dt)
-        case "tomorrow":
-            return format_date(dt + day)
-        case passthrough:
-            return passthrough
 
 
-def build_daily_data(date, pollen_periods, weather_periods):
+class DailyWeather(TypedDict):
-    daily_data = {}
+    dt: int
-    if date in pollen_periods:
+    sunrise: int
-        daily_data["pollen"] = pollen_periods[date]["index"]
+    sunset: int
-    if date in weather_periods:
+    temp: DailyTemperature
-        weather_data = weather_periods[date]
+    humidity: int
-        daily_data.update({
+    pressure: int
-            "low": weather_data["low"],
+    weather: list[WeatherCondition]
-            "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
 
 
-def fallback_handler(func):
+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:
-            result = await func(*args, **kwargs)
+            return await func(*args, **kwargs)
         except Exception as e:
-            logger.exception(e)
+            logger.exception(f"Error in {func.__name__}: {e}")
             return []
-        return result
 
     return wrapper
 
 
-@fallback_handler
+class WeatherService:
-async def fetch_pollen(zipcode):
+    """Service for fetching weather data."""
-    if zipcode in pollen_cache:
-        return pollen_cache[zipcode]
 
-    url = f"https://www.pollen.com/api/forecast/current/pollen/{zipcode}"
+    @staticmethod
-    headers = {
+    @error_handler
-        "User-Agent": (
+    async def fetch_weather(latitude: str, longitude: str, api_key: str) -> list[WeatherReport]:
-            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+        """Fetch weather data from OpenWeatherMap API."""
-            "AppleWebKit/537.36 (KHTML, like Gecko) "
+        cache_key = (latitude, longitude)
-            "Chrome/123.0.0.0 Safari/537.36"
+
-        ),
+        if cache_key in weather_cache:
-        "Accept": "application/json, text/plain, */*",
+            return weather_cache[cache_key]
-        "Referer": url,
+
-        "Cookie": f"geo={zipcode}",
+        base_url = "https://api.openweathermap.org/data/3.0/onecall"
-    }
+        params = {
-    async with httpx.AsyncClient(timeout=10.0) as client:
+            "lat": latitude,
-        response = await client.get(url, headers=headers)
+            "lon": longitude,
-        response.raise_for_status()
+            "appid": api_key,
-        data = response.json()
+            "units": "imperial",
-        result = [
+        }
-            {
+        url = f"{base_url}?{urlencode(params)}"
-                "forecast_date": format_datetime(
+
-                    datetime.fromisoformat(data["ForecastDate"])
+        async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
-                ),
+            response = await client.get(url)
-                "periods": [
+            response.raise_for_status()
-                    {
+            raw_data = response.json()
-                        "index": int(d["Index"] / 12. * 100),
+
-                        "period": relative_day_to_date(d["Type"]),
+            weather_data = WeatherData(raw_data)
-                    }
+            result = WeatherService._format_weather_data(weather_data)
-                    for d in data["Location"]["periods"]
+
-                    if d["Type"].lower().strip() in ["today", "tomorrow"]
+            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
+                    ],
+                }
+            )
         ]
-        pollen_cache[zipcode] = result
-        return result
 
+    @staticmethod
+    def _format_daily_period(period: DailyWeather) -> WeatherPeriod:
+        """Format a single daily weather period."""
+        period_dt = datetime.fromtimestamp(period["dt"])
 
-@fallback_handler
+        return WeatherPeriod(
-async def fetch_weather(lat, lon, weather_api_key):
-    cache_key = (lat, lon)
-    if cache_key in weather_cache:
-        return weather_cache[cache_key]
-
-    url = f"https://api.openweathermap.org/data/3.0/onecall?lat={lat}&lon={lon}&appid={weather_api_key}&units=imperial"
-    async with httpx.AsyncClient(timeout=10.0) as client:
-        response = await client.get(url)
-        response.raise_for_status()
-        data = response.json()
-        current, periods = data["current"], data["daily"][:2]
-        result = [
             {
-                "forecast_date": format_datetime(datetime.fromtimestamp(current["dt"])),
+                "low": int(round(period["temp"]["min"])),
-                "current_temp": int(round(current["temp"])),
+                "high": int(round(period["temp"]["max"])),
-                "current_feels_like": int(round(current["feels_like"])),
+                "desc": period["weather"][0]["description"],
-                "current_humidity": current["humidity"],
+                "humidity": period["humidity"],
-                "sunrise": format_time(datetime.fromtimestamp(current["sunrise"])),
+                "sunrise": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunrise"])),
-                "sunset": format_time(datetime.fromtimestamp(current["sunset"])),
+                "sunset": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunset"])),
-                "current_pressure": current["pressure"],
+                "pressure": period["pressure"],
-                "current_desc": current["weather"][0]["description"],
+                "period": DateTimeFormatter.format_date(period_dt),
-                "periods": [
-                    {
-                        "low": int(round(p["temp"]["min"])),
-                        "high": int(round(p["temp"]["max"])),
-                        "desc": p["weather"][0]["description"],
-                        "humidity": p["humidity"],
-                        "sunrise": format_time(datetime.fromtimestamp(p["sunrise"])),
-                        "sunset": format_time(datetime.fromtimestamp(p["sunset"])),
-                        "pressure": p["pressure"],
-                        "period": format_date(datetime.fromtimestamp(p["dt"])),
-                    }
-                    for p in periods
-                ],
             }
-        ]
+        )
-        weather_cache[cache_key] = result
+
-        return result
+
+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 read_root(token: str):
+async def get_weather_pollen_report(token: str) -> FinalReport:
-    if token != CONFIG["auth_token"]:
+    """
-        raise HTTPException(status_code=403, detail="unauthorized")
+    Get weather and pollen report.
 
-    [
+    Args:
-        pollen,
+        token: Authentication token
-        weather,
+
-    ] = await asyncio.gather(
+    Returns:
-        fetch_pollen(CONFIG["zip"]),
+        Dictionary containing current, today, and tomorrow weather/pollen data
-        fetch_weather(CONFIG["lat"], CONFIG["lon"], CONFIG["weather_api_key"]),
+
+    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 = {p["period"]: p for p in pollen[0]["periods"]} if pollen and pollen[0]["periods"] else {}
+    pollen_periods, weather_periods, current_weather_info = DataAggregator.create_periods_lookup(
-    weather_periods = {p["period"]: p for p in weather[0]["periods"]} if weather and weather[0]["periods"] else {}
+        pollen_data, weather_data
+    )
 
-    # Add current weather data as a "current" period
+    now = datetime.now()
-    if weather and weather[0]:
+    today_date = DateTimeFormatter.format_date(now)
-        data = weather[0]
+    tomorrow_date = DateTimeFormatter.format_date(now + timedelta(days=1))
-        weather_periods["current"] = {
+
-            "period": "current",
+    result: FinalReport = FinalReport(
-            "temp": data["current_temp"],
+        {
-            "feels_like": data["current_feels_like"],
+            "fetched_at": DateTimeFormatter.format_datetime(now),
-            "humidity": data["current_humidity"],
+            "current": current_weather_info,
-            "pressure": data["current_pressure"],
+            "today": DailyData({}),
-            "desc": data["current_desc"],
+            "tomorrow": DailyData({}),
-            "sunrise": data["sunrise"],
-            "sunset": data["sunset"]
         }
+    )
 
-    today_date = format_date(datetime.now())
+    today_data = DataAggregator.build_daily_data(today_date, pollen_periods, weather_periods)
-    tomorrow_date = format_date(datetime.now() + timedelta(days=1))
-
-    result = {
-        "fetched_at": format_datetime(datetime.now()),
-        "current": {},
-        "today": {},
-        "tomorrow": {},
-    }
-
-    if "current" in weather_periods:
-        weather_data = weather_periods["current"]
-        result["current"] = {
-            "temp": weather_data["temp"],
-            "feels_like": weather_data["feels_like"],
-            "desc": weather_data["desc"],
-            "humidity": weather_data["humidity"],
-            "pressure": weather_data["pressure"]
-        }
-
-    today_data = build_daily_data(today_date, pollen_periods, weather_periods)
     if today_data:
         result["today"] = today_data
 
-    tomorrow_data = build_daily_data(tomorrow_date, pollen_periods, weather_periods)
+    tomorrow_data = DataAggregator.build_daily_data(tomorrow_date, pollen_periods, weather_periods)
     if tomorrow_data:
         result["tomorrow"] = tomorrow_data