Add some classes that use persistent to cache data about the weather

[python_utils.git] / cached / weather_data.py

diff --git a/cached/weather_data.py b/cached/weather_data.py
new file mode 100644 (file)
index 0000000..4c46448
--- /dev/null
+++ b/cached/weather_data.py
@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+
+from dataclasses import dataclass
+import datetime
+import json
+from typing import List
+import urllib.request
+
+import argparse_utils
+import config
+import datetime_utils
+import list_utils
+import persistent
+
+cfg = config.add_commandline_args(
+    f'Cached Weather Data List ({__file__})',
+    'Arguments controlling cached weather data',
+)
+cfg.add_argument(
+    '--weather_data_cachefile',
+    type=str,
+    default='/home/scott/.weather_summary_cache',
+    metavar='FILENAME',
+    help='File in which to cache weather data'
+)
+cfg.add_argument(
+    '--weather_data_stalest_acceptable',
+    type=argparse_utils.valid_duration,
+    default=datetime.timedelta(seconds=7200),   # 2 hours
+    metavar='DURATION',
+    help='Maximum acceptable age of cached data.  If zero, forces a refetch'
+)
+
+
+@dataclass
+class WeatherData:
+    date: datetime.date              # The date
+    high: float                      # The predicted high in F
+    low: float                       # The predicted low in F
+    conditions: List[str]            # Conditions per ~3h window
+    most_common_condition: str       # The most common condition
+    icon: str                        # An icon to represent it
+
+
[email protected]_autoloaded_singleton()
+class CachedWeatherData(persistent.Persistent):
+    def __init__(self,
+                 weather_data = None):
+        if weather_data is not None:
+            self.weather_data = weather_data
+            return
+        self.weather_data = {}
+        icon_by_condition = {
+            "Thunderstorm": "⚡",
+            "Drizzle": "",
+            "Rain": "☂️",
+            "Snow": "❄️",
+            "Clear": "☀️",
+            "Clouds": "⛅",
+            "Mist": "🌫",
+            "Smoke": "🚬",
+            "Haze": "🌥️",
+            "Dust": "💨",
+            "Fog": "🌁",
+            "Sand": "🏜️",
+            "Ash": "🌋",
+            "Squall": "🌬",
+            "Tornado": "🌪️"
+        }
+        now = datetime.datetime.now()
+        dates = set()
+        highs = {}
+        lows = {}
+        conditions = {}
+        param = "id=5786882"   # Bellevue, WA
+        key = "c0b160c49743622f62a9cd3cda0270b3"
+        www = urllib.request.urlopen(
+            f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
+        )
+        response = www.read()
+        www.close()
+        parsed_json = json.loads(response)
+        dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
+        dates.add(dt)
+        condition = parsed_json["weather"][0]["main"]
+        icon = icon_by_condition.get(condition, '?')
+        if dt == now.date() and now.hour > 18 and condition == 'Clear':
+            icon = '🌙'
+        self.weather_data[dt] = WeatherData(
+            date = dt,
+            high = float(parsed_json["main"]["temp_max"]),
+            low = float(parsed_json["main"]["temp_min"]),
+            conditions = [condition],
+            most_common_condition = condition,
+            icon = icon,
+        )
+
+        www = urllib.request.urlopen(
+            f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
+        )
+        response = www.read()
+        www.close()
+        parsed_json = json.loads(response)
+        count = parsed_json["cnt"]
+        for x in range(count):
+            data = parsed_json["list"][x]
+            dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
+            dt = dt.date()
+            dates.add(dt)
+            if dt not in highs:
+                highs[dt] = None
+                lows[dt] = None
+                conditions[dt] = []
+            temp = data["main"]["temp"]
+            if highs[dt] is None or temp > highs[dt]:
+                highs[dt] = temp
+            if lows[dt] is None or temp < lows[dt]:
+                lows[dt] = temp
+            cond = data["weather"][0]["main"]
+            conditions[dt].append(cond)
+
+        today = datetime_utils.now_pacific().date()
+        for dt in sorted(dates):
+            if dt == today:
+                high = highs.get(dt, None)
+                if (
+                        high is not None and
+                        self.weather_data[today].high < high
+                ):
+                    self.weather_data[today].high = high
+                continue
+            most_common_condition = list_utils.most_common_item(conditions[dt])
+            icon = icon_by_condition.get(most_common_condition, '?')
+            if dt == now.date() and now.hour > 18 and condition == 'Clear':
+                icon = '🌙'
+            self.weather_data[dt] = WeatherData(
+                date = dt,
+                high = highs[dt],
+                low = lows[dt],
+                conditions = conditions[dt],
+                most_common_condition = most_common_condition,
+                icon = icon
+            )
+
+    @classmethod
+    def load(cls):
+        if persistent.was_file_written_within_n_seconds(
+                config.config['weather_data_cachefile'],
+                config.config['weather_data_stalest_acceptable'].total_seconds(),
+        ):
+            import pickle
+            with open(config.config['weather_data_cachefile'], 'rb') as rf:
+                weather_data = pickle.load(rf)
+                return cls(weather_data)
+        return None
+
+    def save(self):
+        import pickle
+        with open(config.config['weather_data_cachefile'], 'wb') as wf:
+            pickle.dump(
+                self.weather_data,
+                wf,
+                pickle.HIGHEST_PROTOCOL,
+            )