diff --git a/main.py b/main.py index 6d334e9..ca79217 100644 --- a/main.py +++ b/main.py @@ -20,14 +20,12 @@ 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 @@ -38,12 +36,10 @@ class CurrentWeather(TypedDict): humidity: int weather: list[WeatherCondition] - class DailyTemperature(TypedDict): min: float max: float - class DailyWeather(TypedDict): dt: int sunrise: int @@ -53,21 +49,17 @@ 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 @@ -78,12 +70,10 @@ class WeatherPeriod(TypedDict): pressure: int period: str - class PollenPeriod(TypedDict): index: int period: str - class WeatherReport(TypedDict): forecast_date: str current_temp: int @@ -95,12 +85,10 @@ 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 @@ -108,7 +96,6 @@ class CurrentWeatherData(TypedDict): humidity: int pressure: int - class DailyData(TypedDict, total=False): pollen: int low: int @@ -119,14 +106,12 @@ 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) @@ -173,9 +158,7 @@ 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: @@ -251,27 +234,18 @@ 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: @@ -279,7 +253,6 @@ def error_handler(func): except Exception as e: logger.exception(f"Error in {func.__name__}: {e}") return [] - return wrapper @@ -288,9 +261,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[WeatherReport]: """Fetch weather data from OpenWeatherMap API.""" cache_key = (latitude, longitude) @@ -321,50 +292,36 @@ 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: @@ -395,14 +352,10 @@ 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 @@ -415,7 +368,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 = {} @@ -425,55 +378,54 @@ 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"]} - - current_weather_info = CurrentWeatherData( - { - "temp": 0, - "feels_like": 0, - "desc": "", - "humidity": 0, - "pressure": 0, + 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, + }) 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 @@ -497,37 +449,29 @@ 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