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 """C'tor. Do not pass a dict except for testing purposes.
64 The @persistent_autoloaded_singleton decorator handles
65 invoking our load and save methods at construction time for
69 if weather_data is not None:
70 self.weather_data = weather_data
72 self.weather_data = {}
90 now = datetime.datetime.now()
92 highs: Dict[datetime.date, Optional[float]] = {}
93 lows: Dict[datetime.date, Optional[float]] = {}
94 conditions: Dict[datetime.date, List[str]] = {}
95 precip: Dict[datetime.date, float] = {}
96 param = "id=5786882" # Bellevue, WA
97 key = scott_secrets.OPEN_WEATHER_MAP_KEY
98 www = urllib.request.urlopen(
99 f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
101 response = www.read()
103 parsed_json = json.loads(response)
104 logger.debug(parsed_json)
105 dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
107 condition = parsed_json["weather"][0]["main"]
108 icon = icon_by_condition.get(condition, '?')
110 if 'rain' in parsed_json:
111 if '3h' in parsed_json['rain']:
112 p += float(parsed_json['rain']['3h'])
113 elif '1h' in parsed_json['rain']:
114 p += float(parsed_json['rain']['1h'])
115 if 'snow' in parsed_json:
116 if '3h' in parsed_json['snow']:
117 p += float(parsed_json['snow']['3h'])
118 elif '1h' in parsed_json['snow']:
119 p += float(parsed_json['snow']['1h'])
120 if dt == now.date() and now.hour > 18 and condition == 'Clear':
122 self.weather_data[dt] = WeatherData(
124 high=float(parsed_json["main"]["temp_max"]),
125 low=float(parsed_json["main"]["temp_min"]),
126 precipitation_inches=p / 25.4,
127 conditions=[condition],
128 most_common_condition=condition,
132 www = urllib.request.urlopen(
133 f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
135 response = www.read()
137 parsed_json = json.loads(response)
138 logger.debug(parsed_json)
139 count = parsed_json["cnt"]
140 for x in range(count):
141 data = parsed_json["list"][x]
142 dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
150 data["main"]["temp"],
151 data['main']['temp_min'],
152 data['main']['temp_max'],
154 if highs[dt] is None or temp > highs[dt]:
156 if lows[dt] is None or temp < lows[dt]:
158 cond = data["weather"][0]["main"]
160 if 'rain' in parsed_json:
161 if '3h' in parsed_json['rain']:
162 precip[dt] += float(parsed_json['rain']['3h'])
163 elif '1h' in parsed_json['rain']:
164 precip[dt] += float(parsed_json['rain']['1h'])
165 if 'snow' in parsed_json:
166 if '3h' in parsed_json['snow']:
167 precip[dt] += float(parsed_json['snow']['3h'])
168 elif '1h' in parsed_json['snow']:
169 precip[dt] += float(parsed_json['snow']['1h'])
170 conditions[dt].append(cond)
172 today = datetime_utils.now_pacific().date()
173 for dt in sorted(dates):
175 high = highs.get(dt, None)
176 if high is not None and self.weather_data[today].high < high:
177 self.weather_data[today].high = high
179 most_common_condition = list_utils.most_common(conditions[dt])
180 icon = icon_by_condition.get(most_common_condition, '?')
181 if dt == now.date() and now.hour > 18 and condition == 'Clear':
183 self.weather_data[dt] = WeatherData(
185 high=type_utils.unwrap_optional(highs[dt]),
186 low=type_utils.unwrap_optional(lows[dt]),
187 precipitation_inches=precip[dt] / 25.4,
188 conditions=conditions[dt],
189 most_common_condition=most_common_condition,
195 def load(cls) -> Any:
197 """Depending on whether we have fresh data persisted either uses that
198 data to instantiate the class or makes an HTTP request to fetch the
201 Note that because this is a subclass of Persistent this is taken
202 care of automatically.
205 if persistent.was_file_written_within_n_seconds(
206 config.config['weather_data_cachefile'],
207 config.config['weather_data_stalest_acceptable'].total_seconds(),
211 with open(config.config['weather_data_cachefile'], 'rb') as rf:
212 weather_data = pickle.load(rf)
213 return cls(weather_data)
217 def save(self) -> bool:
219 Saves the current data to disk if required. Again, because this is
220 a subclass of Persistent this is taken care of for you.
225 with open(config.config['weather_data_cachefile'], 'wb') as wf:
229 pickle.HIGHEST_PROTOCOL,