Move DwdWeatherWarningsAPI to a library hosted on PyPI (#34820)
* Move DwdWeatherWarningsAPI to a library hosted on PyPI PyPI library uses new DWD WFS API WFS API allows a more detailed query with reduced data sent as return Change CONF_REGION_NAME from Optional to Required because it was never really optional Set attribute region_state to "N/A" because it is not available via the new API Add attributes warning_i_parameters and warning_i_color * Use constants instead of raw strings Streamline methods state and device_state_attributes * Wrap api, use UTC time * Update homeassistant/components/dwd_weather_warnings/sensor.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update homeassistant/components/dwd_weather_warnings/manifest.json Co-authored-by: Martin Hjelmare <marhje52@gmail.com> Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
stephan192
GitHub
parent
5802d65ef7
commit
016cd8f8ef
4 changed files with 81 additions and 146 deletions
|
|
@ -104,6 +104,7 @@ homeassistant/components/discogs/* @thibmaek
|
|||
homeassistant/components/doorbird/* @oblogic7 @bdraco
|
||||
homeassistant/components/dsmr_reader/* @depl0y
|
||||
homeassistant/components/dunehd/* @bieniu
|
||||
homeassistant/components/dwd_weather_warnings/* @runningman84 @stephan192 @Hummel95
|
||||
homeassistant/components/dweet/* @fabaff
|
||||
homeassistant/components/dynalite/* @ziv1234
|
||||
homeassistant/components/dyson/* @etheralm
|
||||
|
|
|
|||
|
|
@ -2,6 +2,6 @@
|
|||
"domain": "dwd_weather_warnings",
|
||||
"name": "Deutsche Wetter Dienst (DWD) Weather Warnings",
|
||||
"documentation": "https://www.home-assistant.io/integrations/dwd_weather_warnings",
|
||||
"after_dependencies": ["rest"],
|
||||
"codeowners": []
|
||||
"codeowners": ["@runningman84", "@stephan192", "@Hummel95"],
|
||||
"requirements": ["dwdwfsapi==1.0.2"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,37 +10,52 @@ Warnungen vor markantem Wetter (Stufe 2)
|
|||
Wetterwarnungen (Stufe 1)
|
||||
"""
|
||||
from datetime import timedelta
|
||||
import json
|
||||
import logging
|
||||
|
||||
from dwdwfsapi import DwdWeatherWarningsAPI
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.rest.sensor import RestData
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.const import ATTR_ATTRIBUTION, CONF_MONITORED_CONDITIONS, CONF_NAME
|
||||
from homeassistant.helpers.aiohttp_client import SERVER_SOFTWARE as HA_USER_AGENT
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.util import Throttle
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTRIBUTION = "Data provided by DWD"
|
||||
ATTR_REGION_NAME = "region_name"
|
||||
ATTR_REGION_ID = "region_id"
|
||||
ATTR_LAST_UPDATE = "last_update"
|
||||
ATTR_WARNING_COUNT = "warning_count"
|
||||
|
||||
API_ATTR_WARNING_NAME = "event"
|
||||
API_ATTR_WARNING_TYPE = "event_code"
|
||||
API_ATTR_WARNING_LEVEL = "level"
|
||||
API_ATTR_WARNING_HEADLINE = "headline"
|
||||
API_ATTR_WARNING_DESCRIPTION = "description"
|
||||
API_ATTR_WARNING_INSTRUCTION = "instruction"
|
||||
API_ATTR_WARNING_START = "start_time"
|
||||
API_ATTR_WARNING_END = "end_time"
|
||||
API_ATTR_WARNING_PARAMETERS = "parameters"
|
||||
API_ATTR_WARNING_COLOR = "color"
|
||||
|
||||
DEFAULT_NAME = "DWD-Weather-Warnings"
|
||||
|
||||
CONF_REGION_NAME = "region_name"
|
||||
|
||||
CURRENT_WARNING_SENSOR = "current_warning_level"
|
||||
ADVANCE_WARNING_SENSOR = "advance_warning_level"
|
||||
|
||||
SCAN_INTERVAL = timedelta(minutes=15)
|
||||
|
||||
MONITORED_CONDITIONS = {
|
||||
"current_warning_level": [
|
||||
CURRENT_WARNING_SENSOR: [
|
||||
"Current Warning Level",
|
||||
None,
|
||||
"mdi:close-octagon-outline",
|
||||
],
|
||||
"advance_warning_level": [
|
||||
ADVANCE_WARNING_SENSOR: [
|
||||
"Advance Warning Level",
|
||||
None,
|
||||
"mdi:close-octagon-outline",
|
||||
|
|
@ -49,7 +64,7 @@ MONITORED_CONDITIONS = {
|
|||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_REGION_NAME): cv.string,
|
||||
vol.Required(CONF_REGION_NAME): cv.string,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(
|
||||
CONF_MONITORED_CONDITIONS, default=list(MONITORED_CONDITIONS)
|
||||
|
|
@ -63,12 +78,11 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
name = config.get(CONF_NAME)
|
||||
region_name = config.get(CONF_REGION_NAME)
|
||||
|
||||
api = DwdWeatherWarningsAPI(region_name)
|
||||
api = WrappedDwDWWAPI(DwdWeatherWarningsAPI(region_name))
|
||||
|
||||
sensors = [
|
||||
DwdWeatherWarningsSensor(api, name, condition)
|
||||
for condition in config[CONF_MONITORED_CONDITIONS]
|
||||
]
|
||||
sensors = []
|
||||
for sensor_type in config[CONF_MONITORED_CONDITIONS]:
|
||||
sensors.append(DwdWeatherWarningsSensor(api, name, sensor_type))
|
||||
|
||||
add_entities(sensors, True)
|
||||
|
||||
|
|
@ -76,179 +90,96 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
class DwdWeatherWarningsSensor(Entity):
|
||||
"""Representation of a DWD-Weather-Warnings sensor."""
|
||||
|
||||
def __init__(self, api, name, variable):
|
||||
def __init__(self, api, name, sensor_type):
|
||||
"""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]
|
||||
self._sensor_type = sensor_type
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return f"{self._name} {self._var_name}"
|
||||
return f"{self._name} {MONITORED_CONDITIONS[self._sensor_type][0]}"
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Icon to use in the frontend, if any."""
|
||||
return self._var_icon
|
||||
return MONITORED_CONDITIONS[self._sensor_type][2]
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._var_units
|
||||
return MONITORED_CONDITIONS[self._sensor_type][1]
|
||||
|
||||
@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]
|
||||
if self._sensor_type == CURRENT_WARNING_SENSOR:
|
||||
return self._api.api.current_warning_level
|
||||
return self._api.api.expected_warning_level
|
||||
|
||||
@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}
|
||||
data = {
|
||||
ATTR_ATTRIBUTION: ATTRIBUTION,
|
||||
ATTR_REGION_NAME: self._api.api.warncell_name,
|
||||
ATTR_REGION_ID: self._api.api.warncell_id,
|
||||
ATTR_LAST_UPDATE: self._api.api.last_update,
|
||||
}
|
||||
|
||||
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"
|
||||
if self._sensor_type == CURRENT_WARNING_SENSOR:
|
||||
searched_warnings = self._api.api.current_warnings
|
||||
else:
|
||||
raise Exception("Unknown warning type")
|
||||
searched_warnings = self._api.api.expected_warnings
|
||||
|
||||
data["warning_count"] = self._api.data[f"{prefix}_warning_count"]
|
||||
i = 0
|
||||
for event in self._api.data[f"{prefix}_warnings"]:
|
||||
i = i + 1
|
||||
data[ATTR_WARNING_COUNT] = len(searched_warnings)
|
||||
|
||||
# dictionary for the attribute containing the complete warning as json
|
||||
event_json = event.copy()
|
||||
for i, warning in enumerate(searched_warnings, 1):
|
||||
data[f"warning_{i}_name"] = warning[API_ATTR_WARNING_NAME]
|
||||
data[f"warning_{i}_type"] = warning[API_ATTR_WARNING_TYPE]
|
||||
data[f"warning_{i}_level"] = warning[API_ATTR_WARNING_LEVEL]
|
||||
data[f"warning_{i}_headline"] = warning[API_ATTR_WARNING_HEADLINE]
|
||||
data[f"warning_{i}_description"] = warning[API_ATTR_WARNING_DESCRIPTION]
|
||||
data[f"warning_{i}_instruction"] = warning[API_ATTR_WARNING_INSTRUCTION]
|
||||
data[f"warning_{i}_start"] = warning[API_ATTR_WARNING_START]
|
||||
data[f"warning_{i}_end"] = warning[API_ATTR_WARNING_END]
|
||||
data[f"warning_{i}_parameters"] = warning[API_ATTR_WARNING_PARAMETERS]
|
||||
data[f"warning_{i}_color"] = warning[API_ATTR_WARNING_COLOR]
|
||||
|
||||
data[f"warning_{i}_name"] = event["event"]
|
||||
data[f"warning_{i}_level"] = event["level"]
|
||||
data[f"warning_{i}_type"] = event["type"]
|
||||
if event["headline"]:
|
||||
data[f"warning_{i}_headline"] = event["headline"]
|
||||
if event["description"]:
|
||||
data[f"warning_{i}_description"] = event["description"]
|
||||
if event["instruction"]:
|
||||
data[f"warning_{i}_instruction"] = event["instruction"]
|
||||
|
||||
if event["start"] is not None:
|
||||
data[f"warning_{i}_start"] = dt_util.as_local(
|
||||
dt_util.utc_from_timestamp(event["start"] / 1000)
|
||||
)
|
||||
event_json["start"] = data[f"warning_{i}_start"]
|
||||
|
||||
if event["end"] is not None:
|
||||
data[f"warning_{i}_end"] = dt_util.as_local(
|
||||
dt_util.utc_from_timestamp(event["end"] / 1000)
|
||||
)
|
||||
event_json["end"] = data[f"warning_{i}_end"]
|
||||
|
||||
data[f"warning_{i}"] = event_json
|
||||
# Dictionary for the attribute containing the complete warning
|
||||
warning_copy = warning.copy()
|
||||
warning_copy[API_ATTR_WARNING_START] = data[f"warning_{i}_start"]
|
||||
warning_copy[API_ATTR_WARNING_END] = data[f"warning_{i}_end"]
|
||||
data[f"warning_{i}"] = warning_copy
|
||||
|
||||
return data
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Could the device be accessed during the last update call."""
|
||||
return self._api.available
|
||||
return self._api.api.data_valid
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data from the DWD-Weather-Warnings API."""
|
||||
_LOGGER.debug(
|
||||
"Update requested for %s (%s) by %s",
|
||||
self._api.api.warncell_name,
|
||||
self._api.api.warncell_id,
|
||||
self._sensor_type,
|
||||
)
|
||||
self._api.update()
|
||||
|
||||
|
||||
class DwdWeatherWarningsAPI:
|
||||
"""Get the latest data and update the states."""
|
||||
class WrappedDwDWWAPI:
|
||||
"""Wrapper for the DWD-Weather-Warnings api."""
|
||||
|
||||
def __init__(self, region_name):
|
||||
"""Initialize the data object."""
|
||||
resource = "https://www.dwd.de/DWD/warnungen/warnapp_landkreise/json/warnings.json?jsonp=loadWarnings"
|
||||
|
||||
# a User-Agent is necessary for this rest api endpoint (#29496)
|
||||
headers = {"User-Agent": HA_USER_AGENT}
|
||||
|
||||
self._rest = RestData("GET", resource, None, headers, None, True)
|
||||
self.region_name = region_name
|
||||
self.region_id = None
|
||||
self.region_state = None
|
||||
self.data = None
|
||||
self.available = True
|
||||
self.update()
|
||||
def __init__(self, api):
|
||||
"""Initialize a DWD-Weather-Warnings wrapper."""
|
||||
self.api = api
|
||||
|
||||
@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[f"{mykey}_warning_level"] = 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[f"{mykey}_warning_level"]
|
||||
for event in my_warnings:
|
||||
if event["level"] >= maxlevel:
|
||||
data[f"{mykey}_warning_level"] = event["level"]
|
||||
|
||||
data[f"{mykey}_warning_count"] = len(my_warnings)
|
||||
data[f"{mykey}_warnings"] = 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
|
||||
"""Get the latest data from the DWD-Weather-Warnings API."""
|
||||
self.api.update()
|
||||
_LOGGER.debug("Update performed")
|
||||
|
|
|
|||
|
|
@ -505,6 +505,9 @@ dovado==0.4.1
|
|||
# homeassistant.components.dsmr
|
||||
dsmr_parser==0.18
|
||||
|
||||
# homeassistant.components.dwd_weather_warnings
|
||||
dwdwfsapi==1.0.2
|
||||
|
||||
# homeassistant.components.dweet
|
||||
dweepy==0.3.0
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue