#!/usr/bin/env python3 # -*- coding: utf-8 -*- # © Copyright 2021-2022, Scott Gasch """How's the weather?""" import datetime import json import logging import os import urllib.request from dataclasses import dataclass from typing import Any, Dict, List, Optional from overrides import overrides import argparse_utils import config import datetime_utils import list_utils import persistent import scott_secrets import type_utils logger = logging.getLogger(__name__) 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=f'{os.environ["HOME"]}/cache/.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 precipitation_inches: float # Number of inches of precipitation / day conditions: List[str] # Conditions per ~3h window most_common_condition: str # The most common condition icon: str # An icon to represent it @persistent.persistent_autoloaded_singleton() # type: ignore class CachedWeatherData(persistent.Persistent): def __init__(self, weather_data: Dict[datetime.date, WeatherData] = 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: Dict[datetime.date, Optional[float]] = {} lows: Dict[datetime.date, Optional[float]] = {} conditions: Dict[datetime.date, List[str]] = {} precip: Dict[datetime.date, float] = {} param = "id=5786882" # Bellevue, WA key = scott_secrets.OPEN_WEATHER_MAP_KEY 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) logger.debug(parsed_json) dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date() dates.add(dt) condition = parsed_json["weather"][0]["main"] icon = icon_by_condition.get(condition, '?') p = 0.0 if 'rain' in parsed_json: if '3h' in parsed_json['rain']: p += float(parsed_json['rain']['3h']) elif '1h' in parsed_json['rain']: p += float(parsed_json['rain']['1h']) if 'snow' in parsed_json: if '3h' in parsed_json['snow']: p += float(parsed_json['snow']['3h']) elif '1h' in parsed_json['snow']: p += float(parsed_json['snow']['1h']) 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"]), precipitation_inches=p / 25.4, 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) logger.debug(parsed_json) 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] = [] for temp in ( data["main"]["temp"], data['main']['temp_min'], data['main']['temp_max'], ): 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"] precip[dt] = 0.0 if 'rain' in parsed_json: if '3h' in parsed_json['rain']: precip[dt] += float(parsed_json['rain']['3h']) elif '1h' in parsed_json['rain']: precip[dt] += float(parsed_json['rain']['1h']) if 'snow' in parsed_json: if '3h' in parsed_json['snow']: precip[dt] += float(parsed_json['snow']['3h']) elif '1h' in parsed_json['snow']: precip[dt] += float(parsed_json['snow']['1h']) 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(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=type_utils.unwrap_optional(highs[dt]), low=type_utils.unwrap_optional(lows[dt]), precipitation_inches=precip[dt] / 25.4, conditions=conditions[dt], most_common_condition=most_common_condition, icon=icon, ) @classmethod @overrides def load(cls) -> Any: 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 @overrides def save(self) -> bool: import pickle with open(config.config['weather_data_cachefile'], 'wb') as wf: pickle.dump( self.weather_data, wf, pickle.HIGHEST_PROTOCOL, ) return True