Last active
October 28, 2016 05:35
-
-
Save tchellomello/309b0a643470646312ee257cd4ea1aff to your computer and use it in GitHub Desktop.
Added more attributes to Home Assistant Wunderground sensor - https://community.home-assistant.io/t/weather-underground-wunderground-weather-sensor/1849/28
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Support for WUnderground weather service. | |
For more details about this platform, please refer to the documentation at | |
https://home-assistant.io/components/sensor.wunderground/ | |
""" | |
from datetime import timedelta | |
import logging | |
import requests | |
import voluptuous as vol | |
from homeassistant.components.sensor import PLATFORM_SCHEMA | |
from homeassistant.const import ( | |
CONF_MONITORED_CONDITIONS, CONF_API_KEY, TEMP_FAHRENHEIT, TEMP_CELSIUS, | |
STATE_UNKNOWN, ATTR_ATTRIBUTION) | |
from homeassistant.helpers.entity import Entity | |
from homeassistant.util import Throttle | |
import homeassistant.helpers.config_validation as cv | |
_RESOURCE = 'http://api.wunderground.com/api/{}/conditions/q/' | |
_ALERTS = 'http://api.wunderground.com/api/{}/alerts/q/' | |
_LOGGER = logging.getLogger(__name__) | |
CONF_ATTRIBUTION = "Data provided by the WUnderground weather service" | |
CONF_PWS_ID = 'pws_id' | |
MIN_TIME_BETWEEN_UPDATES_ALERTS = timedelta(minutes=15) | |
MIN_TIME_BETWEEN_UPDATES_OBSERVATION = timedelta(minutes=5) | |
# Sensor types are defined like: Name, units | |
SENSOR_TYPES = { | |
'alerts': ['Alerts', None], | |
'dewpoint_c': ['Dewpoint (°C)', TEMP_CELSIUS], | |
'dewpoint_f': ['Dewpoint (°F)', TEMP_FAHRENHEIT], | |
'dewpoint_string': ['Dewpoint Summary', None], | |
'feelslike_c': ['Feels Like (°C)', TEMP_CELSIUS], | |
'feelslike_f': ['Feels Like (°F)', TEMP_FAHRENHEIT], | |
'feelslike_string': ['Feels Like', None], | |
'heat_index_c': ['Dewpoint (°C)', TEMP_CELSIUS], | |
'heat_index_f': ['Dewpoint (°F)', TEMP_FAHRENHEIT], | |
'heat_index_string': ['Heat Index Summary', None], | |
'observation_time': ['Observation Time', None], | |
'pressure_in': ['Pressure', 'in'], | |
'pressure_mb': ['Pressure', 'mbar'], | |
'precip_1hr_in': ['Precipation 1hr', 'in'], | |
'precip_1hr_metric': ['Precipation 1hr', 'mm'], | |
'precip_1hr_string': ['Precipation 1hr', None], | |
'precip_today_in': ['Precipation Today', 'in'], | |
'precip_today_metric': ['Precipitation Today', 'mm'], | |
'precip_today_string': ['Precipitation today', None], | |
'pressure_in': ['Pressure', 'in'], | |
'pressure_trend': ['Pressure Trend', None], | |
'relative_humidity': ['Relative Humidity', '%'], | |
'station_id': ['Station ID', None], | |
'solarradiation': ['Solar Radiation', None], | |
'temp_c': ['Temperature (°C)', TEMP_CELSIUS], | |
'temp_f': ['Temperature (°F)', TEMP_FAHRENHEIT], | |
'UV': ['UV', None], | |
'visibility_km': ['Visibility (km)', 'km'], | |
'visibility_mi': ['Visibility (miles)', 'mi'], | |
'weather': ['Weather Summary', None], | |
'wind_degrees': ['Wind Degrees', None], | |
'wind_dir': ['Wind Direction', None], | |
'wind_gust_kph': ['Wind Gust', 'kpH'], | |
'wind_gust_mph': ['Wind Gust', 'mpH'], | |
'wind_kph': ['Wind Speed', 'kpH'], | |
'wind_mph': ['Wind Speed', 'mpH'], | |
'wind_string': ['Wind Summary', None], | |
} | |
# Alert Attributes | |
ALERTS_ATTRS = [ | |
'date', | |
'description', | |
'expires', | |
'message', | |
] | |
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ | |
vol.Required(CONF_API_KEY): cv.string, | |
vol.Optional(CONF_PWS_ID): cv.string, | |
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): | |
vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]), | |
}) | |
def setup_platform(hass, config, add_devices, discovery_info=None): | |
"""Setup the WUnderground sensor.""" | |
rest = WUndergroundData(hass, | |
config.get(CONF_API_KEY), | |
config.get(CONF_PWS_ID, None)) | |
sensors = [] | |
for variable in config[CONF_MONITORED_CONDITIONS]: | |
sensors.append(WUndergroundSensor(rest, variable)) | |
try: | |
rest.update() | |
except ValueError as err: | |
_LOGGER.error("Received error from WUnderground: %s", err) | |
return False | |
add_devices(sensors) | |
return True | |
class WUndergroundSensor(Entity): | |
"""Implementing the WUnderground sensor.""" | |
def __init__(self, rest, condition): | |
"""Initialize the sensor.""" | |
self.rest = rest | |
self._condition = condition | |
@property | |
def name(self): | |
"""Return the name of the sensor.""" | |
return "PWS_" + self._condition | |
@property | |
def state(self): | |
"""Return the state of the sensor.""" | |
if self.rest.data and self._condition in self.rest.data: | |
if self._condition == 'relative_humidity': | |
return int(self.rest.data[self._condition][:-1]) | |
else: | |
return self.rest.data[self._condition] | |
if self._condition == 'alerts': | |
if self.rest.alerts: | |
return len(self.rest.alerts) | |
else: | |
return 0 | |
return STATE_UNKNOWN | |
@property | |
def device_state_attributes(self): | |
"""Return the state attributes.""" | |
attrs = {} | |
attrs[ATTR_ATTRIBUTION] = CONF_ATTRIBUTION | |
if not self.rest.alerts or self._condition != 'alerts': | |
return attrs | |
multiple_alerts = len(self.rest.alerts) > 1 | |
for data in self.rest.alerts: | |
for alert in ALERTS_ATTRS: | |
if data[alert]: | |
if multiple_alerts: | |
dkey = alert.capitalize() + '_' + data['type'] | |
else: | |
dkey = alert.capitalize() | |
attrs[dkey] = data[alert] | |
return attrs | |
@property | |
def entity_picture(self): | |
"""Return the entity picture.""" | |
if self._condition == 'weather': | |
return self.rest.data['icon_url'] | |
@property | |
def unit_of_measurement(self): | |
"""Return the units of measurement.""" | |
return SENSOR_TYPES[self._condition][1] | |
def update(self): | |
"""Update current conditions.""" | |
if self._condition == 'alerts': | |
self.rest.update_alerts() | |
else: | |
self.rest.update() | |
# pylint: disable=too-few-public-methods | |
class WUndergroundData(object): | |
"""Get data from WUnderground.""" | |
def __init__(self, hass, api_key, pws_id=None): | |
"""Initialize the data object.""" | |
self._hass = hass | |
self._api_key = api_key | |
self._pws_id = pws_id | |
self._latitude = hass.config.latitude | |
self._longitude = hass.config.longitude | |
self.data = None | |
self.alerts = None | |
def _build_url(self, baseurl=_RESOURCE): | |
url = baseurl.format(self._api_key) | |
if self._pws_id: | |
url = url + 'pws:{}'.format(self._pws_id) | |
else: | |
url = url + '{},{}'.format(self._latitude, self._longitude) | |
return url + '.json' | |
@Throttle(MIN_TIME_BETWEEN_UPDATES_OBSERVATION) | |
def update(self): | |
"""Get the latest data from WUnderground.""" | |
try: | |
result = requests.get(self._build_url(), timeout=10).json() | |
if "error" in result['response']: | |
raise ValueError(result['response']["error"] | |
["description"]) | |
else: | |
self.data = result["current_observation"] | |
except ValueError as err: | |
_LOGGER.error("Check WUnderground API %s", err.args) | |
self.data = None | |
raise | |
@Throttle(MIN_TIME_BETWEEN_UPDATES_ALERTS) | |
def update_alerts(self): | |
"""Get the latest alerts data from WUnderground.""" | |
try: | |
result = requests.get(self._build_url(_ALERTS), timeout=10).json() | |
if "error" in result['response']: | |
raise ValueError(result['response']["error"] | |
["description"]) | |
else: | |
self.alerts = result["alerts"] | |
except ValueError as err: | |
_LOGGER.error("Check WUnderground API %s", err.args) | |
self.alerts = None | |
raise |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment