216 lines
7.4 KiB
Python
216 lines
7.4 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import itertools
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from zoneinfo import ZoneInfo
|
|
|
|
class WeatherAPI:
|
|
|
|
def __init__(self, apikey=None, gps=None, opts={"lang": "fr", "units": "metric"}):
|
|
self.apikey = apikey or os.environ["WEATHERAPIKEY"]
|
|
self.baseurl = "http://api.weatherapi.com/v1"
|
|
self.default_gps = gps or ("GPS" in os.environ and os.environ["GPS"]) or "48.8127,2.3437"
|
|
self.opts = opts
|
|
|
|
self._cached_file = ".weather-%s-%s.cache"
|
|
|
|
|
|
def get_weather(self, apitype="forecast", gps=None, **params):
|
|
if gps is None:
|
|
gps = self.default_gps
|
|
|
|
cached_filename = self._cached_file % (apitype + ("" if "dt" not in params else "-{}".format(params["dt"])), gps)
|
|
# Read the mod time
|
|
statinfo = None
|
|
try:
|
|
statinfo = os.stat(cached_filename)
|
|
except:
|
|
pass
|
|
|
|
params["q"] = gps
|
|
params["key"] = str(self.apikey)
|
|
params["alerts"] = "yes"
|
|
params["aqi"] = "yes"
|
|
for k in self.opts:
|
|
params[k] = self.opts[k]
|
|
|
|
if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(hours=1) < datetime.now(tz=timezone.utc):
|
|
# Do the request and save it
|
|
try:
|
|
with urllib.request.urlopen(self.baseurl + "/" + apitype + ".json?" + urllib.parse.urlencode(params)) as f:
|
|
with open(cached_filename, 'wb') as fd:
|
|
fd.write(f.read())
|
|
except ConnectionResetError:
|
|
pass
|
|
except urllib.error.URLError:
|
|
pass
|
|
|
|
|
|
# Retrieve cached data
|
|
res = {}
|
|
with open(cached_filename) as f:
|
|
res = json.load(f)
|
|
|
|
return res
|
|
|
|
|
|
def get_icon(icon, night=False, current=False):
|
|
if icon == 1000:
|
|
if not night:
|
|
return "wi-day-sunny.png"
|
|
else:
|
|
return "wi-night-clear.png"
|
|
elif icon == 1003:
|
|
if not night:
|
|
return "wi-day-cloudy.png"
|
|
else:
|
|
return "wi-night-alt-cloudy.png"
|
|
elif icon == 1006:
|
|
return "wi-cloudy.png"
|
|
elif icon == 1009:
|
|
if not night:
|
|
return "wi-day-sunny-overcast.png"
|
|
else:
|
|
return "wi-night-alt-partly-cloudy.png"
|
|
elif icon == 1030 or icon == 1072:
|
|
if not night:
|
|
return "wi-day-haze.png"
|
|
else:
|
|
return "wi-night-fog.png"
|
|
elif icon == 1063:
|
|
if not night:
|
|
return "wi-day-rain.png"
|
|
else:
|
|
return "wi-night-alt-rain.png"
|
|
elif icon == 1066:
|
|
if not night:
|
|
return "wi-day-snow.png"
|
|
else:
|
|
return "wi-night-alt-snow.png"
|
|
elif icon == 1069 or icon == 1204:
|
|
if not night:
|
|
return "wi-day-sleet.png"
|
|
else:
|
|
return "wi-night-alt-sleet.png"
|
|
elif icon == 1087:
|
|
if not night:
|
|
return "wi-day-lightning.png"
|
|
else:
|
|
return "wi-night-alt-lightning.png"
|
|
elif icon == 1114 or icon == 1219 or icon == 1222 or icon == 1258:
|
|
return "wi-snow.png"
|
|
elif icon == 1117 or icon == 1225:
|
|
return "wi-snow-wind.png"
|
|
elif icon == 1135 or icon == 1147:
|
|
return "wi-fog.png"
|
|
elif icon == 1150 or icon == 1153:
|
|
return "wi-sprinkle.png"
|
|
elif icon == 1168 or icon == 1171 or icon == 1198 or icon == 1201:
|
|
return "wi-snowflake-cold.png"
|
|
elif icon == 1180 or icon == 1183 or icon == 1186:
|
|
if not night:
|
|
return "wi-day-rain.png"
|
|
else:
|
|
return "wi-night-alt-rain.png"
|
|
elif icon == 1189 or icon == 1192 or icon == 1195:
|
|
return "wi-rain.png"
|
|
elif icon == 1240 or icon == 1243:
|
|
return "wi-showers.png"
|
|
elif icon == 1246:
|
|
return "wi-tsunami.png"
|
|
elif icon == 1207 or icon == 1249 or icon == 1252:
|
|
return "wi-sleet.png"
|
|
elif icon == 1210 or icon == 1213 or icon == 1216 or icon == 1255:
|
|
if not night:
|
|
return "wi-day-snow.png"
|
|
else:
|
|
return "wi-night-snow.png"
|
|
elif icon == 1237:
|
|
return "wi-hail.png"
|
|
elif icon == 1261 or icon == 1264:
|
|
if not night:
|
|
return "wi-day-hail.png"
|
|
else:
|
|
return "wi-night-hail.png"
|
|
elif icon == 1273 or icon == 1276:
|
|
if not night:
|
|
return "wi-day-thunderstorm.png"
|
|
else:
|
|
return "wi-night-thunderstorm.png"
|
|
elif icon == 1279 or icon == 1282:
|
|
if not night:
|
|
return "wi-day-snow-thunderstorm.png"
|
|
else:
|
|
return "wi-night-snow-thunderstorm.png"
|
|
else:
|
|
return "wi-alien.png"
|
|
|
|
|
|
def get_moon_icon(self, day=0):
|
|
moon_phase = list(itertools.islice(self.get_daily(), 1))[day]["moon_phase"]
|
|
|
|
if moon_phase == "New Moon":
|
|
return "wi-moon-alt-new.png"
|
|
elif moon_phase == "Waxing Crescent":
|
|
return "wi-moon-alt-waxing-crescent-4.png"
|
|
elif moon_phase == "First Quarter":
|
|
return "wi-moon-alt-first-quarter.png"
|
|
elif moon_phase == "Waxing Gibbous":
|
|
return "wi-moon-alt-waxing-gibbous-4.png"
|
|
elif moon_phase == "Full Moon":
|
|
return "wi-moon-alt-full.png"
|
|
elif moon_phase == "Waning Gibbous":
|
|
return "wi-moon-alt-waning-gibbous-4.png"
|
|
elif moon_phase == "Last Quarter":
|
|
return "wi-moon-alt-third-quarter.png"
|
|
else:
|
|
return "wi-moon-alt-waning-crescent-3.png"
|
|
|
|
|
|
def get_currently(self, *args, **kwargs):
|
|
return self.get_weather("current", *args, **kwargs)["current"]
|
|
|
|
|
|
def get_forecast(self, *args, **kwargs):
|
|
for i in [0, 1, 2, 3]:
|
|
enddt = datetime.now() + timedelta(days=i)
|
|
v = self.get_weather(*args, **kwargs, dt=enddt.strftime("%Y-%m-%d"))["forecast"]["forecastday"][0]
|
|
v["day"]["date"] = enddt
|
|
yield v
|
|
|
|
def get_hourly(self, *args, **kwargs):
|
|
now = datetime.now().astimezone(ZoneInfo("Europe/Paris"))
|
|
for d in self.get_forecast(*args, **kwargs):
|
|
for h in d["hour"]:
|
|
if d["day"]["date"].day == now.day and now > self.read_timestamp(h["time_epoch"]) + timedelta(hours=1):
|
|
continue
|
|
yield h
|
|
|
|
def get_daily(self, *args, **kwargs):
|
|
for d in self.get_forecast(*args, **kwargs):
|
|
yield {
|
|
**d["day"],
|
|
**d["astro"]
|
|
}
|
|
|
|
def get_alerts(self, *args, **kwargs):
|
|
for i in [0, 1, 2, 3]:
|
|
enddt = datetime.now() + timedelta(days=i)
|
|
for a in self.get_weather(*args, **kwargs, dt=enddt.strftime("%Y-%m-%d"))["alerts"]["alert"]:
|
|
yield a
|
|
|
|
def read_timestamp(self, timestamp, *args, **kwargs):
|
|
PARIS = ZoneInfo("Europe/Paris")
|
|
return datetime.fromtimestamp(int(timestamp), tz=timezone.utc).astimezone(PARIS)
|
|
|
|
|
|
#WeatherAPI = TomorrowAPI
|
|
|
|
if __name__ == '__main__':
|
|
dsa = WeatherAPI()
|
|
print(json.dumps(dsa.get_currently()))
|
|
#print(json.dumps([d for d in dsa.get_daily()][0]))
|