wip refactor

main.py

@@ -1,6 +1,6 @@
 import asyncio
 from datetime import datetime, timedelta
-from typing import Dict, List, Optional, Any, Tuple
+from typing import TypedDict, Union
 import functools
 import logging
 import os
@@ -11,7 +11,6 @@ from cachetools import TTLCache
 import httpx
 
 
-# Configure logging
 logger = logging.getLogger("uvicorn.error")
 
 # Constants
@@ -22,6 +21,98 @@ POLLEN_MAX_INDEX = 12.0
 POLLEN_PERCENTAGE_SCALE = 100
 HTTP_TIMEOUT = 10.0
 
+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, Union[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
+
 # Initialize FastAPI app
 app = FastAPI(title="TRMNL Weather & Pollen Report")
 
@@ -49,7 +140,6 @@ class Config:
         return value
 
 
-# Global config instance
 config = Config()
 
 
@@ -94,7 +184,7 @@ class DateTimeFormatter:
 class WeatherData:
     """Data model for weather information."""
 
-    def __init__(self, raw_data: Dict[str, Any]):
+    def __init__(self, raw_data: WeatherApiResponse):
         self.raw_data = raw_data
         self._validate_data()
 
@@ -106,12 +196,12 @@ class WeatherData:
                 raise ValueError(f"Missing required field in weather data: {field}")
 
     @property
-    def current(self) -> Dict[str, Any]:
+    def current(self) -> CurrentWeather:
         """Get current weather data."""
         return self.raw_data["current"]
 
     @property
-    def daily_periods(self) -> List[Dict[str, Any]]:
+    def daily_periods(self) -> list[DailyWeather]:
         """Get daily forecast periods (limited to next 2 days)."""
         return self.raw_data["daily"][:2]
 
@@ -119,7 +209,7 @@ class WeatherData:
 class PollenData:
     """Data model for pollen information."""
 
-    def __init__(self, raw_data: Dict[str, Any]):
+    def __init__(self, raw_data: PollenApiResponse):
         self.raw_data = raw_data
         self._validate_data()
 
@@ -137,22 +227,22 @@ class PollenData:
         return DateTimeFormatter.format_datetime(dt)
 
     @property
-    def periods(self) -> List[Dict[str, Any]]:
+    def periods(self) -> list[PollenPeriod]:
         """Get pollen periods for today and tomorrow."""
         periods = self.raw_data["Location"].get("periods", [])
-        valid_periods = []
+        valid_periods: list[PollenPeriod] = []
 
         for period in periods:
-            period_type = period.get("Type", "").lower().strip()
+            period_type = str(period.get("Type", "")).lower().strip()
             if period_type in ["today", "tomorrow"]:
-                index_value = period.get("Index", 0)
+                index_value = float(period.get("Index", 0))
                 # Convert pollen index to percentage scale (0-100)
                 pollen_percentage = int(index_value / POLLEN_MAX_INDEX * POLLEN_PERCENTAGE_SCALE)
 
-                valid_periods.append({
+                valid_periods.append(PollenPeriod({
                     "index": pollen_percentage,
                     "period": DateTimeFormatter.relative_day_to_date(period_type),
-                })
+                }))
 
         return valid_periods
 
@@ -174,7 +264,7 @@ class WeatherService:
 
     @staticmethod
     @error_handler
-    async def fetch_weather(latitude: str, longitude: str, api_key: str) -> List[Dict[str, Any]]:
+    async def fetch_weather(latitude: str, longitude: str, api_key: str) -> list[WeatherReport]:
         """Fetch weather data from OpenWeatherMap API."""
         cache_key = (latitude, longitude)
 
@@ -198,14 +288,14 @@ class WeatherService:
             return result
 
     @staticmethod
-    def _format_weather_data(weather_data: WeatherData) -> List[Dict[str, Any]]:
+    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 [{
+        return [WeatherReport({
             "forecast_date": DateTimeFormatter.format_datetime(current_dt),
             "current_temp": int(round(current["temp"])),
             "current_feels_like": int(round(current["feels_like"])),
@@ -218,14 +308,14 @@ class WeatherService:
                 WeatherService._format_daily_period(period)
                 for period in daily_periods
             ],
-        }]
+        })]
 
     @staticmethod
-    def _format_daily_period(period: Dict[str, Any]) -> Dict[str, Any]:
+    def _format_daily_period(period: DailyWeather) -> WeatherPeriod:
         """Format a single daily weather period."""
         period_dt = datetime.fromtimestamp(period["dt"])
 
-        return {
+        return WeatherPeriod({
             "low": int(round(period["temp"]["min"])),
             "high": int(round(period["temp"]["max"])),
             "desc": period["weather"][0]["description"],
@@ -234,7 +324,7 @@ class WeatherService:
             "sunset": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunset"])),
             "pressure": period["pressure"],
             "period": DateTimeFormatter.format_date(period_dt),
-        }
+        })
 
 
 class PollenService:
@@ -242,7 +332,7 @@ class PollenService:
 
     @staticmethod
     @error_handler
