cached/weather_data.py

   1 #!/usr/bin/env python3
   2
   3 import datetime
   4 import json
   5 import logging
   6 import os
   7 import urllib.request
   8 from dataclasses import dataclass
   9 from typing import Any, List
  10
  11 from overrides import overrides
  12
  13 import argparse_utils
  14 import config
  15 import datetime_utils
  16 import list_utils
  17 import persistent
  18
  19 logger = logging.getLogger(__name__)
  20
  21 cfg = config.add_commandline_args(
  22     f'Cached Weather Data List ({__file__})',
  23     'Arguments controlling cached weather data',
  24 )
  25 cfg.add_argument(
  26     '--weather_data_cachefile',
  27     type=str,
  28     default=f'{os.environ["HOME"]}/cache/.weather_summary_cache',
  29     metavar='FILENAME',
  30     help='File in which to cache weather data',
  31 )
  32 cfg.add_argument(
  33     '--weather_data_stalest_acceptable',
  34     type=argparse_utils.valid_duration,
  35     default=datetime.timedelta(seconds=7200),  # 2 hours
  36     metavar='DURATION',
  37     help='Maximum acceptable age of cached data.  If zero, forces a refetch',
  38 )
  39
  40
  41 @dataclass
  42 class WeatherData:
  43     date: datetime.date  # The date
  44     high: float  # The predicted high in F
  45     low: float  # The predicted low in F
  46     precipitation_inches: float  # Number of inches of precipitation / day
  47     conditions: List[str]  # Conditions per ~3h window
  48     most_common_condition: str  # The most common condition
  49     icon: str  # An icon to represent it
  50
  51
  52 @persistent.persistent_autoloaded_singleton()  # type: ignore
  53 class CachedWeatherData(persistent.Persistent):
  54     def __init__(self, weather_data=None):
  55         if weather_data is not None:
  56             self.weather_data = weather_data
  57             return
  58         self.weather_data = {}
  59         icon_by_condition = {
  60             "Thunderstorm": "⚡",
  61             "Drizzle": "",
  62             "Rain": "☂️",
  63             "Snow": "❄️",
  64             "Clear": "☀️",
  65             "Clouds": "⛅",
  66             "Mist": "🌫",
  67             "Smoke": "🚬",
  68             "Haze": "🌥️",
  69             "Dust": "💨",
  70             "Fog": "🌁",
  71             "Sand": "🏜️",
  72             "Ash": "🌋",
  73             "Squall": "🌬",
  74             "Tornado": "🌪️",
  75         }
  76         now = datetime.datetime.now()
  77         dates = set()
  78         highs = {}
  79         lows = {}
  80         conditions = {}
  81         precip = {}
  82         param = "id=5786882"  # Bellevue, WA
  83         key = "c0b160c49743622f62a9cd3cda0270b3"
  84         www = urllib.request.urlopen(
  85             f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
  86         )
  87         response = www.read()
  88         www.close()
  89         parsed_json = json.loads(response)
  90         logger.debug(parsed_json)
  91         dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
  92         dates.add(dt)
  93         condition = parsed_json["weather"][0]["main"]
  94         icon = icon_by_condition.get(condition, '?')
  95         p = 0.0
  96         if 'rain' in parsed_json:
  97             if '3h' in parsed_json['rain']:
  98                 p += float(parsed_json['rain']['3h'])
  99             elif '1h' in parsed_json['rain']:
 100                 p += float(parsed_json['rain']['1h'])
 101         if 'snow' in parsed_json:
 102             if '3h' in parsed_json['snow']:
 103                 p += float(parsed_json['snow']['3h'])
 104             elif '1h' in parsed_json['snow']:
 105                 p += float(parsed_json['snow']['1h'])
 106         if dt == now.date() and now.hour > 18 and condition == 'Clear':
 107             icon = '🌙'
 108         self.weather_data[dt] = WeatherData(
 109             date=dt,
 110             high=float(parsed_json["main"]["temp_max"]),
 111             low=float(parsed_json["main"]["temp_min"]),
 112             precipitation_inches=p / 25.4,
 113             conditions=[condition],
 114             most_common_condition=condition,
 115             icon=icon,
 116         )
 117
 118         www = urllib.request.urlopen(
 119             f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
 120         )
 121         response = www.read()
 122         www.close()
 123         parsed_json = json.loads(response)
 124         logger.debug(parsed_json)
 125         count = parsed_json["cnt"]
 126         for x in range(count):
 127             data = parsed_json["list"][x]
 128             dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
 129             dt = dt.date()
 130             dates.add(dt)
 131             if dt not in highs:
 132                 highs[dt] = None
 133                 lows[dt] = None
 134                 conditions[dt] = []
 135             for temp in (
 136                 data["main"]["temp"],
 137                 data['main']['temp_min'],
 138                 data['main']['temp_max'],
 139             ):
 140                 if highs[dt] is None or temp > highs[dt]:
 141                     highs[dt] = temp
 142                 if lows[dt] is None or temp < lows[dt]:
 143                     lows[dt] = temp
 144             cond = data["weather"][0]["main"]
 145             precip[dt] = 0.0
 146             if 'rain' in parsed_json:
 147                 if '3h' in parsed_json['rain']:
 148                     precip[dt] += float(parsed_json['rain']['3h'])
 149                 elif '1h' in parsed_json['rain']:
 150                     precip[dt] += float(parsed_json['rain']['1h'])
 151             if 'snow' in parsed_json:
 152                 if '3h' in parsed_json['snow']:
 153                     precip[dt] += float(parsed_json['snow']['3h'])
 154                 elif '1h' in parsed_json['snow']:
 155                     precip[dt] += float(parsed_json['snow']['1h'])
 156             conditions[dt].append(cond)
 157
 158         today = datetime_utils.now_pacific().date()
 159         for dt in sorted(dates):
 160             if dt == today:
 161                 high = highs.get(dt, None)
 162                 if high is not None and self.weather_data[today].high < high:
 163                     self.weather_data[today].high = high
 164                 continue
 165             most_common_condition = list_utils.most_common(conditions[dt])
 166             icon = icon_by_condition.get(most_common_condition, '?')
 167             if dt == now.date() and now.hour > 18 and condition == 'Clear':
 168                 icon = '🌙'
 169             self.weather_data[dt] = WeatherData(
 170                 date=dt,
 171                 high=highs[dt],
 172                 low=lows[dt],
 173                 precipitation_inches=precip[dt] / 25.4,
 174                 conditions=conditions[dt],
 175                 most_common_condition=most_common_condition,
 176                 icon=icon,
 177             )
 178
 179     @classmethod
 180     @overrides
 181     def load(cls) -> Any:
 182         if persistent.was_file_written_within_n_seconds(
 183             config.config['weather_data_cachefile'],
 184             config.config['weather_data_stalest_acceptable'].total_seconds(),
 185         ):
 186             import pickle
 187
 188             with open(config.config['weather_data_cachefile'], 'rb') as rf:
 189                 weather_data = pickle.load(rf)
 190                 return cls(weather_data)
 191         return None
 192
 193     @overrides
 194     def save(self) -> bool:
 195         import pickle
 196
 197         with open(config.config['weather_data_cachefile'], 'wb') as wf:
 198             pickle.dump(
 199                 self.weather_data,
 200                 wf,
 201                 pickle.HIGHEST_PROTOCOL,
 202             )
 203         return True