cached/weather_data.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3
   4 # © Copyright 2021-2022, Scott Gasch
   5
   6 """A cache of weather data for Bellevue, WA.
   7 :class:`CachedWeatherData` class that derives from :class:`Persistent`
   8 so that, on creation, the decorator transparently pulls in data from
   9 disk, if possible, to avoid a network request.
  10 """
  11
  12 import datetime
  13 import json
  14 import logging
  15 import os
  16 import urllib.request
  17 from dataclasses import dataclass
  18 from typing import Any, Dict, List, Optional
  19
  20 from overrides import overrides
  21
  22 import argparse_utils
  23 import config
  24 import datetime_utils
  25 import list_utils
  26 import persistent
  27 import scott_secrets
  28 import type_utils
  29
  30 logger = logging.getLogger(__name__)
  31
  32 cfg = config.add_commandline_args(
  33     f'Cached Weather Data List ({__file__})',
  34     'Arguments controlling cached weather data',
  35 )
  36 cfg.add_argument(
  37     '--weather_data_cachefile',
  38     type=str,
  39     default=f'{os.environ["HOME"]}/cache/.weather_summary_cache',
  40     metavar='FILENAME',
  41     help='File in which to cache weather data',
  42 )
  43 cfg.add_argument(
  44     '--weather_data_stalest_acceptable',
  45     type=argparse_utils.valid_duration,
  46     default=datetime.timedelta(seconds=7200),  # 2 hours
  47     metavar='DURATION',
  48     help='Maximum acceptable age of cached data.  If zero, forces a refetch',
  49 )
  50
  51
  52 @dataclass
  53 class WeatherData:
  54     date: datetime.date
  55     """The date of the forecast"""
  56
  57     high: float
  58     """The predicted high temperature in F"""
  59
  60     low: float
  61     """The predicted low temperature in F"""
  62
  63     precipitation_inches: float
  64     """Number of inches of precipitation / day"""
  65
  66     conditions: List[str]
  67     """Conditions per ~3h window"""
  68
  69     most_common_condition: str
  70     """The most common condition of the day"""
  71
  72     icon: str
  73     """An icon representing the most common condition of the day"""
  74
  75
  76 @persistent.persistent_autoloaded_singleton()  # type: ignore
  77 class CachedWeatherData(persistent.Persistent):
  78     def __init__(self, weather_data: Dict[datetime.date, WeatherData] = None):
  79         """C'tor.  Do not pass a dict except for testing purposes.
  80
  81         The @persistent_autoloaded_singleton decorator handles
  82         invoking our load and save methods at construction time for
  83         you.
  84         """
  85
  86         if weather_data is not None:
  87             self.weather_data = weather_data
  88             return
  89         self.weather_data = {}
  90         icon_by_condition = {
  91             "Thunderstorm": "⚡",
  92             "Drizzle": "",
  93             "Rain": "☂️",
  94             "Snow": "❄️",
  95             "Clear": "☀️",
  96             "Clouds": "⛅",
  97             "Mist": "🌫",
  98             "Smoke": "🚬",
  99             "Haze": "🌥️",
 100             "Dust": "💨",
 101             "Fog": "🌁",
 102             "Sand": "🏜️",
 103             "Ash": "🌋",
 104             "Squall": "🌬",
 105             "Tornado": "🌪️",
 106         }
 107         now = datetime.datetime.now()
 108         dates = set()
 109         highs: Dict[datetime.date, Optional[float]] = {}
 110         lows: Dict[datetime.date, Optional[float]] = {}
 111         conditions: Dict[datetime.date, List[str]] = {}
 112         precip: Dict[datetime.date, float] = {}
 113         param = "id=5786882"  # Bellevue, WA
 114         key = scott_secrets.OPEN_WEATHER_MAP_KEY
 115         www = urllib.request.urlopen(
 116             f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
 117         )
 118         response = www.read()
 119         www.close()
 120         parsed_json = json.loads(response)
 121         logger.debug(parsed_json)
 122         dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
 123         dates.add(dt)
 124         condition = parsed_json["weather"][0]["main"]
 125         icon = icon_by_condition.get(condition, '?')
 126         p = 0.0
 127         if 'rain' in parsed_json:
 128             if '3h' in parsed_json['rain']:
 129                 p += float(parsed_json['rain']['3h'])
 130             elif '1h' in parsed_json['rain']:
 131                 p += float(parsed_json['rain']['1h'])
 132         if 'snow' in parsed_json:
 133             if '3h' in parsed_json['snow']:
 134                 p += float(parsed_json['snow']['3h'])
 135             elif '1h' in parsed_json['snow']:
 136                 p += float(parsed_json['snow']['1h'])
 137         if dt == now.date() and now.hour > 18 and condition == 'Clear':
 138             icon = '🌙'
 139         self.weather_data[dt] = WeatherData(
 140             date=dt,
 141             high=float(parsed_json["main"]["temp_max"]),
 142             low=float(parsed_json["main"]["temp_min"]),
 143             precipitation_inches=p / 25.4,
 144             conditions=[condition],
 145             most_common_condition=condition,
 146             icon=icon,
 147         )
 148
 149         www = urllib.request.urlopen(
 150             f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
 151         )
 152         response = www.read()
 153         www.close()
 154         parsed_json = json.loads(response)
 155         logger.debug(parsed_json)
 156         count = parsed_json["cnt"]
 157         for x in range(count):
 158             data = parsed_json["list"][x]
 159             dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
 160             dt = dt.date()
 161             dates.add(dt)
 162             if dt not in highs:
 163                 highs[dt] = None
 164                 lows[dt] = None
 165                 conditions[dt] = []
 166             for temp in (
 167                 data["main"]["temp"],
 168                 data['main']['temp_min'],
 169                 data['main']['temp_max'],
 170             ):
 171                 if highs[dt] is None or temp > highs[dt]:
 172                     highs[dt] = temp
 173                 if lows[dt] is None or temp < lows[dt]:
 174                     lows[dt] = temp
 175             cond = data["weather"][0]["main"]
 176             precip[dt] = 0.0
 177             if 'rain' in parsed_json:
 178                 if '3h' in parsed_json['rain']:
 179                     precip[dt] += float(parsed_json['rain']['3h'])
 180                 elif '1h' in parsed_json['rain']:
 181                     precip[dt] += float(parsed_json['rain']['1h'])
 182             if 'snow' in parsed_json:
 183                 if '3h' in parsed_json['snow']:
 184                     precip[dt] += float(parsed_json['snow']['3h'])
 185                 elif '1h' in parsed_json['snow']:
 186                     precip[dt] += float(parsed_json['snow']['1h'])
 187             conditions[dt].append(cond)
 188
 189         today = datetime_utils.now_pacific().date()
 190         for dt in sorted(dates):
 191             if dt == today:
 192                 high = highs.get(dt, None)
 193                 if high is not None and self.weather_data[today].high < high:
 194                     self.weather_data[today].high = high
 195                 continue
 196             most_common_condition = list_utils.most_common(conditions[dt])
 197             icon = icon_by_condition.get(most_common_condition, '?')
 198             if dt == now.date() and now.hour > 18 and condition == 'Clear':
 199                 icon = '🌙'
 200             self.weather_data[dt] = WeatherData(
 201                 date=dt,
 202                 high=type_utils.unwrap_optional(highs[dt]),
 203                 low=type_utils.unwrap_optional(lows[dt]),
 204                 precipitation_inches=precip[dt] / 25.4,
 205                 conditions=conditions[dt],
 206                 most_common_condition=most_common_condition,
 207                 icon=icon,
 208             )
 209
 210     @classmethod
 211     @overrides
 212     def load(cls) -> Any:
 213
 214         """Depending on whether we have fresh data persisted either uses that
 215         data to instantiate the class or makes an HTTP request to fetch the
 216         necessary data.
 217
 218         Note that because this is a subclass of Persistent this is taken
 219         care of automatically.
 220         """
 221
 222         if persistent.was_file_written_within_n_seconds(
 223             config.config['weather_data_cachefile'],
 224             config.config['weather_data_stalest_acceptable'].total_seconds(),
 225         ):
 226             import pickle
 227
 228             with open(config.config['weather_data_cachefile'], 'rb') as rf:
 229                 weather_data = pickle.load(rf)
 230                 return cls(weather_data)
 231         return None
 232
 233     @overrides
 234     def save(self) -> bool:
 235         """
 236         Saves the current data to disk if required.  Again, because this is
 237         a subclass of Persistent this is taken care of for you.
 238         """
 239
 240         import pickle
 241
 242         with open(config.config['weather_data_cachefile'], 'wb') as wf:
 243             pickle.dump(
 244                 self.weather_data,
 245                 wf,
 246                 pickle.HIGHEST_PROTOCOL,
 247             )
 248         return True