cached/weather_data.py

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