in Bellevue, WA.
--- /dev/null
+#!/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
+
+
+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,
+ )
--- /dev/null
+#!/usr/bin/env python3
+
+from dataclasses import dataclass
+import datetime
+import logging
+import urllib.request
+
+import astral # type: ignore
+from astral.sun import sun # type: ignore
+from bs4 import BeautifulSoup # type: ignore
+import pytz
+
+import argparse_utils
+import config
+import datetime_utils
+import dateparse.dateparse_utils as dp
+import persistent
+import text_utils
+
+logger = logging.getLogger(__name__)
+
+cfg = config.add_commandline_args(
+ f'Cached Weather Forecast ({__file__})',
+ 'Arguments controlling detailed weather rendering'
+)
+cfg.add_argument(
+ '--weather_forecast_cachefile',
+ type=str,
+ default='/home/scott/.weather_forecast_cache',
+ metavar='FILENAME',
+ help='File in which to cache weather data'
+)
+cfg.add_argument(
+ '--weather_forecast_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 WeatherForecast:
+ date: datetime.date # The date
+ sunrise: datetime.datetime # Sunrise datetime
+ sunset: datetime.datetime # Sunset datetime
+ description: str # Textual description of weather
+
+
+class CachedDetailedWeatherForecast(object):
+ def __init__(self, forecasts = None):
+ if forecasts is not None:
+ self.forecasts = forecasts
+ return
+
+ now = datetime_utils.now_pacific()
+ self.forecasts = {}
+
+ # Ask the raspberry pi about the outside temperature.
+ www = urllib.request.urlopen(
+ "http://10.0.0.75/~pi/outside_temp"
+ )
+ current_temp = www.read().decode("utf-8")
+ current_temp = float(current_temp)
+ current_temp *= (9/5)
+ current_temp += 32.0
+ current_temp = round(current_temp)
+ www.close()
+
+ # Get a weather forecast for Bellevue.
+ www = urllib.request.urlopen(
+ "https://forecast.weather.gov/MapClick.php?lat=47.652775&lon=-122.170716"
+ )
+ forecast_response = www.read()
+ www.close()
+
+ soup = BeautifulSoup(forecast_response, "html.parser")
+ forecast = soup.find(id='detailed-forecast-body')
+ parser = dp.DateParser()
+
+ last_dt = now
+ dt = now
+ for (day, txt) in zip(
+ forecast.find_all('b'),
+ forecast.find_all(class_='col-sm-10 forecast-text')
+ ):
+ last_dt = dt
+ try:
+ dt = parser.parse(day.get_text())
+ except Exception:
+ dt = last_dt
+ assert dt is not None
+
+ # Compute sunrise/sunset times on dt.
+ city = astral.LocationInfo(
+ "Bellevue", "USA", "US/Pacific", 47.653, -122.171
+ )
+ s = sun(city.observer, date=dt, tzinfo=pytz.timezone("US/Pacific"))
+ sunrise = s['sunrise']
+ sunset = s['sunset']
+
+ if dt.date == now.date:
+ blurb = f'{day.get_text()}: The current outside tempterature is {current_temp}. ' + txt.get_text()
+ else:
+ blurb = f'{day.get_text()}: {txt.get_text()}'
+ blurb = text_utils.wrap_string(blurb, 80)
+
+ if dt.date() in self.forecasts:
+ self.forecasts[dt.date()].description += '\n' + blurb
+ else:
+ self.forecasts[dt.date()] = WeatherForecast(
+ date = dt,
+ sunrise = sunrise,
+ sunset = sunset,
+ description = blurb,
+ )
+
+ @classmethod
+ def load(cls):
+ if persistent.was_file_written_within_n_seconds(
+ config.config['weather_forecast_cachefile'],
+ config.config['weather_forecast_stalest_acceptable'].total_seconds(),
+ ):
+ import pickle
+ with open(config.config['weather_forecast_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_forecast_cachefile'], 'wb') as wf:
+ pickle.dump(
+ self.forecasts,
+ wf,
+ pickle.HIGHEST_PROTOCOL,
+ )
class ParseException(Exception):
"""An exception thrown during parsing because of unrecognized input."""
def __init__(self, message: str) -> None:
- logger.error(message)
self.message = message
def syntaxError(
self, recognizer, offendingSymbol, line, column, msg, e
):
- logger.error(msg)
raise ParseException(msg)
def reportAmbiguity(
projects / python_utils.git / commitdiff