cached/weather_data.py

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