adds ruff linter and formatter, and a makefile

main.py

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