masq/i3pystatus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimize weather module, rewrite Weather.com backend

Browse Source 
Weather.com's XML feed is now defunct, this commit includes a rewritten
Weather.com module which uses the same JSON feed used by the website
itself.

The weather updates now happen in a separate thread rather than
happening in the ``run()`` function. Since the ``run()`` function is
executed before any of the mouse events are run, this was causing the
mouse event callbacks to result in a flurry of weather updates, which
caused the Weather Underground API to be overutilized beyond its rate
limit.
This commit is contained in:
Erik Johnson 2016-09-15 16:53:11 -05:00
parent 03df1a644a
commit d798a8c3d8
3 changed files with 567 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

197
i3pystatus/weather/__init__.py
View File

@ -1,18 +1,61 @@
import json
import re
import threading
import time
from urllib.request import urlopen
from i3pystatus import SettingsBase, IntervalModule, formatp
from i3pystatus.core.util import user_open, internet, require
class Backend(SettingsBase):
class WeatherBackend(SettingsBase):
settings = ()
@require(internet)
def api_request(self, url):
self.logger.debug('Making API request to %s', url)
try:
with urlopen(url) as content:
try:
content_type = dict(content.getheaders())['Content-Type']
charset = re.search(r'charset=(.*)', content_type).group(1)
except AttributeError:
charset = 'utf-8'
response_json = content.read().decode(charset).strip()
if not response_json:
self.logger.debug('JSON response from %s was blank', url)
return {}
try:
response = json.loads(response_json)
except json.decoder.JSONDecodeError as exc:
self.logger.error('Error loading JSON: %s', exc)
self.logger.debug('JSON text that failed to load: %s',
response_json)
return {}
self.logger.log(5, 'API response: %s', response)
error = self.check_response(response)
if error:
self.logger.error('Error in JSON response: %s', error)
return {}
return response
except Exception as exc:
self.logger.error(
'Failed to make API request to %s. Exception follows:', url,
exc_info=True
)
return {}
def check_response(response):
raise NotImplementedError
class Weather(IntervalModule):
'''
This is a generic weather-checker which must use a configured weather
backend. For list of all available backends see :ref:`weatherbackends`.
Left clicking on the module will launch the forecast page for the location
being checked.
Double-clicking on the module will launch the forecast page for the
location being checked, and single-clicking will trigger an update.
.. _weather-formatters:
@ -45,17 +88,26 @@ class Weather(IntervalModule):
metric or imperial units are being used
* `{humidity}` — Current humidity, excluding percentage symbol
* `{uv_index}` — UV Index
* `{update_error}` — When the configured weather backend encounters an
error during an update, this formatter will be set to the value of the
backend's **update_error** config value. Otherwise, this formatter will
be an empty string.
This module supports the :ref:`formatp <formatp>` extended string format
syntax. This allows for values to be hidden when they evaluate as False.
This comes in handy for the :py:mod:`weathercom <.weather.weathercom>`
backend, which at a certain point in the afternoon will have a blank
``{high_temp}`` value. Using the following snippet in your format string
will only display the high temperature information if it is not blank:
The default **format** string value for this module makes use of this
syntax to conditionally show the value of the **update_error** config value
when the backend encounters an error during an update.
The extended string format syntax also comes in handy for the
:py:mod:`weathercom <.weather.weathercom>` backend, which at a certain
point in the afternoon will have a blank ``{high_temp}`` value. Using the
following snippet in your format string will only display the high
temperature information if it is not blank:
::
{current_temp}{temp_unit}[ Hi: {high_temp}[{temp_unit}]] Lo: {low_temp}{temp_unit}
{current_temp}{temp_unit}[ Hi: {high_temp}] Lo: {low_temp}[ {update_error}]
Brackets are evaluated from the outside-in, so the fact that the only
formatter in the outer block (``{high_temp}``) is empty would keep the
@ -67,6 +119,44 @@ class Weather(IntervalModule):
- :ref:`Weather.com <weather-usage-weathercom>`
- :ref:`Weather Underground <weather-usage-wunderground>`
.. rubric:: Troubleshooting
If an error is encountered while updating, the ``{update_error}`` formatter
will be set, and (provided it is in your ``format`` string) will show up
next to the forecast to alert you to the error. The error message will (by
default be logged to ``~/li3pystatus-<pid>`` where ``<pid>`` is the PID of
the update thread. However, it may be more convenient to manually set the
logfile to make the location of the log data predictable and avoid clutter
in your home directory. Additionally, using the ``DEBUG`` log level can
be helpful in revealing why the module is not working as expected. For
example:
.. code-block:: python
import logging
from i3pystatus import Status
from i3pystatus.weather import weathercom
status = Status(logfile='/home/username/var/i3pystatus.log')
status.register(
'weather',
format='{condition} {current_temp}{temp_unit}[ {icon}][ Hi: {high_temp}][ Lo: {low_temp}][ {update_error}]',
colorize=True,
hints={'markup': 'pango'},
update_error='<span color="#ff0000">!</span>',
log_level=logging.DEBUG,
backend=weathercom.Weathercom(
location_code='94107:4:US',
units='imperial',
log_level=logging.DEBUG,
),
)
.. note::
The log level must be set separately in both the module and backend
contexts.
'''
settings = (
@ -77,7 +167,10 @@ class Weather(IntervalModule):
('color', 'Display color (or fallback color if ``colorize`` is True). '
'If not specified, falls back to default i3bar color.'),
('backend', 'Weather backend instance'),
'interval',
('refresh_icon', 'Text to display (in addition to any text currently '
'shown by the module) when refreshing weather data. '
'**NOTE:** Depending on how quickly the update is '
'performed, the icon may not be displayed.'),
'format',
)
required = ('backend',)
@ -98,16 +191,77 @@ class Weather(IntervalModule):
color = None
backend = None
interval = 1800
format = '{current_temp}{temp_unit}'
refresh_icon = '⟳'
format = '{current_temp}{temp_unit}[ {update_error}]'
on_leftclick = 'open_forecast_url'
output = {'full_text': ''}
def open_forecast_url(self):
on_doubleleftclick = ['launch_web']
on_leftclick = ['check_weather']
def launch_web(self):
if self.backend.forecast_url and self.backend.forecast_url != 'N/A':
self.logger.debug('Launching %s in browser', self.backend.forecast_url)
user_open(self.backend.forecast_url)
def init(self):
pass
if self.backend is None:
raise RuntimeError('A backend is required')
self.backend.data = {
'city': '',
'condition': '',
'observation_time': '',
'current_temp': '',
'low_temp': '',
'high_temp': '',
'temp_unit': '',
'feelslike': '',
'dewpoint': '',
'wind_speed': '',
'wind_unit': '',
'wind_direction': '',
'wind_gust': '',
'pressure': '',
'pressure_unit': '',
'pressure_trend': '',
'visibility': '',
'visibility_unit': '',
'humidity': '',
'uv_index': '',
'update_error': '',
}
self.backend.init()
self.condition = threading.Condition()
self.thread = threading.Thread(target=self.update_thread, daemon=True)
self.thread.start()
def update_thread(self):
try:
self.check_weather()
while True:
with self.condition:
self.condition.wait(self.interval)
self.check_weather()
except Exception:
msg = 'Exception in {thread} at {time}, module {name}'.format(
thread=threading.current_thread().name,
time=time.strftime('%c'),
name=self.__class__.__name__,
)
self.logger.error(msg, exc_info=True)
@require(internet)
def check_weather(self):
'''
Check the weather using the configured backend
'''
self.output['full_text'] = \
self.refresh_icon + self.output.get('full_text', '')
self.backend.check_weather()
self.refresh_display()
def get_color_data(self, condition):
'''
@ -117,11 +271,13 @@ class Weather(IntervalModule):
if condition not in self.color_icons:
# Check for similarly-named conditions if no exact match found
condition_lc = condition.lower()
if 'cloudy' in condition_lc:
if 'cloudy' in condition_lc or 'clouds' in condition_lc:
if 'partly' in condition_lc:
condition = 'Partly Cloudy'
else:
condition = 'Cloudy'
elif condition_lc == 'overcast':
condition = 'Cloudy'
elif 'thunder' in condition_lc or 't-storm' in condition_lc:
condition = 'Thunderstorm'
elif 'snow' in condition_lc:
@ -139,13 +295,16 @@ class Weather(IntervalModule):
if condition not in self.color_icons \
else self.color_icons[condition]
@require(internet)
def run(self):
data = self.backend.weather_data()
data['icon'], condition_color = self.get_color_data(data['condition'])
def refresh_display(self):
self.logger.debug('Weather data: %s', self.backend.data)
self.backend.data['icon'], condition_color = \
self.get_color_data(self.backend.data['condition'])
color = condition_color if self.colorize else self.color
self.output = {
'full_text': formatp(self.format, **data).strip(),
'full_text': formatp(self.format, **self.backend.data).strip(),
'color': color,
}
def run(self):
pass

313
i3pystatus/weather/weathercom.py
View File

@ -1,17 +1,120 @@
from i3pystatus.core.util import internet, require
from i3pystatus.weather import Backend
from i3pystatus.weather import WeatherBackend
from datetime import datetime
from urllib.request import urlopen
from html.parser import HTMLParser
import json
import re
import xml.etree.ElementTree as ElementTree
WEATHER_COM_URL = \
'http://wxdata.weather.com/wxdata/weather/local/%s?unit=%s&dayf=1&cc=*'
ON_LEFTCLICK_URL = 'https://weather.com/weather/today/l/%s'
API_PARAMS = ('api_key', 'lang', 'latitude', 'longitude')
API_URL = 'https://api.weather.com/v2/turbo/vt1precipitation;vt1currentdatetime;vt1pollenforecast;vt1dailyForecast;vt1observation?units=%s&language=%s&geocode=%s,%s&format=json&apiKey=%s'
FORECAST_URL = 'https://weather.com/weather/today/l/%s'
class Weathercom(Backend):
class WeathercomHTMLParser(HTMLParser):
'''
Obtain data points required by the Weather.com API which are obtained
through some other source at runtime and added as <script> elements to the
page source.
'''
def __init__(self, logger, location_code):
self.logger = logger
self.location_code = location_code
for attr in API_PARAMS:
setattr(self, attr, None)
# Not required for API call, but still parsed from the forecast page
self.city_name = ''
super(WeathercomHTMLParser, self).__init__()
def safe_eval(self, data):
'''
Execute an eval with no builtins and no locals
'''
try:
return eval(data, {'__builtins__': None}, {})
except Exception as exc:
self.logger.log(
5,
'Failed to eval() data: %s\n\nOriginal data follows:\n%s',
exc, data
)
return {}
def read_forecast_page(self):
with urlopen(FORECAST_URL % self.location_code) as content:
try:
content_type = dict(content.getheaders())['Content-Type']
charset = re.search(r'charset=(.*)', content_type).group(1)
except AttributeError:
charset = 'utf-8'
html = content.read().decode(charset)
try:
self.feed(html)
except Exception as exc:
self.logger.debug(
'Exception raised while parsing forecast page',
exc
)
def handle_data(self, content):
try:
tag_text = self.get_starttag_text().lower()
except AttributeError:
tag_text = ''
if tag_text == '<script>':
if 'apiKey' in content:
# Key is part of a javascript data structure which looks
# similar to the following:
#
# 'sunTurbo': {
# 'baseUrl': 'https://api.weather.com',
# 'apiKey': 'c1ea9f47f6a88b9acb43aba7faf389d4',
# 'locale': 'en-US' || 'en-us'
# }
#
# For our purposes this is close enough to a Python data
# structure such that it should be able to be eval'ed to a
# Python dict.
sunturbo = content.find('\'sunTurbo\'')
if sunturbo != -1:
# Look for the left curly brace after the 'sunTurbo' key
lbrace = content.find('{', sunturbo)
if lbrace != -1:
# Now look for the right curly brace
rbrace = content.find('}', lbrace)
if rbrace != -1:
api_info = content[lbrace:rbrace + 1]
# Change '||' to 'or' to allow it to be eval'ed
api_info = api_info.replace('||', 'or')
api_data = self.safe_eval(api_info)
for attr, key in (('api_key', 'apiKey'),
('lang', 'locale')):
try:
setattr(self, attr, api_data[key])
except (KeyError, TypeError):
self.logger.debug(
'\'%s\' key not present in %s',
key, api_data
)
if 'explicit_location' in content and self.location_code in content:
lbrace = content.find('{')
rbrace = content.rfind('}')
if lbrace != rbrace != -1:
loc_data = json.loads(content[lbrace:rbrace + 1])
for attr, key in (('latitude', 'lat'),
('longitude', 'long'),
('city_name', 'prsntNm')):
try:
setattr(self, attr, loc_data[key])
except (KeyError, TypeError):
self.logger.debug('\'%s\' key not present in %s',
key, loc_data)
class Weathercom(WeatherBackend):
'''
This module gets the weather from weather.com. The ``location_code``
parameter should be set to the location code from weather.com. To obtain
@ -27,16 +130,18 @@ class Weathercom(Backend):
from i3pystatus import Status
from i3pystatus.weather import weathercom
status = Status()
status = Status(logfile='/home/username/var/i3pystatus.log')
status.register(
'weather',
format='{condition} {current_temp}{temp_unit}{icon}\
[ Hi: {high_temp}] Lo: {low_temp}',
format='{condition} {current_temp}{temp_unit}[ {icon}][ Hi: {high_temp}][ Lo: {low_temp}][ {update_error}]',
interval=900,
colorize=True,
hints={'markup': 'pango'},
backend=weathercom.Weathercom(
location_code='94107:4:US',
units='imperial',
update_error='<span color="#ff0000">!</span>',
),
)
@ -48,91 +153,143 @@ class Weathercom(Backend):
settings = (
('location_code', 'Location code from www.weather.com'),
('units', '\'metric\' or \'imperial\''),
('update_error', 'Value for the ``{update_error}`` formatter when an '
'error is encountered while checking weather data'),
)
required = ('location_code',)
location_code = None
units = 'metric'
update_error = '!'
# This will be set once weather data has been checked
forecast_url = None
@require(internet)
def weather_data(self):
'''
Fetches the current weather from wxdata.weather.com service.
'''
if self.forecast_url is None and ':' in self.location_code:
def init(self):
if self.location_code is None:
raise RuntimeError('A location_code is required')
self.location_code = str(self.location_code)
if ':' in self.location_code:
# Set the URL so that clicking the weather will launch the
# weather.com forecast page. Only set it though if there is a colon
# in the location_code. Technically, the weather.com API will
# successfully return weather data if a U.S. ZIP code is used as
# the location_code (e.g. 94107), but if substituted in
# ON_LEFTCLICK_URL it may or may not result in a valid URL.
self.forecast_url = ON_LEFTCLICK_URL % self.location_code
# FORECAST_URl it may or may not result in a valid URL.
self.forecast_url = FORECAST_URL % self.location_code
unit = '' if self.units == 'imperial' or self.units == '' else 'm'
url = WEATHER_COM_URL % (self.location_code, unit)
with urlopen(url) as handler:
try:
content_type = dict(handler.getheaders())['Content-Type']
charset = re.search(r'charset=(.*)', content_type).group(1)
except AttributeError:
charset = 'utf-8'
xml = handler.read().decode(charset)
doc = ElementTree.XML(xml)
parser = WeathercomHTMLParser(self.logger, self.location_code)
parser.read_forecast_page()
# Cut off the timezone from the end of the string (it's after the last
# space, hence the use of rpartition). International timezones (or ones
# outside the system locale) don't seem to be handled well by
# datetime.datetime.strptime().
try:
observation_time_str = doc.findtext('cc/lsup').rpartition(' ')[0]
observation_time = datetime.strptime(observation_time_str,
'%m/%d/%y %I:%M %p')
except (ValueError, AttributeError):
observation_time = datetime.fromtimestamp(0)
for attr in API_PARAMS:
value = getattr(parser, attr, None)
if value is None:
raise RuntimeError(
'Unable to parse %s from forecast page' % attr)
setattr(self, attr, value)
self.city_name = parser.city_name
pressure_trend_str = doc.findtext('cc/bar/d').lower()
if pressure_trend_str == 'rising':
pressure_trend = '+'
elif pressure_trend_str == 'falling':
pressure_trend = '-'
else:
pressure_trend = ''
if not doc.findtext('dayf/day[@d="0"]/part[@p="d"]/icon').strip():
# If the "d" (day) part of today's forecast's keys are empty, there
# is no high temp anymore (this happens in the afternoon), but
# instead of handling things in a sane way and setting the high
# temp to an empty string or something like that, the API returns
# the current temp as the high temp, which is incorrect. This if
# statement catches it so that we can correctly report that there
# is no high temp at this point of the day.
high_temp = ''
else:
high_temp = doc.findtext('dayf/day[@d="0"]/hi')
return dict(
city=doc.findtext('loc/dnam'),
condition=doc.findtext('cc/t'),
observation_time=observation_time,
current_temp=doc.findtext('cc/tmp'),
low_temp=doc.findtext('dayf/day[@d="0"]/low'),
high_temp=high_temp,
temp_unit='°' + doc.findtext('head/ut').upper(),
feelslike=doc.findtext('cc/flik'),
dewpoint=doc.findtext('cc/dewp'),
wind_speed=doc.findtext('cc/wind/s'),
wind_unit=doc.findtext('head/us'),
wind_direction=doc.findtext('cc/wind/t'),
wind_gust=doc.findtext('cc/wind/gust'),
pressure=doc.findtext('cc/bar/r'),
pressure_unit=doc.findtext('head/up'),
pressure_trend=pressure_trend,
visibility=doc.findtext('cc/vis'),
visibility_unit=doc.findtext('head/ud'),
humidity=doc.findtext('cc/hmid'),
uv_index=doc.findtext('cc/uv/i'),
units = 'e' if self.units == 'imperial' or self.units == '' else 'm'
self.url = API_URL % (
'e' if self.units in ('imperial', '') else 'm',
self.lang, self.latitude, self.longitude, self.api_key
)
def check_response(self, response):
# Errors for weather.com API manifest in HTTP error codes, not in the
# JSON response.
return False
@require(internet)
def check_weather(self):
'''
Fetches the current weather from wxdata.weather.com service.
'''
self.data['update_error'] = ''
try:
response = self.api_request(self.url)
if not response:
self.data['update_error'] = self.update_error
return
observed = response.get('vt1observation', {})
forecast = response.get('vt1dailyForecast', {})
# Cut off the timezone from the end of the string (it's after the last
# space, hence the use of rpartition). International timezones (or ones
# outside the system locale) don't seem to be handled well by
# datetime.datetime.strptime().
try:
observation_time_str = str(observed.get('observationTime', ''))
observation_time = datetime.strptime(observation_time_str,
'%Y-%d-%yT%H:%M:%S%z')
except (ValueError, AttributeError):
observation_time = datetime.fromtimestamp(0)
try:
pressure_trend_str = observed.get('barometerTrend', '').lower()
except AttributeError:
pressure_trend_str = ''
if pressure_trend_str == 'rising':
pressure_trend = '+'
elif pressure_trend_str == 'falling':
pressure_trend = '-'
else:
pressure_trend = ''
try:
high_temp = forecast.get('day', {}).get('temperature', [])[0]
except (AttributeError, IndexError):
high_temp = ''
else:
if high_temp is None:
# In the mid-afternoon, the high temp disappears from the
# forecast, so just set high_temp to an empty string.
high_temp = ''
try:
low_temp = forecast.get('night', {}).get('temperature', [])[0]
except (AttributeError, IndexError):
low_temp = ''
if self.units == 'imperial':
temp_unit = '°F'
wind_unit = 'mph'
pressure_unit = 'in'
visibility_unit = 'mi'
else:
temp_unit = '°C'
wind_unit = 'kph'
pressure_unit = 'mb'
visibility_unit = 'km'
self.data['city'] = self.city_name
self.data['condition'] = str(observed.get('phrase', ''))
self.data['observation_time'] = observation_time
self.data['current_temp'] = str(observed.get('temperature', ''))
self.data['low_temp'] = str(low_temp)
self.data['high_temp'] = str(high_temp)
self.data['temp_unit'] = temp_unit
self.data['feelslike'] = str(observed.get('feelsLike', ''))
self.data['dewpoint'] = str(observed.get('dewPoint', ''))
self.data['wind_speed'] = str(observed.get('windSpeed', ''))
self.data['wind_unit'] = wind_unit
self.data['wind_direction'] = str(observed.get('windDirCompass', ''))
# Gust can be None, using "or" to ensure empty string in this case
self.data['wind_gust'] = str(observed.get('gust', '') or '')
self.data['pressure'] = str(observed.get('altimeter', ''))
self.data['pressure_unit'] = pressure_unit
self.data['pressure_trend'] = pressure_trend
self.data['visibility'] = str(observed.get('visibility', ''))
self.data['visibility_unit'] = visibility_unit
self.data['humidity'] = str(observed.get('humidity', ''))
self.data['uv_index'] = str(observed.get('uvIndex', ''))
except Exception:
# Don't let an uncaught exception kill the update thread
self.logger.error(
'Uncaught error occurred while checking weather. '
'Exception follows:', exc_info=True
)
self.data['update_error'] = self.update_error

274
i3pystatus/weather/wunderground.py
View File

@ -1,16 +1,14 @@
from i3pystatus import IntervalModule
from i3pystatus.core.util import internet, require
from i3pystatus.weather import WeatherBackend
from datetime import datetime
from urllib.request import urlopen
import json
import re
GEOLOOKUP_URL = 'http://api.wunderground.com/api/%s/geolookup%s/q/%s.json'
STATION_QUERY_URL = 'http://api.wunderground.com/api/%s/%s/q/%s.json'
class Wunderground(IntervalModule):
class Wunderground(WeatherBackend):
'''
This module retrieves weather data using the Weather Underground API.
@ -19,10 +17,6 @@ class Wunderground(IntervalModule):
sign up for a developer API key free at
https://www.wunderground.com/weather/api/
A developer API key is allowed 500 queries per day, and no more than 10
in a given minute. Therefore, it is recommended to be conservative when
setting the update interval.
Valid values for ``location_code`` include:
* **State/City_Name** - CA/San_Francisco
@ -34,11 +28,29 @@ class Wunderground(IntervalModule):
* **Personal Weather Station (PWS)** - pws:KILCHICA30
When not using a ``pws`` or ``icao`` station ID, the location will be
queried, and the closest station will be used. For a list of PWS
station IDs, visit the following URL:
queried (this uses an API query), and the closest station will be used.
For a list of PWS station IDs, visit the following URL:
http://www.wunderground.com/weatherstation/ListStations.asp
.. rubric:: API usage
An API key is allowed 500 queries per day, and no more than 10 in a
given minute. Therefore, it is recommended to be conservative when
setting the update interval (the default is 1800 seconds, or 30
minutes), and one should be careful how often one restarts i3pystatus
and how often a refresh is forced by left-clicking the module.
As noted above, when not using a ``pws`` or ``icao`` station ID, an API
query will be used to determine the station ID to use. This will be
done once when i3pystatus is started, and not repeated until the next
time i3pystatus is started.
When updating weather data, one API query will be used to obtain the
current conditions. The high/low temperature forecast requires an
additonal API query, and is optional (disabled by default). To enable
forecast checking, set ``forecast=True``.
.. _weather-usage-wunderground:
.. rubric:: Usage example
@ -48,16 +60,19 @@ class Wunderground(IntervalModule):
from i3pystatus import Status
from i3pystatus.weather import wunderground
status = Status()
status = Status(logfile='/home/username/var/i3pystatus.log')
status.register(
'weather',
format='{condition} {current_temp}{temp_unit}{icon}[ Hi: {high_temp}] Lo: {low_temp}',
format='{condition} {current_temp}{temp_unit}[ {icon}][ Hi: {high_temp}][ Lo: {low_temp}][ {update_error}]',
colorize=True,
hints={'markup': 'pango'},
backend=wunderground.Wunderground(
api_key='dbafe887d56ba4ad',
location_code='pws:MAT645',
units='imperial',
forecast=True,
update_error='<span color="#ff0000">!</span>',
),
)
@ -66,9 +81,6 @@ class Wunderground(IntervalModule):
See :ref:`here <weather-formatters>` for a list of formatters which can be
used.
'''
interval = 300
settings = (
('api_key', 'Weather Underground API key'),
('location_code', 'Location code from wunderground.com'),
@ -78,6 +90,8 @@ class Wunderground(IntervalModule):
'additional API request per weather update). If set to '
'``False``, then the ``low_temp`` and ``high_temp`` '
'formatters will be set to empty strings.'),
('update_error', 'Value for the ``{update_error}`` formatter when an '
'error is encountered while checking weather data'),
)
required = ('api_key', 'location_code')
@ -87,71 +101,53 @@ class Wunderground(IntervalModule):
units = 'metric'
use_pws = True
forecast = False
update_error = '!'
# These will be set once weather data has been checked
station_id = None
forecast_url = None
@require(internet)
def api_request(self, url):
'''
Execute an HTTP POST to the specified URL and return the content
'''
with urlopen(url) as content:
try:
content_type = dict(content.getheaders())['Content-Type']
charset = re.search(r'charset=(.*)', content_type).group(1)
except AttributeError:
charset = 'utf-8'
response = json.loads(content.read().decode(charset))
try:
raise Exception(response['response']['error']['description'])
except KeyError:
pass
return response
@require(internet)
def geolookup(self):
def init(self):
'''
Use the location_code to perform a geolookup and find the closest
station. If the location is a pws or icao station ID, no lookup will be
peformed.
'''
if self.station_id is None:
try:
for no_lookup in ('pws', 'icao'):
sid = self.location_code.partition(no_lookup + ':')[-1]
if sid:
self.station_id = self.location_code
return
except AttributeError:
# Numeric or some other type, either way we'll just stringify
# it below and perform a lookup.
pass
try:
for no_lookup in ('pws', 'icao'):
sid = self.location_code.partition(no_lookup + ':')[-1]
if sid:
self.station_id = self.location_code
return
except AttributeError:
# Numeric or some other type, either way we'll just stringify
# it below and perform a lookup.
pass
extra_opts = '/pws:0' if not self.use_pws else ''
api_url = GEOLOOKUP_URL % (self.api_key,
extra_opts,
self.location_code)
response = self.api_request(api_url)
station_type = 'pws' if self.use_pws else 'airport'
try:
stations = response['location']['nearby_weather_stations']
nearest = stations[station_type]['station'][0]
except (KeyError, IndexError):
raise Exception('No locations matched location_code %s'
% self.location_code)
extra_opts = '/pws:0' if not self.use_pws else ''
api_url = GEOLOOKUP_URL % (self.api_key,
extra_opts,
self.location_code)
response = self.api_request(api_url)
station_type = 'pws' if self.use_pws else 'airport'
try:
stations = response['location']['nearby_weather_stations']
nearest = stations[station_type]['station'][0]
except (KeyError, IndexError):
raise Exception(
'No locations matched location_code %s' % self.location_code)
if self.use_pws:
nearest_pws = nearest.get('id', '')
if not nearest_pws:
raise Exception('No id entry for station')
self.station_id = 'pws:%s' % nearest_pws
else:
nearest_airport = nearest.get('icao', '')
if not nearest_airport:
raise Exception('No icao entry for station')
self.station_id = 'icao:%s' % nearest_airport
if self.use_pws:
nearest_pws = nearest.get('id', '')
if not nearest_pws:
raise Exception('No id entry for nearest PWS')
self.station_id = 'pws:%s' % nearest_pws
else:
nearest_airport = nearest.get('icao', '')
if not nearest_airport:
raise Exception('No icao entry for nearest airport')
self.station_id = 'icao:%s' % nearest_airport
@require(internet)
def get_forecast(self):
@ -160,6 +156,7 @@ class Wunderground(IntervalModule):
data for the configured/queried weather station, and return the low and
high temperatures. Otherwise, return two empty strings.
'''
no_data = ('', '')
if self.forecast:
query_url = STATION_QUERY_URL % (self.api_key,
'forecast',
@ -168,73 +165,110 @@ class Wunderground(IntervalModule):
response = self.api_request(query_url)['forecast']
response = response['simpleforecast']['forecastday'][0]
except (KeyError, IndexError, TypeError):
raise Exception('No forecast data found for %s' % self.station_id)
self.logger.error(
'No forecast data found for %s', self.station_id)
self.data['update_error'] = self.update_error
return no_data
unit = 'celsius' if self.units == 'metric' else 'fahrenheit'
low_temp = response.get('low', {}).get(unit, '')
high_temp = response.get('high', {}).get(unit, '')
return low_temp, high_temp
else:
return '', ''
return no_data
def check_response(self, response):
try:
return response['response']['error']['description']
except KeyError:
# No error in response
return False
@require(internet)
def weather_data(self):
def check_weather(self):
'''
Query the configured/queried station and return the weather data
'''
# If necessary, do a geolookup to set the station_id
self.geolookup()
query_url = STATION_QUERY_URL % (self.api_key,
'conditions',
self.station_id)
self.data['update_error'] = ''
try:
response = self.api_request(query_url)['current_observation']
self.forecast_url = response.pop('ob_url', None)
except KeyError:
raise Exception('No weather data found for %s' % self.station_id)
query_url = STATION_QUERY_URL % (self.api_key,
'conditions',
self.station_id)
try:
response = self.api_request(query_url)['current_observation']
self.forecast_url = response.pop('ob_url', None)
except KeyError:
self.logger.error('No weather data found for %s', self.station_id)
self.data['update_error'] = self.update_error
return
low_temp, high_temp = self.get_forecast()
if self.forecast:
query_url = STATION_QUERY_URL % (self.api_key,
'forecast',
self.station_id)
try:
forecast = self.api_request(query_url)['forecast']
forecast = forecast['simpleforecast']['forecastday'][0]
except (KeyError, IndexError, TypeError):
self.logger.error(
'No forecast data found for %s', self.station_id)
# This is a non-fatal error, so don't return but do set the
# error flag.
self.data['update_error'] = self.update_error
if self.units == 'metric':
temp_unit = 'c'
speed_unit = 'kph'
distance_unit = 'km'
pressure_unit = 'mb'
else:
temp_unit = 'f'
speed_unit = 'mph'
distance_unit = 'mi'
pressure_unit = 'in'
unit = 'celsius' if self.units == 'metric' else 'fahrenheit'
low_temp = forecast.get('low', {}).get(unit, '')
high_temp = forecast.get('high', {}).get(unit, '')
else:
low_temp = high_temp = ''
def _find(key, data=None):
data = data or response
return data.get(key, 'N/A')
if self.units == 'metric':
temp_unit = 'c'
speed_unit = 'kph'
distance_unit = 'km'
pressure_unit = 'mb'
else:
temp_unit = 'f'
speed_unit = 'mph'
distance_unit = 'mi'
pressure_unit = 'in'
try:
observation_time = int(_find('observation_epoch'))
except TypeError:
observation_time = 0
def _find(key, data=None, default=''):
if data is None:
data = response
return str(data.get(key, default))
return dict(
city=_find('city', response['observation_location']),
condition=_find('weather'),
observation_time=datetime.fromtimestamp(observation_time),
current_temp=_find('temp_' + temp_unit),
low_temp=low_temp,
high_temp=high_temp,
temp_unit='°' + temp_unit.upper(),
feelslike=_find('feelslike_' + temp_unit),
dewpoint=_find('dewpoint_' + temp_unit),
wind_speed=_find('wind_' + speed_unit),
wind_unit=speed_unit,
wind_direction=_find('wind_dir'),
wind_gust=_find('wind_gust_' + speed_unit),
pressure=_find('pressure_' + pressure_unit),
pressure_unit=pressure_unit,
pressure_trend=_find('pressure_trend'),
visibility=_find('visibility_' + distance_unit),
visibility_unit=distance_unit,
humidity=_find('relative_humidity').rstrip('%'),
uv_index=_find('uv'),
)
try:
observation_time = datetime.fromtimestamp(
int(_find('observation_epoch'))
)
except TypeError:
observation_time = datetime.fromtimestamp(0)
self.data['city'] = _find('city', response['observation_location'])
self.data['condition'] = _find('weather')
self.data['observation_time'] = observation_time
self.data['current_temp'] = _find('temp_' + temp_unit)
self.data['low_temp'] = low_temp
self.data['high_temp'] = high_temp
self.data['temp_unit'] = '°' + temp_unit.upper()
self.data['feelslike'] = _find('feelslike_' + temp_unit)
self.data['dewpoint'] = _find('dewpoint_' + temp_unit)
self.data['wind_speed'] = _find('wind_' + speed_unit)
self.data['wind_unit'] = speed_unit
self.data['wind_direction'] = _find('wind_dir')
self.data['wind_gust'] = _find('wind_gust_' + speed_unit)
self.data['pressure'] = _find('pressure_' + pressure_unit)
self.data['pressure_unit'] = pressure_unit
self.data['pressure_trend'] = _find('pressure_trend')
self.data['visibility'] = _find('visibility_' + distance_unit)
self.data['visibility_unit'] = distance_unit
self.data['humidity'] = _find('relative_humidity').rstrip('%')
self.data['uv_index'] = _find('UV')
except Exception:
# Don't let an uncaught exception kill the update thread
self.logger.error(
'Uncaught error occurred while checking weather. '
'Exception follows:', exc_info=True
)
self.data['update_error'] = self.update_error