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.Persistent):
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 load(cls) -> Any:
214 """Depending on whether we have fresh data persisted either uses that
215 data to instantiate the class or makes an HTTP request to fetch the
218 Note that because this is a subclass of Persistent this is taken
219 care of automatically.
222 if persistent.was_file_written_within_n_seconds(
223 config.config['weather_data_cachefile'],
224 config.config['weather_data_stalest_acceptable'].total_seconds(),
228 with open(config.config['weather_data_cachefile'], 'rb') as rf:
229 weather_data = pickle.load(rf)
230 return cls(weather_data)
234 def save(self) -> bool:
236 Saves the current data to disk if required. Again, because this is
237 a subclass of Persistent this is taken care of for you.
242 with open(config.config['weather_data_cachefile'], 'wb') as wf:
246 pickle.HIGHEST_PROTOCOL,