cached/weather_data.py

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