#!/usr/bin/env python3 # -*- coding: utf-8 -*- # © Copyright 2021-2022, Scott Gasch """A cache of weather data for Bellevue, WA. :class:`CachedWeatherData` class that derives from :class:`Persistent` so that, on creation, the decorator transparently pulls in data from disk, if possible, to avoid a network request. """ import datetime import json import logging import os import urllib.request from dataclasses import dataclass from typing import Any, Dict, List, Optional from overrides import overrides import argparse_utils import config import datetime_utils import list_utils import persistent import scott_secrets import type_utils logger = logging.getLogger(__name__) cfg = config.add_commandline_args( f'Cached Weather Data List ({__file__})', 'Arguments controlling cached weather data', ) cfg.add_argument( '--weather_data_cachefile', type=str, default=f'{os.environ["HOME"]}/cache/.weather_summary_cache', metavar='FILENAME', help='File in which to cache weather data', ) cfg.add_argument( '--weather_data_stalest_acceptable', type=argparse_utils.valid_duration, default=datetime.timedelta(seconds=7200), # 2 hours metavar='DURATION', help='Maximum acceptable age of cached data. If zero, forces a refetch', ) @dataclass class WeatherData: date: datetime.date """The date of the forecast""" high: float """The predicted high temperature in F""" low: float """The predicted low temperature in F""" precipitation_inches: float """Number of inches of precipitation / day""" conditions: List[str] """Conditions per ~3h window""" most_common_condition: str """The most common condition of the day""" icon: str """An icon representing the most common condition of the day""" @persistent.persistent_autoloaded_singleton() # type: ignore class CachedWeatherData(persistent.PicklingFileBasedPersistent): def __init__(self, weather_data: Dict[datetime.date, WeatherData] = None): """C'tor. Do not pass a dict except for testing purposes. The @persistent_autoloaded_singleton decorator handles invoking our load and save methods at construction time for you. """ if weather_data is not None: self.weather_data = weather_data return self.weather_data = {} icon_by_condition = { "Thunderstorm": "⚡", "Drizzle": "", "Rain": "☂️", "Snow": "❄️", "Clear": "☀️", "Clouds": "⛅", "Mist": "🌫", "Smoke": "🚬", "Haze": "🌥️", "Dust": "💨", "Fog": "🌁", "Sand": "🏜️", "Ash": "🌋", "Squall": "🌬", "Tornado": "🌪️", } now = datetime.datetime.now() dates = set() highs: Dict[datetime.date, Optional[float]] = {} lows: Dict[datetime.date, Optional[float]] = {} conditions: Dict[datetime.date, List[str]] = {} precip: Dict[datetime.date, float] = {} param = "id=5786882" # Bellevue, WA key = scott_secrets.OPEN_WEATHER_MAP_KEY www = urllib.request.urlopen( f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial' ) response = www.read() www.close() parsed_json = json.loads(response) logger.debug(parsed_json) dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date() dates.add(dt) condition = parsed_json["weather"][0]["main"] icon = icon_by_condition.get(condition, '?') p = 0.0 if 'rain' in parsed_json: if '3h' in parsed_json['rain']: p += float(parsed_json['rain']['3h']) elif '1h' in parsed_json['rain']: p += float(parsed_json['rain']['1h']) if 'snow' in parsed_json: if '3h' in parsed_json['snow']: p += float(parsed_json['snow']['3h']) elif '1h' in parsed_json['snow']: p += float(parsed_json['snow']['1h']) if dt == now.date() and now.hour > 18 and condition == 'Clear': icon = '🌙' self.weather_data[dt] = WeatherData( date=dt, high=float(parsed_json["main"]["temp_max"]), low=float(parsed_json["main"]["temp_min"]), precipitation_inches=p / 25.4, conditions=[condition], most_common_condition=condition, icon=icon, ) www = urllib.request.urlopen( f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial" ) response = www.read() www.close() parsed_json = json.loads(response) logger.debug(parsed_json) count = parsed_json["cnt"] for x in range(count): data = parsed_json["list"][x] dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S') dt = dt.date() dates.add(dt) if dt not in highs: highs[dt] = None lows[dt] = None conditions[dt] = [] for temp in ( data["main"]["temp"], data['main']['temp_min'], data['main']['temp_max'], ): if highs[dt] is None or temp > highs[dt]: highs[dt] = temp if lows[dt] is None or temp < lows[dt]: lows[dt] = temp cond = data["weather"][0]["main"] precip[dt] = 0.0 if 'rain' in parsed_json: if '3h' in parsed_json['rain']: precip[dt] += float(parsed_json['rain']['3h']) elif '1h' in parsed_json['rain']: precip[dt] += float(parsed_json['rain']['1h']) if 'snow' in parsed_json: if '3h' in parsed_json['snow']: precip[dt] += float(parsed_json['snow']['3h']) elif '1h' in parsed_json['snow']: precip[dt] += float(parsed_json['snow']['1h']) conditions[dt].append(cond) today = datetime_utils.now_pacific().date() for dt in sorted(dates): if dt == today: high = highs.get(dt, None) if high is not None and self.weather_data[today].high < high: self.weather_data[today].high = high continue most_common_condition = list_utils.most_common(conditions[dt]) icon = icon_by_condition.get(most_common_condition, '?') if dt == now.date() and now.hour > 18 and condition == 'Clear': icon = '🌙' self.weather_data[dt] = WeatherData( date=dt, high=type_utils.unwrap_optional(highs[dt]), low=type_utils.unwrap_optional(lows[dt]), precipitation_inches=precip[dt] / 25.4, conditions=conditions[dt], most_common_condition=most_common_condition, icon=icon, ) @staticmethod @overrides def get_filename() -> str: return config.config['weather_data_cachefile'] @staticmethod @overrides def should_we_save_data(filename: str) -> bool: return True @staticmethod @overrides def should_we_load_data(filename: str) -> bool: return persistent.was_file_written_within_n_seconds( filename, config.config['weather_data_stalest_acceptable'].total_seconds(), ) @overrides def get_persistent_data(self) -> Any: return self.weather_data