From 9ede0f57e601b341890fd0a92237a47f41e85b7f Mon Sep 17 00:00:00 2001 From: runningman84 Date: Tue, 5 Sep 2017 17:40:47 +0200 Subject: [PATCH] Added DWD WarnApp Sensor (#8657) * Added DWD WarnApp Sensor * Fixed some idents and spaces * Removed unused imports * Removed comment * Some fixes * Added throttle * Renamed sensor to dwd weather warnings * Renamed test file * shorten lines * shorten lines * Implemented changes requested by fabaff * added ATTRIBUTION * move ATTRIBUTION to existing method * fixed lint tests * Fix linter issues * Fix linter issues * Fix linter * Fixed linter --- .coveragerc | 1 + .../components/sensor/dwd_weather_warnings.py | 243 ++++++++++++++++++ 2 files changed, 244 insertions(+) create mode 100644 homeassistant/components/sensor/dwd_weather_warnings.py diff --git a/.coveragerc b/.coveragerc index 5e27aed0182..ecf35b8030d 100644 --- a/.coveragerc +++ b/.coveragerc @@ -451,6 +451,7 @@ omit = homeassistant/components/sensor/dovado.py homeassistant/components/sensor/dte_energy_bridge.py homeassistant/components/sensor/dublin_bus_transport.py + homeassistant/components/sensor/dwd_weather_warnings.py homeassistant/components/sensor/ebox.py homeassistant/components/sensor/eddystone_temperature.py homeassistant/components/sensor/eliqonline.py diff --git a/homeassistant/components/sensor/dwd_weather_warnings.py b/homeassistant/components/sensor/dwd_weather_warnings.py new file mode 100644 index 00000000000..0eeaa9424e8 --- /dev/null +++ b/homeassistant/components/sensor/dwd_weather_warnings.py @@ -0,0 +1,243 @@ +""" +Support for getting statistical data from a DWD Weather Warnings. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/sensor.dwd_weather_warnings/ + +Data is fetched from DWD: +https://rcccm.dwd.de/DE/wetter/warnungen_aktuell/objekt_einbindung/objekteinbindung.html + +Warnungen vor extremem Unwetter (Stufe 4) +Unwetterwarnungen (Stufe 3) +Warnungen vor markantem Wetter (Stufe 2) +Wetterwarnungen (Stufe 1) +""" +import logging +import json +from datetime import timedelta + +import voluptuous as vol + +import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.entity import Entity +from homeassistant.components.sensor import PLATFORM_SCHEMA +from homeassistant.const import ( + ATTR_ATTRIBUTION, CONF_NAME, CONF_MONITORED_CONDITIONS) +from homeassistant.util import Throttle +import homeassistant.util.dt as dt_util +from homeassistant.components.sensor.rest import RestData + +_LOGGER = logging.getLogger(__name__) + +ATTRIBUTION = "Data provided by DWD" + +DEFAULT_NAME = 'DWD-Weather-Warnings' + +CONF_REGION_NAME = 'region_name' + +SCAN_INTERVAL = timedelta(minutes=15) + +MONITORED_CONDITIONS = { + 'current_warning_level': ['Current Warning Level', + None, 'mdi:close-octagon-outline'], + 'advance_warning_level': ['Advance Warning Level', + None, 'mdi:close-octagon-outline'], +} + +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + vol.Optional(CONF_REGION_NAME): cv.string, + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Optional(CONF_MONITORED_CONDITIONS, default=MONITORED_CONDITIONS): + vol.All(cv.ensure_list, [vol.In(MONITORED_CONDITIONS)]), +}) + + +def setup_platform(hass, config, add_devices, discovery_info=None): + """Set up the DWD-Weather-Warnings sensor.""" + name = config.get(CONF_NAME) + region_name = config.get(CONF_REGION_NAME) + + api = DwdWeatherWarningsAPI(region_name) + + sensors = [DwdWeatherWarningsSensor(api, name, condition) + for condition in config[CONF_MONITORED_CONDITIONS]] + + add_devices(sensors, True) + + +class DwdWeatherWarningsSensor(Entity): + """Representation of a DWD-Weather-Warnings sensor.""" + + def __init__(self, api, name, variable): + """Initialize a DWD-Weather-Warnings sensor.""" + self._api = api + self._name = name + self._var_id = variable + + variable_info = MONITORED_CONDITIONS[variable] + self._var_name = variable_info[0] + self._var_units = variable_info[1] + self._var_icon = variable_info[2] + + @property + def name(self): + """Return the name of the sensor.""" + return "{} {}".format(self._name, self._var_name) + + @property + def icon(self): + """Icon to use in the frontend, if any.""" + return self._var_icon + + @property + def unit_of_measurement(self): + """Return the unit the value is expressed in.""" + return self._var_units + + # pylint: disable=no-member + @property + def state(self): + """Return the state of the device.""" + try: + return round(self._api.data[self._var_id], 2) + except TypeError: + return self._api.data[self._var_id] + + # pylint: disable=no-member + @property + def device_state_attributes(self): + """Return the state attributes of the DWD-Weather-Warnings.""" + data = { + ATTR_ATTRIBUTION: ATTRIBUTION, + 'region_name': self._api.region_name + } + + if self._api.region_id is not None: + data['region_id'] = self._api.region_id + + if self._api.region_state is not None: + data['region_state'] = self._api.region_state + + if self._api.data['time'] is not None: + data['last_update'] = dt_util.as_local( + dt_util.utc_from_timestamp(self._api.data['time'] / 1000)) + + if self._var_id == 'current_warning_level': + prefix = 'current' + elif self._var_id == 'advance_warning_level': + prefix = 'advance' + else: + raise Exception('Unknown warning type') + + data['warning_count'] = self._api.data[prefix + '_warning_count'] + i = 0 + for event in self._api.data[prefix + '_warnings']: + i = i + 1 + + data['warning_{}_name'.format(i)] = event['event'] + data['warning_{}_level'.format(i)] = event['level'] + data['warning_{}_type'.format(i)] = event['type'] + if len(event['headline']) > 0: + data['warning_{}_headline'.format(i)] = event['headline'] + if len(event['description']) > 0: + data['warning_{}_description'.format(i)] = event['description'] + if len(event['instruction']) > 0: + data['warning_{}_instruction'.format(i)] = event['instruction'] + + if event['start'] is not None: + data['warning_{}_start'.format(i)] = dt_util.as_local( + dt_util.utc_from_timestamp(event['start'] / 1000)) + + if event['end'] is not None: + data['warning_{}_end'.format(i)] = dt_util.as_local( + dt_util.utc_from_timestamp(event['end'] / 1000)) + + return data + + @property + def available(self): + """Could the device be accessed during the last update call.""" + return self._api.available + + def update(self): + """Get the latest data from the DWD-Weather-Warnings API.""" + self._api.update() + + +class DwdWeatherWarningsAPI(object): + """Get the latest data and update the states.""" + + def __init__(self, region_name): + """Initialize the data object.""" + resource = "{}{}{}?{}".format( + 'https://', + 'www.dwd.de', + '/DWD/warnungen/warnapp_landkreise/json/warnings.json', + 'jsonp=loadWarnings' + ) + + self._rest = RestData('GET', resource, None, None, None, True) + self.region_name = region_name + self.region_id = None + self.region_state = None + self.data = None + self.available = True + self.update() + + @Throttle(SCAN_INTERVAL) + def update(self): + """Get the latest data from the DWD-Weather-Warnings.""" + try: + self._rest.update() + + json_string = self._rest.data[24:len(self._rest.data) - 2] + json_obj = json.loads(json_string) + + data = {'time': json_obj['time']} + + for mykey, myvalue in { + 'current': 'warnings', + 'advance': 'vorabInformation' + }.items(): + + _LOGGER.debug("Found %d %s global DWD warnings", + len(json_obj[myvalue]), mykey) + + data['{}_warning_level'.format(mykey)] = 0 + my_warnings = [] + + if self.region_id is not None: + # get a specific region_id + if self.region_id in json_obj[myvalue]: + my_warnings = json_obj[myvalue][self.region_id] + + else: + # loop through all items to find warnings, region_id + # and region_state for region_name + for key in json_obj[myvalue]: + my_region = json_obj[myvalue][key][0]['regionName'] + if my_region != self.region_name: + continue + my_warnings = json_obj[myvalue][key] + my_state = json_obj[myvalue][key][0]['stateShort'] + self.region_id = key + self.region_state = my_state + break + + # Get max warning level + maxlevel = data['{}_warning_level'.format(mykey)] + for event in my_warnings: + if event['level'] >= maxlevel: + data['{}_warning_level'.format(mykey)] = event['level'] + + data['{}_warning_count'.format(mykey)] = len(my_warnings) + data['{}_warnings'.format(mykey)] = my_warnings + + _LOGGER.debug("Found %d %s local DWD warnings", + len(my_warnings), mykey) + + self.data = data + self.available = True + except TypeError: + _LOGGER.error("Unable to fetch data from DWD-Weather-Warnings") + self.available = False