-    async def fetch_pollen(zip_code: str) -> List[Dict[str, Any]]:
+    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]
@@ -265,10 +355,10 @@ class PollenService:
             raw_data = response.json()
 
             pollen_data = PollenData(raw_data)
-            result = [{
+            result = [PollenReport({
                 "forecast_date": pollen_data.forecast_date,
                 "periods": pollen_data.periods,
-            }]
+            })]
 
             pollen_cache[zip_code] = result
             return result
@@ -280,11 +370,11 @@ class DataAggregator:
     @staticmethod
     def build_daily_data(
         date: str,
-        pollen_periods: Dict[str, Dict[str, Any]],
-        weather_periods: Dict[str, Dict[str, Any]]
-    ) -> Dict[str, Any]:
+        pollen_periods: dict[str, PollenPeriod],
+        weather_periods: dict[str, WeatherPeriod]
+    ) -> DailyData:
         """Build daily data combining weather and pollen information."""
-        daily_data = {}
+        daily_data: DailyData = {}
 
         if date in pollen_periods:
             daily_data["pollen"] = pollen_periods[date]["index"]
@@ -305,44 +395,47 @@ class DataAggregator:
 
     @staticmethod
     def create_periods_lookup(
-        pollen_data: List[Dict[str, Any]],
-        weather_data: List[Dict[str, Any]]
-    ) -> Tuple[Dict[str, Dict[str, Any]], Dict[str, Dict[str, Any]]]:
+        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 = {}
-        weather_periods = {}
+        pollen_periods: dict[str, PollenPeriod] = {}
+        weather_periods: dict[str, WeatherPeriod] = {}
 
-        # Process pollen data
         if pollen_data and pollen_data[0].get("periods"):
             pollen_periods = {
                 p["period"]: p for p in pollen_data[0]["periods"]
             }
 
-        # Process weather data
         if weather_data and weather_data[0].get("periods"):
             weather_periods = {
                 p["period"]: p for p in weather_data[0]["periods"]
             }
 
-        # Add current weather data
+        # Add current weather data as a separate structure
+        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]
-            weather_periods["current"] = {
-                "period": "current",
+            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"],
-                "desc": data["current_desc"],
-                "sunrise": data["sunrise"],
-                "sunset": data["sunset"],
-            }
+            })
 
-        return pollen_periods, weather_periods
+        return pollen_periods, weather_periods, current_weather_info
 
 
 @app.get("/")
-async def get_weather_pollen_report(token: str) -> Dict[str, Any]:
+async def get_weather_pollen_report(token: str) -> FinalReport:
     """
     Get weather and pollen report.
 
@@ -355,51 +448,34 @@ async def get_weather_pollen_report(token: str) -> Dict[str, Any]:
     Raises:
         HTTPException: If authentication fails
     """
-    # Validate authentication
     if token != config.auth_token:
         raise HTTPException(status_code=403, detail="Unauthorized")
 
-    # Fetch data concurrently
     pollen_data, weather_data = await asyncio.gather(
         PollenService.fetch_pollen(config.zip_code),
         WeatherService.fetch_weather(config.latitude, config.longitude, config.weather_api_key),
     )
 
-    # Create period lookups
-    pollen_periods, weather_periods = DataAggregator.create_periods_lookup(
+    pollen_periods, weather_periods, current_weather_info = DataAggregator.create_periods_lookup(
         pollen_data, weather_data
     )
 
-    # Calculate date strings
     now = datetime.now()
     today_date = DateTimeFormatter.format_date(now)
     tomorrow_date = DateTimeFormatter.format_date(now + timedelta(days=1))
 
     # Build response
-    result = {
+    result: FinalReport = FinalReport({
         "fetched_at": DateTimeFormatter.format_datetime(now),
-        "current": {},
-        "today": {},
-        "tomorrow": {},
-    }
+        "current": current_weather_info,
+        "today": DailyData({}),
+        "tomorrow": DailyData({}),
+    })
 
-    # Add current weather data
-    if "current" in weather_periods:
-        current_weather = weather_periods["current"]
-        result["current"] = {
-            "temp": current_weather["temp"],
-            "feels_like": current_weather["feels_like"],
-            "desc": current_weather["desc"],
-            "humidity": current_weather["humidity"],
-            "pressure": current_weather["pressure"],
-        }
-
-    # Add today's data
     today_data = DataAggregator.build_daily_data(today_date, pollen_periods, weather_periods)
     if today_data:
         result["today"] = today_data
 
-    # Add tomorrow's data
     tomorrow_data = DataAggregator.build_daily_data(tomorrow_date, pollen_periods, weather_periods)
     if tomorrow_data:
         result["tomorrow"] = tomorrow_data
@@ -408,6 +484,6 @@ async def get_weather_pollen_report(token: str) -> Dict[str, Any]:
 
 
 @app.get("/health")
-async def health_check() -> Dict[str, str]:
+async def health_check() -> dict[str, str]:
     """Health check endpoint."""
     return {"status": "healthy"}