#!/usr/bin/env python3 import logging import json import urllib.request import urllib.error import urllib.parse from datetime import datetime from collections import defaultdict from typing import Dict, List import file_writer import renderer import kiosk_secrets as secrets logger = logging.getLogger(__name__) class weather_renderer(renderer.abstaining_renderer): """A renderer to fetch forecast from wunderground.""" def __init__(self, name_to_timeout_dict: Dict[str, int], file_prefix: str) -> None: super().__init__(name_to_timeout_dict) self.file_prefix = file_prefix @staticmethod def pick_icon(conditions: List[str], rain: List[float], snow: List[float]) -> str: # rain snow clouds sun # fog.gif # hazy.gif # clear.gif # mostlycloudy.gif F F 6+ X # partlycloudy.gif F F 4+ 4- # cloudy.gif # partlysunny.gif F F X 5+ # mostlysunny.gif F F X 6+ # rain.gif T F X X # sleet.gif T T X X # flurries.gif F T X X (<1") # snow.gif F T X X (else) # sunny.gif F F X 7+ # tstorms.gif seen_rain = False seen_snow = False cloud_count = 0 clear_count = 0 total_snow = 0.0 count = min(len(conditions), len(rain), len(snow)) for x in range(0, count): seen_rain = rain[x] > 0 seen_snow = snow[x] > 0 total_snow += snow[x] txt = conditions[x].lower() if "cloud" in txt: cloud_count += 1 if "clear" in txt or "sun" in txt: clear_count += 1 if seen_rain and seen_snow: if total_snow < 10: return "sleet.gif" else: return "snow.gif" if seen_snow: if total_snow < 10: return "flurries.gif" else: return "snow.gif" if seen_rain: return "rain.gif" if cloud_count >= 6: return "mostlycloudy.gif" elif cloud_count >= 4: return "partlycloudy.gif" if clear_count >= 7: return "sunny.gif" elif clear_count >= 6: return "mostlysunny.gif" elif clear_count >= 4: return "partlysunny.gif" return "clear.gif" def periodic_render(self, key: str) -> bool: return self.fetch_weather() def fetch_weather(self) -> bool: if self.file_prefix == "stevens": text_location = "Stevens Pass, WA" param = "lat=47.7322&lon=-121.1025" elif self.file_prefix == "telma": text_location = "Telma, WA" param = "lat=47.84&lon=-120.81" else: text_location = "Bellevue, WA" param = "id=5786882" secret = secrets.openweather_key url = f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={secret}&units=imperial" logger.info(f"GETting {url}") www = urllib.request.urlopen(url) response = www.read() www.close() if www.getcode() != 200: logger.error("Bad response: {response}") raise Exception(response) parsed_json = json.loads(response) logger.info("URL read ok") # https://openweathermap.org/forecast5 # {"cod":"200", # "message":0.0036, # "cnt":40, # "list":[ # {"dt":1485799200, # "main":{"temp":261.45,"temp_min":259.086,"temp_max":261.45,"pressure":1023.48,"sea_level":1045.39,"grnd_level":1023.48,"humidity":79,"temp_kf":2.37}, # "weather":[ # {"id":800,"main":"Clear","description":"clear sky","icon":"02n"} # ], # "clouds":{"all":8}, # "wind":{"speed":4.77,"deg":232.505}, # "snow":{}, # "sys":{"pod":"n"}, # "dt_txt":"2017-01-30 18:00:00" # }, # {"dt":1485810000,.... with file_writer.file_writer(f"weather-{self.file_prefix}_3_10800.html") as f: f.write( f"""

Upcoming weather at {text_location}:


""" ) f.write( """
""" ) count = parsed_json["cnt"] ts = {} highs = {} lows = {} wind: Dict[str, List[float]] = defaultdict(list) conditions: Dict[str, List[str]] = defaultdict(list) rain: Dict[str, List[float]] = defaultdict(list) snow: Dict[str, List[float]] = defaultdict(list) precip: Dict[str, List[float]] = defaultdict(list) for x in range(0, count): data = parsed_json["list"][x] dt = data["dt_txt"] # 2019-10-07 18:00:00 (date, time) = dt.split(" ") _ = data["dt"] if _ not in ts or _ > ts[date]: ts[date] = _ temp = data["main"]["temp"] # High and low temp if date not in highs or highs[date] < temp: highs[date] = temp if date not in lows or lows[date] > temp: lows[date] = temp # Windspeed and conditions wind[date].append(data["wind"]["speed"]) conditions[date].append(data["weather"][0]["main"]) # 3h precipitation (rain / snow) if "rain" in data and "3h" in data["rain"]: rain[date].append(data["rain"]["3h"]) else: rain[date].append(0) if "snow" in data and "3h" in data["snow"]: snow[date].append(data["snow"]["3h"]) else: snow[date].append(0) # {u'clouds': {u'all': 0}, # u'sys': {u'pod': u'd'}, # u'dt_txt': u'2019-10-09 21:00:00', # u'weather': [ # {u'main': u'Clear', # u'id': 800, # u'icon': u'01d', # u'description': u'clear sky'} # ], # u'dt': 1570654800, # u'main': { # u'temp_kf': 0, # u'temp': 54.74, # u'grnd_level': 1018.95, # u'temp_max': 54.74, # u'sea_level': 1026.46, # u'humidity': 37, # u'pressure': 1026.46, # u'temp_min': 54.74 # }, # u'wind': {u'speed': 6.31, u'deg': 10.09}} days_seen = set() for date in sorted(highs.keys()): day = datetime.fromtimestamp(ts[date]) formatted_date = day.strftime("%a %e %b") if formatted_date in days_seen: continue days_seen.add(formatted_date) total = len(days_seen) first_day = True days_seen = set() for n, date in enumerate(sorted(highs.keys())): if n % 3 == 0: if n > 0: f.write("") f.write('
') remaining = total - n if remaining >= 3: width = "33%" else: width = f"{100/remaining}%" aggregate_daily_precip = 0.0 for r, s in zip(rain[date], snow[date]): hourly_aggregate = r + s # The weather report is always way wrong about Stevens. if self.file_prefix == "stevens": hourly_aggregate *= 3.5 aggregate_daily_precip += hourly_aggregate precip[date].append(hourly_aggregate) logger.debug( f"Aggregate precip on {date} was {aggregate_daily_precip} mm" ) if first_day: while len(precip[date]) < 8: precip[date].insert(0, 0) first_day = False day = datetime.fromtimestamp(ts[date]) formatted_date = day.strftime("%a %e %b") if formatted_date in days_seen: continue days_seen.add(formatted_date) f.write( f'
\n' ) # Date f.write( f""" """ ) # Conditions icon icon = weather_renderer.pick_icon( conditions[date], rain[date], snow[date] ) f.write( f""" """ ) # Low temp -- left color = "#000099" if lows[date] <= 32.5: color = "#009999" f.write( f""" """ ) # Total aggregate_precip in inches (from mm) aggregate_daily_precip /= 25.4 if aggregate_daily_precip > 0.001: f.write( f""" """ ) else: f.write(" \n") # High temp + precip chart color = "#800000" if highs[date] >= 80: color = "#AA0000" f.write( f""" """ ) # Precip graph f.write( f"""
{formatted_date}
{int(lows[date])}°F  
{aggregate_daily_precip:3.1f}”
    {int(highs[date])}°F
""" ) f.write("
") return True # x = weather_renderer({"Stevens": 1000}, "stevens") # x.periodic_render("Stevens")