cached/weather_data.py

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