2 # -*- coding: utf-8 -*-
4 # © Copyright 2021-2022, Scott Gasch
6 """How's the weather?"""
13 from dataclasses import dataclass
14 from typing import Any, Dict, List, Optional
16 from overrides import overrides
26 logger = logging.getLogger(__name__)
28 cfg = config.add_commandline_args(
29 f'Cached Weather Data List ({__file__})',
30 'Arguments controlling cached weather data',
33 '--weather_data_cachefile',
35 default=f'{os.environ["HOME"]}/cache/.weather_summary_cache',
37 help='File in which to cache weather data',
40 '--weather_data_stalest_acceptable',
41 type=argparse_utils.valid_duration,
42 default=datetime.timedelta(seconds=7200), # 2 hours
44 help='Maximum acceptable age of cached data. If zero, forces a refetch',
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
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
65 self.weather_data = {}
83 now = datetime.datetime.now()
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'
96 parsed_json = json.loads(response)
97 logger.debug(parsed_json)
98 dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
100 condition = parsed_json["weather"][0]["main"]
101 icon = icon_by_condition.get(condition, '?')
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':
115 self.weather_data[dt] = WeatherData(
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,
125 www = urllib.request.urlopen(
126 f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
128 response = www.read()
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')
143 data["main"]["temp"],
144 data['main']['temp_min'],
145 data['main']['temp_max'],
147 if highs[dt] is None or temp > highs[dt]:
149 if lows[dt] is None or temp < lows[dt]:
151 cond = data["weather"][0]["main"]
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)
165 today = datetime_utils.now_pacific().date()
166 for dt in sorted(dates):
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
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':
176 self.weather_data[dt] = WeatherData(
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,
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(),
195 with open(config.config['weather_data_cachefile'], 'rb') as rf:
196 weather_data = pickle.load(rf)
197 return cls(weather_data)
201 def save(self) -> bool:
204 with open(config.config['weather_data_cachefile'], 'wb') as wf:
208 pickle.HIGHEST_PROTOCOL,