cached/weather_data.py

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