diff --git a/main.py b/main.py index 2ebfe33..272b884 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,6 @@ import asyncio from datetime import datetime, timedelta -from typing import TypedDict, Union +from typing import Dict, List, Optional, Any, Tuple import functools import logging import os @@ -11,6 +11,7 @@ from cachetools import TTLCache import httpx +# Configure logging logger = logging.getLogger("uvicorn.error") # Constants @@ -21,98 +22,6 @@ 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") @@ -140,6 +49,7 @@ class Config: return value +# Global config instance config = Config() @@ -184,7 +94,7 @@ class DateTimeFormatter: class WeatherData: """Data model for weather information.""" - def __init__(self, raw_data: WeatherApiResponse): + def __init__(self, raw_data: Dict[str, Any]): self.raw_data = raw_data self._validate_data() @@ -196,12 +106,12 @@ class WeatherData: raise ValueError(f"Missing required field in weather data: {field}") @property - def current(self) -> CurrentWeather: + def current(self) -> Dict[str, Any]: """Get current weather data.""" return self.raw_data["current"] @property - def daily_periods(self) -> list[DailyWeather]: + def daily_periods(self) -> List[Dict[str, Any]]: """Get daily forecast periods (limited to next 2 days).""" return self.raw_data["daily"][:2] @@ -209,7 +119,7 @@ class WeatherData: class PollenData: """Data model for pollen information.""" - def __init__(self, raw_data: PollenApiResponse): + def __init__(self, raw_data: Dict[str, Any]): self.raw_data = raw_data self._validate_data() @@ -227,22 +137,22 @@ class PollenData: return DateTimeFormatter.format_datetime(dt) @property - def periods(self) -> list[PollenPeriod]: + def periods(self) -> List[Dict[str, Any]]: """Get pollen periods for today and tomorrow.""" periods = self.raw_data["Location"].get("periods", []) - valid_periods: list[PollenPeriod] = [] + valid_periods = [] for period in periods: - period_type = str(period.get("Type", "")).lower().strip() + period_type = period.get("Type", "").lower().strip() if period_type in ["today", "tomorrow"]: - index_value = float(period.get("Index", 0)) + index_value = 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(PollenPeriod({ + valid_periods.append({ "index": pollen_percentage, "period": DateTimeFormatter.relative_day_to_date(period_type), - })) + }) return valid_periods @@ -264,7 +174,7 @@ class WeatherService: @staticmethod @error_handler - async def fetch_weather(latitude: str, longitude: str, api_key: str) -> list[WeatherReport]: + async def fetch_weather(latitude: str, longitude: str, api_key: str) -> List[Dict[str, Any]]: """Fetch weather data from OpenWeatherMap API.""" cache_key = (latitude, longitude) @@ -288,14 +198,14 @@ class WeatherService: return result @staticmethod - def _format_weather_data(weather_data: WeatherData) -> list[WeatherReport]: + def _format_weather_data(weather_data: WeatherData) -> List[Dict[str, Any]]: """Format weather data for API response.""" current = weather_data.current daily_periods = weather_data.daily_periods current_dt = datetime.fromtimestamp(current["dt"]) - return [WeatherReport({ + return [{ "forecast_date": DateTimeFormatter.format_datetime(current_dt), "current_temp": int(round(current["temp"])), "current_feels_like": int(round(current["feels_like"])), @@ -308,14 +218,14 @@ class WeatherService: WeatherService._format_daily_period(period) for period in daily_periods ], - })] + }] @staticmethod - def _format_daily_period(period: DailyWeather) -> WeatherPeriod: + def _format_daily_period(period: Dict[str, Any]) -> Dict[str, Any]: """Format a single daily weather period.""" period_dt = datetime.fromtimestamp(period["dt"]) - return WeatherPeriod({ + return { "low": int(round(period["temp"]["min"])), "high": int(round(period["temp"]["max"])), "desc": period["weather"][0]["description"], @@ -324,7 +234,7 @@ class WeatherService: "sunset": DateTimeFormatter.format_time(datetime.fromtimestamp(period["sunset"])), "pressure": period["pressure"], "period": DateTimeFormatter.format_date(period_dt), - }) + } class PollenService: @@ -332,7 +242,7 @@ class PollenService: @staticmethod @error_handler - async def fetch_pollen(zip_code: str) -> list[PollenReport]: + async def fetch_pollen(zip_code: str) -> List[Dict[str, Any]]: """Fetch pollen data from pollen.com API.""" if zip_code in pollen_cache: return pollen_cache[zip_code] @@ -355,10 +265,10 @@ class PollenService: raw_data = response.json() pollen_data = PollenData(raw_data) - result = [PollenReport({ + result = [{ "forecast_date": pollen_data.forecast_date, "periods": pollen_data.periods, - })] + }] pollen_cache[zip_code] = result return result @@ -370,11 +280,11 @@ class DataAggregator: @staticmethod def build_daily_data( date: str, - pollen_periods: dict[str, PollenPeriod], - weather_periods: dict[str, WeatherPeriod] - ) -> DailyData: + pollen_periods: Dict[str, Dict[str, Any]], + weather_periods: Dict[str, Dict[str, Any]] + ) -> Dict[str, Any]: """Build daily data combining weather and pollen information.""" - daily_data: DailyData = {} + daily_data = {} if date in pollen_periods: daily_data["pollen"] = pollen_periods[date]["index"] @@ -395,47 +305,44 @@ class DataAggregator: @staticmethod def create_periods_lookup( - pollen_data: list[PollenReport], - weather_data: list[WeatherReport] - ) -> tuple[dict[str, PollenPeriod], dict[str, WeatherPeriod], CurrentWeatherData]: + pollen_data: List[Dict[str, Any]], + weather_data: List[Dict[str, Any]] + ) -> Tuple[Dict[str, Dict[str, Any]], Dict[str, Dict[str, Any]]]: """Create lookup dictionaries for pollen and weather periods.""" - pollen_periods: dict[str, PollenPeriod] = {} - weather_periods: dict[str, WeatherPeriod] = {} + pollen_periods = {} + weather_periods = {} + # 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 as a separate structure - current_weather_info = CurrentWeatherData({ - "temp": 0, - "feels_like": 0, - "desc": "", - "humidity": 0, - "pressure": 0, - }) - + # Add current weather data if weather_data and weather_data[0]: data = weather_data[0] - current_weather_info = CurrentWeatherData({ + weather_periods["current"] = { + "period": "current", "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, current_weather_info + return pollen_periods, weather_periods @app.get("/") -async def get_weather_pollen_report(token: str) -> FinalReport: +async def get_weather_pollen_report(token: str) -> Dict[str, Any]: """ Get weather and pollen report. @@ -448,34 +355,51 @@ async def get_weather_pollen_report(token: str) -> FinalReport: 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), ) - pollen_periods, weather_periods, current_weather_info = DataAggregator.create_periods_lookup( + # Create period lookups + pollen_periods, weather_periods = 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: FinalReport = FinalReport({ + result = { "fetched_at": DateTimeFormatter.format_datetime(now), - "current": current_weather_info, - "today": DailyData({}), - "tomorrow": DailyData({}), - }) + "current": {}, + "today": {}, + "tomorrow": {}, + } + # 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 @@ -484,6 +408,6 @@ async def get_weather_pollen_report(token: str) -> FinalReport: @app.get("/health") -async def health_check() -> dict[str, str]: +async def health_check() -> Dict[str, str]: """Health check endpoint.""" return {"status": "healthy"}