2 # -*- coding: utf-8 -*-
4 # © Copyright 2021-2022, Scott Gasch
6 """A cache of weather data for Bellevue, WA.
7 :class:`CachedWeatherData` class that derives from :class:`Persistent`
8 so that, on creation, the decorator transparently pulls in data from
9 disk, if possible, to avoid a network request.
17 from dataclasses import dataclass
18 from typing import Any, Dict, List, Optional
20 from overrides import overrides
30 logger = logging.getLogger(__name__)
32 cfg = config.add_commandline_args(
33 f'Cached Weather Data List ({__file__})',
34 'Arguments controlling cached weather data',
37 '--weather_data_cachefile',
39 default=f'{os.environ["HOME"]}/cache/.weather_summary_cache',
41 help='File in which to cache weather data',
44 '--weather_data_stalest_acceptable',
45 type=argparse_utils.valid_duration,
46 default=datetime.timedelta(seconds=7200), # 2 hours
48 help='Maximum acceptable age of cached data. If zero, forces a refetch',
55 """The date of the forecast"""
58 """The predicted high temperature in F"""
61 """The predicted low temperature in F"""
63 precipitation_inches: float
64 """Number of inches of precipitation / day"""
67 """Conditions per ~3h window"""
69 most_common_condition: str
70 """The most common condition of the day"""
73 """An icon representing the most common condition of the day"""
76 @persistent.persistent_autoloaded_singleton() # type: ignore
77 class CachedWeatherData(persistent.PicklingFileBasedPersistent):
78 def __init__(self, weather_data: Dict[datetime.date, WeatherData] = None):
79 """C'tor. Do not pass a dict except for testing purposes.
81 The @persistent_autoloaded_singleton decorator handles
82 invoking our load and save methods at construction time for
86 if weather_data is not None:
87 self.weather_data = weather_data
89 self.weather_data = {}
107 now = datetime.datetime.now()
109 highs: Dict[datetime.date, Optional[float]] = {}
110 lows: Dict[datetime.date, Optional[float]] = {}
111 conditions: Dict[datetime.date, List[str]] = {}
112 precip: Dict[datetime.date, float] = {}
113 param = "id=5786882" # Bellevue, WA
114 key = scott_secrets.OPEN_WEATHER_MAP_KEY
115 www = urllib.request.urlopen(
116 f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
118 response = www.read()
120 parsed_json = json.loads(response)
121 logger.debug(parsed_json)
122 dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
124 condition = parsed_json["weather"][0]["main"]
125 icon = icon_by_condition.get(condition, '?')
127 if 'rain' in parsed_json:
128 if '3h' in parsed_json['rain']:
129 p += float(parsed_json['rain']['3h'])
130 elif '1h' in parsed_json['rain']:
131 p += float(parsed_json['rain']['1h'])
132 if 'snow' in parsed_json:
133 if '3h' in parsed_json['snow']:
134 p += float(parsed_json['snow']['3h'])
135 elif '1h' in parsed_json['snow']:
136 p += float(parsed_json['snow']['1h'])
137 if dt == now.date() and now.hour > 18 and condition == 'Clear':
139 self.weather_data[dt] = WeatherData(
141 high=float(parsed_json["main"]["temp_max"]),
142 low=float(parsed_json["main"]["temp_min"]),
143 precipitation_inches=p / 25.4,
144 conditions=[condition],
145 most_common_condition=condition,
149 www = urllib.request.urlopen(
150 f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
152 response = www.read()
154 parsed_json = json.loads(response)
155 logger.debug(parsed_json)
156 count = parsed_json["cnt"]
157 for x in range(count):
158 data = parsed_json["list"][x]
159 dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
167 data["main"]["temp"],
168 data['main']['temp_min'],
169 data['main']['temp_max'],
171 if highs[dt] is None or temp > highs[dt]:
173 if lows[dt] is None or temp < lows[dt]:
175 cond = data["weather"][0]["main"]
177 if 'rain' in parsed_json:
178 if '3h' in parsed_json['rain']:
179 precip[dt] += float(parsed_json['rain']['3h'])
180 elif '1h' in parsed_json['rain']:
181 precip[dt] += float(parsed_json['rain']['1h'])
182 if 'snow' in parsed_json:
183 if '3h' in parsed_json['snow']:
184 precip[dt] += float(parsed_json['snow']['3h'])
185 elif '1h' in parsed_json['snow']:
186 precip[dt] += float(parsed_json['snow']['1h'])
187 conditions[dt].append(cond)
189 today = datetime_utils.now_pacific().date()
190 for dt in sorted(dates):
192 high = highs.get(dt, None)
193 if high is not None and self.weather_data[today].high < high:
194 self.weather_data[today].high = high
196 most_common_condition = list_utils.most_common(conditions[dt])
197 icon = icon_by_condition.get(most_common_condition, '?')
198 if dt == now.date() and now.hour > 18 and condition == 'Clear':
200 self.weather_data[dt] = WeatherData(
202 high=type_utils.unwrap_optional(highs[dt]),
203 low=type_utils.unwrap_optional(lows[dt]),
204 precipitation_inches=precip[dt] / 25.4,
205 conditions=conditions[dt],
206 most_common_condition=most_common_condition,
212 def get_filename() -> str:
213 return config.config['weather_data_cachefile']
217 def should_we_save_data(filename: str) -> bool:
222 def should_we_load_data(filename: str) -> bool:
223 return persistent.was_file_written_within_n_seconds(
225 config.config['weather_data_stalest_acceptable'].total_seconds(),
229 def get_persistent_data(self) -> Any:
230 return self.weather_data