#!/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"""
{formatted_date}
|
"""
)
# 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"""
{int(lows[date])}°F
|
"""
)
# Total aggregate_precip in inches (from mm)
aggregate_daily_precip /= 25.4
if aggregate_daily_precip > 0.001:
f.write(
f"""
{aggregate_daily_precip:3.1f}”
|
"""
)
else:
f.write(" | \n")
# High temp + precip chart
color = "#800000"
if highs[date] >= 80:
color = "#AA0000"
f.write(
f"""
{int(highs[date])}°F
|
"""
)
# Precip graph
f.write(
f"""
|
"""
)
f.write("
")
return True
# x = weather_renderer({"Stevens": 1000}, "stevens")
# x.periodic_render("Stevens")