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