diff --git a/main.py b/main.py index 272b884..2ebfe33 100644 --- a/main.py +++ b/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"}