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         if weather_data is not None:
  63             self.weather_data = weather_data
  64             return
  65         self.weather_data = {}
  66         icon_by_condition = {
  67             "Thunderstorm": "⚡",
  68             "Drizzle": "",
  69             "Rain": "☂️",
  70             "Snow": "❄️",
  71             "Clear": "☀️",
  72             "Clouds": "⛅",
  73             "Mist": "🌫",
  74             "Smoke": "🚬",
  75             "Haze": "🌥️",
  76             "Dust": "💨",
  77             "Fog": "🌁",
  78             "Sand": "🏜️",
  79             "Ash": "🌋",
  80             "Squall": "🌬",
  81             "Tornado": "🌪️",
  82         }
  83         now = datetime.datetime.now()
  84         dates = set()
  85         highs: Dict[datetime.date, Optional[float]] = {}
  86         lows: Dict[datetime.date, Optional[float]] = {}
  87         conditions: Dict[datetime.date, List[str]] = {}
  88         precip: Dict[datetime.date, float] = {}
  89         param = "id=5786882"  # Bellevue, WA
  90         key = scott_secrets.OPEN_WEATHER_MAP_KEY
  91         www = urllib.request.urlopen(
  92             f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
  93         )
  94         response = www.read()
  95         www.close()
  96         parsed_json = json.loads(response)
  97         logger.debug(parsed_json)
  98         dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
  99         dates.add(dt)
 100         condition = parsed_json["weather"][0]["main"]
 101         icon = icon_by_condition.get(condition, '?')
 102         p = 0.0
 103         if 'rain' in parsed_json:
 104             if '3h' in parsed_json['rain']:
 105                 p += float(parsed_json['rain']['3h'])
 106             elif '1h' in parsed_json['rain']:
 107                 p += float(parsed_json['rain']['1h'])
 108         if 'snow' in parsed_json:
 109             if '3h' in parsed_json['snow']:
 110                 p += float(parsed_json['snow']['3h'])
 111             elif '1h' in parsed_json['snow']:
 112                 p += float(parsed_json['snow']['1h'])
 113         if dt == now.date() and now.hour > 18 and condition == 'Clear':
 114             icon = '🌙'
 115         self.weather_data[dt] = WeatherData(
 116             date=dt,
 117             high=float(parsed_json["main"]["temp_max"]),
 118             low=float(parsed_json["main"]["temp_min"]),
 119             precipitation_inches=p / 25.4,
 120             conditions=[condition],
 121             most_common_condition=condition,
 122             icon=icon,
 123         )
 124
 125         www = urllib.request.urlopen(
 126             f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
 127         )
 128         response = www.read()
 129         www.close()
 130         parsed_json = json.loads(response)
 131         logger.debug(parsed_json)
 132         count = parsed_json["cnt"]
 133         for x in range(count):
 134             data = parsed_json["list"][x]
 135             dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
 136             dt = dt.date()
 137             dates.add(dt)
 138             if dt not in highs:
 139                 highs[dt] = None
 140                 lows[dt] = None
 141                 conditions[dt] = []
 142             for temp in (
 143                 data["main"]["temp"],
 144                 data['main']['temp_min'],
 145                 data['main']['temp_max'],
 146             ):
 147                 if highs[dt] is None or temp > highs[dt]:
 148                     highs[dt] = temp
 149                 if lows[dt] is None or temp < lows[dt]:
 150                     lows[dt] = temp
 151             cond = data["weather"][0]["main"]
 152             precip[dt] = 0.0
 153             if 'rain' in parsed_json:
 154                 if '3h' in parsed_json['rain']:
 155                     precip[dt] += float(parsed_json['rain']['3h'])
 156                 elif '1h' in parsed_json['rain']:
 157                     precip[dt] += float(parsed_json['rain']['1h'])
 158             if 'snow' in parsed_json:
 159                 if '3h' in parsed_json['snow']:
 160                     precip[dt] += float(parsed_json['snow']['3h'])
 161                 elif '1h' in parsed_json['snow']:
 162                     precip[dt] += float(parsed_json['snow']['1h'])
 163             conditions[dt].append(cond)
 164
 165         today = datetime_utils.now_pacific().date()
 166         for dt in sorted(dates):
 167             if dt == today:
 168                 high = highs.get(dt, None)
 169                 if high is not None and self.weather_data[today].high < high:
 170                     self.weather_data[today].high = high
 171                 continue
 172             most_common_condition = list_utils.most_common(conditions[dt])
 173             icon = icon_by_condition.get(most_common_condition, '?')
 174             if dt == now.date() and now.hour > 18 and condition == 'Clear':
 175                 icon = '🌙'
 176             self.weather_data[dt] = WeatherData(
 177                 date=dt,
 178                 high=type_utils.unwrap_optional(highs[dt]),
 179                 low=type_utils.unwrap_optional(lows[dt]),
 180                 precipitation_inches=precip[dt] / 25.4,
 181                 conditions=conditions[dt],
 182                 most_common_condition=most_common_condition,
 183                 icon=icon,
 184             )
 185
 186     @classmethod
 187     @overrides
 188     def load(cls) -> Any:
 189         if persistent.was_file_written_within_n_seconds(
 190             config.config['weather_data_cachefile'],
 191             config.config['weather_data_stalest_acceptable'].total_seconds(),
 192         ):
 193             import pickle
 194
 195             with open(config.config['weather_data_cachefile'], 'rb') as rf:
 196                 weather_data = pickle.load(rf)
 197                 return cls(weather_data)
 198         return None
 199
 200     @overrides
 201     def save(self) -> bool:
 202         import pickle
 203
 204         with open(config.config['weather_data_cachefile'], 'wb') as wf:
 205             pickle.dump(
 206                 self.weather_data,
 207                 wf,
 208                 pickle.HIGHEST_PROTOCOL,
 209             )
 210         return True