diff --git a/main.py b/main.py index ca79217..6d334e9 100644 --- a/main.py +++ b/main.py @@ -20,12 +20,14 @@ 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 @@ -36,10 +38,12 @@ class CurrentWeather(TypedDict): humidity: int weather: list[WeatherCondition] + class DailyTemperature(TypedDict): min: float max: float + class DailyWeather(TypedDict): dt: int sunrise: int @@ -49,17 +53,21 @@ class DailyWeather(TypedDict): 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 @@ -70,10 +78,12 @@ class WeatherPeriod(TypedDict): pressure: int period: str + class PollenPeriod(TypedDict): index: int period: str + class WeatherReport(TypedDict): forecast_date: str current_temp: int @@ -85,10 +95,12 @@ class WeatherReport(TypedDict): current_desc: str periods: list[WeatherPeriod] + class PollenReport(TypedDict): forecast_date: str periods: list[PollenPeriod] + class CurrentWeatherData(TypedDict): temp: int feels_like: int @@ -96,6 +108,7 @@ class CurrentWeatherData(TypedDict): humidity: int pressure: int + class DailyData(TypedDict, total=False): pollen: int low: int @@ -106,12 +119,14 @@ class DailyData(TypedDict, total=False): 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) @@ -158,7 +173,9 @@ class DateTimeFormatter: @staticmethod def format_datetime(dt: datetime) -> str: """Format datetime as date and time string.""" - return f"{DateTimeFormatter.format_date(dt)} {DateTimeFormatter.format_time(dt)}" + return ( + f"{DateTimeFormatter.format_date(dt)} {DateTimeFormatter.format_time(dt)}" + ) @staticmethod def relative_day_to_date(relative_day: str) -> str: @@ -234,18 +251,27 @@ class PollenData: if period_type in ["today", "tomorrow"]: 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) + pollen_percentage = int( + index_value / POLLEN_MAX_INDEX * POLLEN_PERCENTAGE_SCALE + ) - valid_periods.append(PollenPeriod({ - "index": pollen_percentage, - "period": DateTimeFormatter.relative_day_to_date(period_type), - })) + 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: @@ -253,6 +279,7 @@ def error_handler(func): except Exception as e: logger.exception(f"Error in {func.__name__}: {e}") return [] + return wrapper @@ -261,7 +288,9 @@ 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[WeatherReport]: """Fetch weather data from OpenWeatherMap API.""" cache_key = (latitude, longitude) @@ -292,36 +321,50 @@ class WeatherService: 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 - ], - })] + 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"])), - "high": int(round(period["temp"]["max"])), - "desc": period["weather"][0]["description"], - "humidity": period["humidity"], - "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), - }) + return WeatherPeriod( + { + "low": int(round(period["temp"]["min"])), + "high": int(round(period["temp"]["max"])), + "desc": period["weather"][0]["description"], + "humidity": period["humidity"], + "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: @@ -352,10 +395,14 @@ class PollenService: raw_data = response.json() pollen_data = PollenData(raw_data) - result = [PollenReport({ - "forecast_date": pollen_data.forecast_date, - "periods": pollen_data.periods, - })] + result = [ + PollenReport( + { + "forecast_date": pollen_data.forecast_date, + "periods": pollen_data.periods, + } + ) + ] pollen_cache[zip_code] = result return result @@ -368,7 +415,7 @@ class DataAggregator: def build_daily_data( date: str, pollen_periods: dict[str, PollenPeriod], - weather_periods: dict[str, WeatherPeriod] + weather_periods: dict[str, WeatherPeriod], ) -> DailyData: """Build daily data combining weather and pollen information.""" daily_data: DailyData = {} @@ -378,54 +425,55 @@ class DataAggregator: 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"], - }) + 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] + 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"] - } + 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"] - } + 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, - }) + 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"], - }) + 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 @@ -449,29 +497,37 @@ async def get_weather_pollen_report(token: str) -> FinalReport: pollen_data, weather_data = await asyncio.gather( PollenService.fetch_pollen(config.zip_code), - WeatherService.fetch_weather(config.latitude, config.longitude, config.weather_api_key), + WeatherService.fetch_weather( + config.latitude, config.longitude, config.weather_api_key + ), ) - pollen_periods, weather_periods, current_weather_info = DataAggregator.create_periods_lookup( - pollen_data, weather_data + pollen_periods, weather_periods, current_weather_info = ( + DataAggregator.create_periods_lookup(pollen_data, weather_data) ) now = datetime.now() today_date = DateTimeFormatter.format_date(now) tomorrow_date = DateTimeFormatter.format_date(now + timedelta(days=1)) - result: FinalReport = FinalReport({ - "fetched_at": DateTimeFormatter.format_datetime(now), - "current": current_weather_info, - "today": DailyData({}), - "tomorrow": DailyData({}), - }) + result: FinalReport = FinalReport( + { + "fetched_at": DateTimeFormatter.format_datetime(now), + "current": current_weather_info, + "today": DailyData({}), + "tomorrow": DailyData({}), + } + ) - today_data = DataAggregator.build_daily_data(today_date, pollen_periods, weather_periods) + today_data = DataAggregator.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 = DataAggregator.build_daily_data( + tomorrow_date, pollen_periods, weather_periods + ) if tomorrow_data: result["tomorrow"] = tomorrow_data