jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use new AEMET library data for sensor platform (#102972)

Browse Source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Álvaro Fernández Rojas 2024-01-10 16:36:20 +01:00 committed by GitHub
parent de9bb20135
commit 6a6c447c28
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 218 additions and 666 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
homeassistant/components/aemet/const.py
View File

@ -59,8 +59,6 @@ ENTRY_WEATHER_COORDINATOR = "weather_coordinator"
ATTR_API_CONDITION = "condition"
ATTR_API_FORECAST_CONDITION = "condition"
ATTR_API_FORECAST_DAILY = "forecast-daily"
ATTR_API_FORECAST_HOURLY = "forecast-hourly"
ATTR_API_FORECAST_PRECIPITATION = "precipitation"
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY = "precipitation_probability"
ATTR_API_FORECAST_TEMP = "temperature"
@ -101,49 +99,6 @@ CONDITIONS_MAP = {
AOD_COND_SUNNY: ATTR_CONDITION_SUNNY,
}
FORECAST_MONITORED_CONDITIONS = [
ATTR_API_FORECAST_CONDITION,
ATTR_API_FORECAST_PRECIPITATION,
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_API_FORECAST_TEMP,
ATTR_API_FORECAST_TEMP_LOW,
ATTR_API_FORECAST_TIME,
ATTR_API_FORECAST_WIND_BEARING,
ATTR_API_FORECAST_WIND_SPEED,
]
MONITORED_CONDITIONS = [
ATTR_API_CONDITION,
ATTR_API_HUMIDITY,
ATTR_API_PRESSURE,
ATTR_API_RAIN,
ATTR_API_RAIN_PROB,
ATTR_API_SNOW,
ATTR_API_SNOW_PROB,
ATTR_API_STATION_ID,
ATTR_API_STATION_NAME,
ATTR_API_STATION_TIMESTAMP,
ATTR_API_STORM_PROB,
ATTR_API_TEMPERATURE,
ATTR_API_TEMPERATURE_FEELING,
ATTR_API_TOWN_ID,
ATTR_API_TOWN_NAME,
ATTR_API_TOWN_TIMESTAMP,
ATTR_API_WIND_BEARING,
ATTR_API_WIND_MAX_SPEED,
ATTR_API_WIND_SPEED,
]
FORECAST_MODE_DAILY = "daily"
FORECAST_MODE_HOURLY = "hourly"
FORECAST_MODES = [
FORECAST_MODE_DAILY,
FORECAST_MODE_HOURLY,
]
FORECAST_MODE_ATTR_API = {
FORECAST_MODE_DAILY: ATTR_API_FORECAST_DAILY,
FORECAST_MODE_HOURLY: ATTR_API_FORECAST_HOURLY,
}
FORECAST_MAP = {
AOD_FORECAST_DAILY: {
AOD_CONDITION: ATTR_FORECAST_CONDITION,

346
homeassistant/components/aemet/sensor.py
View File

@ -1,6 +1,41 @@
"""Support for the AEMET OpenData service."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import Final
from aemet_opendata.const import (
AOD_CONDITION,
AOD_FEEL_TEMP,
AOD_FORECAST_CURRENT,
AOD_FORECAST_DAILY,
AOD_FORECAST_HOURLY,
AOD_HUMIDITY,
AOD_ID,
AOD_NAME,
AOD_PRECIPITATION,
AOD_PRECIPITATION_PROBABILITY,
AOD_PRESSURE,
AOD_RAIN,
AOD_RAIN_PROBABILITY,
AOD_SNOW,
AOD_SNOW_PROBABILITY,
AOD_STATION,
AOD_STORM_PROBABILITY,
AOD_TEMP,
AOD_TEMP_MAX,
AOD_TEMP_MIN,
AOD_TIMESTAMP,
AOD_TOWN,
AOD_WEATHER,
AOD_WIND_DIRECTION,
AOD_WIND_SPEED,
AOD_WIND_SPEED_MAX,
)
from aemet_opendata.helpers import dict_nested_value
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -18,7 +53,6 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util
from .const import (
@ -51,172 +85,270 @@ from .const import (
ATTR_API_WIND_MAX_SPEED,
ATTR_API_WIND_SPEED,
ATTRIBUTION,
CONDITIONS_MAP,
DOMAIN,
ENTRY_NAME,
ENTRY_WEATHER_COORDINATOR,
FORECAST_MODE_ATTR_API,
FORECAST_MODE_DAILY,
FORECAST_MODES,
FORECAST_MONITORED_CONDITIONS,
MONITORED_CONDITIONS,
)
from .entity import AemetEntity
from .weather_update_coordinator import WeatherUpdateCoordinator
FORECAST_SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key=ATTR_API_FORECAST_CONDITION,
name="Condition",
@dataclass(frozen=True, kw_only=True)
class AemetSensorEntityDescription(SensorEntityDescription):
"""A class that describes AEMET OpenData sensor entities."""
keys: list[str] | None = None
value_fn: Callable[[str], datetime | float | int | str | None] = lambda value: value
FORECAST_SENSORS: Final[tuple[AemetSensorEntityDescription, ...]] = (
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_CONDITION}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_CONDITION],
name="Daily forecast condition",
value_fn=CONDITIONS_MAP.get,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_PRECIPITATION,
name="Precipitation",
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_CONDITION}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_CONDITION],
name="Hourly forecast condition",
value_fn=CONDITIONS_MAP.get,
),
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_PRECIPITATION}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_PRECIPITATION],
name="Hourly forecast precipitation",
native_unit_of_measurement=UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR,
device_class=SensorDeviceClass.PRECIPITATION_INTENSITY,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_PRECIPITATION_PROBABILITY,
name="Precipitation probability",
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_PRECIPITATION_PROBABILITY}",
keys=[
AOD_TOWN,
AOD_FORECAST_DAILY,
AOD_FORECAST_CURRENT,
AOD_PRECIPITATION_PROBABILITY,
],
name="Daily forecast precipitation probability",
native_unit_of_measurement=PERCENTAGE,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_TEMP,
name="Temperature",
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_PRECIPITATION_PROBABILITY}",
keys=[
AOD_TOWN,
AOD_FORECAST_HOURLY,
AOD_FORECAST_CURRENT,
AOD_PRECIPITATION_PROBABILITY,
],
name="Hourly forecast precipitation probability",
native_unit_of_measurement=PERCENTAGE,
),
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_TEMP}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_TEMP_MAX],
name="Daily forecast temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_TEMP_LOW,
name="Temperature Low",
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_TEMP_LOW}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_TEMP_MIN],
name="Daily forecast temperature low",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_TIME,
name="Time",
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_TEMP}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_TEMP],
name="Hourly forecast temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
),
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_TIME}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_TIMESTAMP],
name="Daily forecast time",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=dt_util.parse_datetime,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_WIND_BEARING,
name="Wind bearing",
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_TIME}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_TIMESTAMP],
name="Hourly forecast time",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=dt_util.parse_datetime,
),
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_WIND_BEARING}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_WIND_DIRECTION],
name="Daily forecast wind bearing",
native_unit_of_measurement=DEGREE,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_WIND_MAX_SPEED,
name="Wind max speed",
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_WIND_BEARING}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_WIND_DIRECTION],
name="Hourly forecast wind bearing",
native_unit_of_measurement=DEGREE,
),
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_WIND_MAX_SPEED}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_WIND_SPEED_MAX],
name="Hourly forecast wind max speed",
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
device_class=SensorDeviceClass.WIND_SPEED,
),
SensorEntityDescription(
key=ATTR_API_FORECAST_WIND_SPEED,
name="Wind speed",
AemetSensorEntityDescription(
key=f"forecast-daily-{ATTR_API_FORECAST_WIND_SPEED}",
keys=[AOD_TOWN, AOD_FORECAST_DAILY, AOD_FORECAST_CURRENT, AOD_WIND_SPEED],
name="Daily forecast wind speed",
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
device_class=SensorDeviceClass.WIND_SPEED,
),
AemetSensorEntityDescription(
entity_registry_enabled_default=False,
key=f"forecast-hourly-{ATTR_API_FORECAST_WIND_SPEED}",
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_FORECAST_CURRENT, AOD_WIND_SPEED],
name="Hourly forecast wind speed",
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
device_class=SensorDeviceClass.WIND_SPEED,
),
)
WEATHER_SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
WEATHER_SENSORS: Final[tuple[AemetSensorEntityDescription, ...]] = (
AemetSensorEntityDescription(
key=ATTR_API_CONDITION,
keys=[AOD_WEATHER, AOD_CONDITION],
name="Condition",
value_fn=CONDITIONS_MAP.get,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_HUMIDITY,
keys=[AOD_WEATHER, AOD_HUMIDITY],
name="Humidity",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.HUMIDITY,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_PRESSURE,
keys=[AOD_WEATHER, AOD_PRESSURE],
name="Pressure",
native_unit_of_measurement=UnitOfPressure.HPA,
device_class=SensorDeviceClass.PRESSURE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_RAIN,
keys=[AOD_WEATHER, AOD_RAIN],
name="Rain",
native_unit_of_measurement=UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR,
device_class=SensorDeviceClass.PRECIPITATION_INTENSITY,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_RAIN_PROB,
keys=[AOD_WEATHER, AOD_RAIN_PROBABILITY],
name="Rain probability",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_SNOW,
keys=[AOD_WEATHER, AOD_SNOW],
name="Snow",
native_unit_of_measurement=UnitOfVolumetricFlux.MILLIMETERS_PER_HOUR,
device_class=SensorDeviceClass.PRECIPITATION_INTENSITY,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_SNOW_PROB,
keys=[AOD_WEATHER, AOD_SNOW_PROBABILITY],
name="Snow probability",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_STATION_ID,
keys=[AOD_STATION, AOD_ID],
name="Station ID",
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_STATION_NAME,
keys=[AOD_STATION, AOD_NAME],
name="Station name",
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_STATION_TIMESTAMP,
keys=[AOD_STATION, AOD_TIMESTAMP],
name="Station timestamp",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=dt_util.parse_datetime,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_STORM_PROB,
keys=[AOD_WEATHER, AOD_STORM_PROBABILITY],
name="Storm probability",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_TEMPERATURE,
keys=[AOD_WEATHER, AOD_TEMP],
name="Temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_TEMPERATURE_FEELING,
keys=[AOD_WEATHER, AOD_FEEL_TEMP],
name="Temperature feeling",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_TOWN_ID,
keys=[AOD_TOWN, AOD_ID],
name="Town ID",
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_TOWN_NAME,
keys=[AOD_TOWN, AOD_NAME],
name="Town name",
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_TOWN_TIMESTAMP,
keys=[AOD_TOWN, AOD_FORECAST_HOURLY, AOD_TIMESTAMP],
name="Town timestamp",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=dt_util.parse_datetime,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_WIND_BEARING,
keys=[AOD_WEATHER, AOD_WIND_DIRECTION],
name="Wind bearing",
native_unit_of_measurement=DEGREE,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_WIND_MAX_SPEED,
keys=[AOD_WEATHER, AOD_WIND_SPEED_MAX],
name="Wind max speed",
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
device_class=SensorDeviceClass.WIND_SPEED,
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
AemetSensorEntityDescription(
key=ATTR_API_WIND_SPEED,
keys=[AOD_WEATHER, AOD_WIND_SPEED],
name="Wind speed",
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
device_class=SensorDeviceClass.WIND_SPEED,
@ -232,108 +364,46 @@ async def async_setup_entry(
) -> None:
"""Set up AEMET OpenData sensor entities based on a config entry."""
domain_data = hass.data[DOMAIN][config_entry.entry_id]
name = domain_data[ENTRY_NAME]
weather_coordinator = domain_data[ENTRY_WEATHER_COORDINATOR]
name: str = domain_data[ENTRY_NAME]
coordinator: WeatherUpdateCoordinator = domain_data[ENTRY_WEATHER_COORDINATOR]
unique_id = config_entry.unique_id
entities: list[AbstractAemetSensor] = [
AemetSensor(name, unique_id, weather_coordinator, description)
for description in WEATHER_SENSOR_TYPES
if description.key in MONITORED_CONDITIONS
]
entities.extend(
[
AemetForecastSensor(
f"{domain_data[ENTRY_NAME]} {mode} Forecast",
f"{unique_id}-forecast-{mode}",
weather_coordinator,
mode,
description,
entities: list[AemetSensor] = []
for description in FORECAST_SENSORS + WEATHER_SENSORS:
if dict_nested_value(coordinator.data["lib"], description.keys) is not None:
entities.append(
AemetSensor(
name,
coordinator,
description,
config_entry,
)
)
for mode in FORECAST_MODES
for description in FORECAST_SENSOR_TYPES
if description.key in FORECAST_MONITORED_CONDITIONS
]
)
async_add_entities(entities)
class AbstractAemetSensor(CoordinatorEntity[WeatherUpdateCoordinator], SensorEntity):
"""Abstract class for an AEMET OpenData sensor."""
class AemetSensor(AemetEntity, SensorEntity):
"""Implementation of an AEMET OpenData sensor."""
_attr_attribution = ATTRIBUTION
entity_description: AemetSensorEntityDescription
def __init__(
self,
name,
unique_id,
name: str,
coordinator: WeatherUpdateCoordinator,
description: SensorEntityDescription,
description: AemetSensorEntityDescription,
config_entry: ConfigEntry,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_name = f"{name} {description.name}"
self._attr_unique_id = unique_id
class AemetSensor(AbstractAemetSensor):
"""Implementation of an AEMET OpenData sensor."""
def __init__(
self,
name,
unique_id_prefix,
weather_coordinator: WeatherUpdateCoordinator,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(
name=name,
unique_id=f"{unique_id_prefix}-{description.key}",
coordinator=weather_coordinator,
description=description,
)
self._attr_unique_id = f"{config_entry.unique_id}-{description.key}"
@property
def native_value(self):
"""Return the state of the device."""
return self.coordinator.data.get(self.entity_description.key)
class AemetForecastSensor(AbstractAemetSensor):
"""Implementation of an AEMET OpenData forecast sensor."""
def __init__(
self,
name,
unique_id_prefix,
weather_coordinator: WeatherUpdateCoordinator,
forecast_mode,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(
name=name,
unique_id=f"{unique_id_prefix}-{description.key}",
coordinator=weather_coordinator,
description=description,
)
self._forecast_mode = forecast_mode
self._attr_entity_registry_enabled_default = (
self._forecast_mode == FORECAST_MODE_DAILY
)
@property
def native_value(self):
"""Return the state of the device."""
forecast = None
forecasts = self.coordinator.data.get(
FORECAST_MODE_ATTR_API[self._forecast_mode]
)
if forecasts:
forecast = forecasts[0].get(self.entity_description.key)
if self.entity_description.key == ATTR_API_FORECAST_TIME:
forecast = dt_util.parse_datetime(forecast)
return forecast
value = self.get_aemet_value(self.entity_description.keys)
return self.entity_description.value_fn(value)

485
homeassistant/components/aemet/weather_update_coordinator.py
View File

@ -7,117 +7,28 @@ import logging
from typing import Any, Final, cast
from aemet_opendata.const import (
AEMET_ATTR_DATE,
AEMET_ATTR_DAY,
AEMET_ATTR_DIRECTION,
AEMET_ATTR_ELABORATED,
AEMET_ATTR_FEEL_TEMPERATURE,
AEMET_ATTR_FORECAST,
AEMET_ATTR_HUMIDITY,
AEMET_ATTR_MAX,
AEMET_ATTR_MIN,
AEMET_ATTR_PRECIPITATION,
AEMET_ATTR_PRECIPITATION_PROBABILITY,
AEMET_ATTR_SKY_STATE,
AEMET_ATTR_SNOW,
AEMET_ATTR_SNOW_PROBABILITY,
AEMET_ATTR_SPEED,
AEMET_ATTR_STATION_DATE,
AEMET_ATTR_STATION_HUMIDITY,
AEMET_ATTR_STATION_PRESSURE,
AEMET_ATTR_STATION_PRESSURE_SEA,
AEMET_ATTR_STATION_TEMPERATURE,
AEMET_ATTR_STORM_PROBABILITY,
AEMET_ATTR_TEMPERATURE,
AEMET_ATTR_WIND,
AEMET_ATTR_WIND_GUST,
AOD_CONDITION,
AOD_FORECAST,
AOD_FORECAST_DAILY,
AOD_FORECAST_HOURLY,
AOD_TOWN,
ATTR_DATA,
)
from aemet_opendata.exceptions import AemetError
from aemet_opendata.forecast import ForecastValue
from aemet_opendata.helpers import (
dict_nested_value,
get_forecast_day_value,
get_forecast_hour_value,
get_forecast_interval_value,
)
from aemet_opendata.helpers import dict_nested_value
from aemet_opendata.interface import AEMET
from homeassistant.components.weather import Forecast
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import (
ATTR_API_CONDITION,
ATTR_API_FORECAST_CONDITION,
ATTR_API_FORECAST_DAILY,
ATTR_API_FORECAST_HOURLY,
ATTR_API_FORECAST_PRECIPITATION,
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_API_FORECAST_TEMP,
ATTR_API_FORECAST_TEMP_LOW,
ATTR_API_FORECAST_TIME,
ATTR_API_FORECAST_WIND_BEARING,
ATTR_API_FORECAST_WIND_MAX_SPEED,
ATTR_API_FORECAST_WIND_SPEED,
ATTR_API_HUMIDITY,
ATTR_API_PRESSURE,
ATTR_API_RAIN,
ATTR_API_RAIN_PROB,
ATTR_API_SNOW,
ATTR_API_SNOW_PROB,
ATTR_API_STATION_ID,
ATTR_API_STATION_NAME,
ATTR_API_STATION_TIMESTAMP,
ATTR_API_STORM_PROB,
ATTR_API_TEMPERATURE,
ATTR_API_TEMPERATURE_FEELING,
ATTR_API_TOWN_ID,
ATTR_API_TOWN_NAME,
ATTR_API_TOWN_TIMESTAMP,
ATTR_API_WIND_BEARING,
ATTR_API_WIND_MAX_SPEED,
ATTR_API_WIND_SPEED,
CONDITIONS_MAP,
DOMAIN,
FORECAST_MAP,
)
from .const import CONDITIONS_MAP, DOMAIN, FORECAST_MAP
_LOGGER = logging.getLogger(__name__)
API_TIMEOUT: Final[int] = 120
STATION_MAX_DELTA = timedelta(hours=2)
WEATHER_UPDATE_INTERVAL = timedelta(minutes=10)
def format_condition(condition: str) -> str:
"""Return condition from dict CONDITIONS_MAP."""
val = ForecastValue.parse_condition(condition)
return CONDITIONS_MAP.get(val, val)
def format_float(value) -> float | None:
"""Try converting string to float."""
try:
return float(value)
except (TypeError, ValueError):
return None
def format_int(value) -> int | None:
"""Try converting string to int."""
try:
return int(value)
except (TypeError, ValueError):
return None
class WeatherUpdateCoordinator(DataUpdateCoordinator):
"""Weather data update coordinator."""
@ -143,139 +54,14 @@ class WeatherUpdateCoordinator(DataUpdateCoordinator):
await self.aemet.update()
except AemetError as error:
raise UpdateFailed(error) from error
weather_response = self.aemet.legacy_weather()
return self._convert_weather_response(weather_response)
def _convert_weather_response(self, weather_response):
"""Format the weather response correctly."""
if not weather_response or not weather_response.hourly:
return None
elaborated = dt_util.parse_datetime(
weather_response.hourly[ATTR_DATA][0][AEMET_ATTR_ELABORATED] + "Z"
)
now = dt_util.now()
now_utc = dt_util.utcnow()
hour = now.hour
# Get current day
day = None
for cur_day in weather_response.hourly[ATTR_DATA][0][AEMET_ATTR_FORECAST][
AEMET_ATTR_DAY
]:
cur_day_date = dt_util.parse_datetime(cur_day[AEMET_ATTR_DATE])
if now.date() == cur_day_date.date():
day = cur_day
break
# Get latest station data
station_data = None
station_dt = None
if weather_response.station:
for _station_data in weather_response.station[ATTR_DATA]:
if AEMET_ATTR_STATION_DATE in _station_data:
_station_dt = dt_util.parse_datetime(
_station_data[AEMET_ATTR_STATION_DATE] + "Z"
)
if not station_dt or _station_dt > station_dt:
station_data = _station_data
station_dt = _station_dt
condition = None
humidity = None
pressure = None
rain = None
rain_prob = None
snow = None
snow_prob = None
station_id = None
station_name = None
station_timestamp = None
storm_prob = None
temperature = None
temperature_feeling = None
town_id = None
town_name = None
town_timestamp = dt_util.as_utc(elaborated)
wind_bearing = None
wind_max_speed = None
wind_speed = None
# Get weather values
if day:
condition = self._get_condition(day, hour)
humidity = self._get_humidity(day, hour)
rain = self._get_rain(day, hour)
rain_prob = self._get_rain_prob(day, hour)
snow = self._get_snow(day, hour)
snow_prob = self._get_snow_prob(day, hour)
station_id = self._get_station_id()
station_name = self._get_station_name()
storm_prob = self._get_storm_prob(day, hour)
temperature = self._get_temperature(day, hour)
temperature_feeling = self._get_temperature_feeling(day, hour)
town_id = self._get_town_id()
town_name = self._get_town_name()
wind_bearing = self._get_wind_bearing(day, hour)
wind_max_speed = self._get_wind_max_speed(day, hour)
wind_speed = self._get_wind_speed(day, hour)
# Overwrite weather values with closest station data (if present)
if station_data:
station_timestamp = dt_util.as_utc(station_dt)
if (now_utc - station_dt) <= STATION_MAX_DELTA:
if AEMET_ATTR_STATION_HUMIDITY in station_data:
humidity = format_float(station_data[AEMET_ATTR_STATION_HUMIDITY])
if AEMET_ATTR_STATION_PRESSURE_SEA in station_data:
pressure = format_float(
station_data[AEMET_ATTR_STATION_PRESSURE_SEA]
)
elif AEMET_ATTR_STATION_PRESSURE in station_data:
pressure = format_float(station_data[AEMET_ATTR_STATION_PRESSURE])
if AEMET_ATTR_STATION_TEMPERATURE in station_data:
temperature = format_float(
station_data[AEMET_ATTR_STATION_TEMPERATURE]
)
else:
_LOGGER.warning("Station data is outdated")
# Get forecast from weather data
forecast_daily = self._get_daily_forecast_from_weather_response(
weather_response, now
)
forecast_hourly = self._get_hourly_forecast_from_weather_response(
weather_response, now
)
data = self.aemet.data()
forecasts: list[dict[str, Forecast]] = {
AOD_FORECAST_DAILY: self.aemet_forecast(data, AOD_FORECAST_DAILY),
AOD_FORECAST_HOURLY: self.aemet_forecast(data, AOD_FORECAST_HOURLY),
}
return {
ATTR_API_CONDITION: condition,
ATTR_API_FORECAST_DAILY: forecast_daily,
ATTR_API_FORECAST_HOURLY: forecast_hourly,
ATTR_API_HUMIDITY: humidity,
ATTR_API_TEMPERATURE: temperature,
ATTR_API_TEMPERATURE_FEELING: temperature_feeling,
ATTR_API_PRESSURE: pressure,
ATTR_API_RAIN: rain,
ATTR_API_RAIN_PROB: rain_prob,
ATTR_API_SNOW: snow,
ATTR_API_SNOW_PROB: snow_prob,
ATTR_API_STATION_ID: station_id,
ATTR_API_STATION_NAME: station_name,
ATTR_API_STATION_TIMESTAMP: station_timestamp,
ATTR_API_STORM_PROB: storm_prob,
ATTR_API_TOWN_ID: town_id,
ATTR_API_TOWN_NAME: town_name,
ATTR_API_TOWN_TIMESTAMP: town_timestamp,
ATTR_API_WIND_BEARING: wind_bearing,
ATTR_API_WIND_MAX_SPEED: wind_max_speed,
ATTR_API_WIND_SPEED: wind_speed,
"forecast": forecasts,
"forecast": {
AOD_FORECAST_DAILY: self.aemet_forecast(data, AOD_FORECAST_DAILY),
AOD_FORECAST_HOURLY: self.aemet_forecast(data, AOD_FORECAST_HOURLY),
},
"lib": data,
}
@ -297,262 +83,3 @@ class WeatherUpdateCoordinator(DataUpdateCoordinator):
cur_forecast[ha_key] = value
forecast_list += [cur_forecast]
return cast(list[Forecast], forecast_list)
def _get_daily_forecast_from_weather_response(self, weather_response, now):
if weather_response.daily:
parse = False
forecast = []
for day in weather_response.daily[ATTR_DATA][0][AEMET_ATTR_FORECAST][
AEMET_ATTR_DAY
]:
day_date = dt_util.parse_datetime(day[AEMET_ATTR_DATE])
if now.date() == day_date.date():
parse = True
if parse:
cur_forecast = self._convert_forecast_day(day_date, day)
if cur_forecast:
forecast.append(cur_forecast)
return forecast
return None
def _get_hourly_forecast_from_weather_response(self, weather_response, now):
if weather_response.hourly:
parse = False
hour = now.hour
forecast = []
for day in weather_response.hourly[ATTR_DATA][0][AEMET_ATTR_FORECAST][
AEMET_ATTR_DAY
]:
day_date = dt_util.parse_datetime(day[AEMET_ATTR_DATE])
hour_start = 0
if now.date() == day_date.date():
parse = True
hour_start = now.hour
if parse:
for hour in range(hour_start, 24):
cur_forecast = self._convert_forecast_hour(day_date, day, hour)
if cur_forecast:
forecast.append(cur_forecast)
return forecast
return None
def _convert_forecast_day(self, date, day):
if not (condition := self._get_condition_day(day)):
return None
return {
ATTR_API_FORECAST_CONDITION: condition,
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY: self._get_precipitation_prob_day(
day
),
ATTR_API_FORECAST_TEMP: self._get_temperature_day(day),
ATTR_API_FORECAST_TEMP_LOW: self._get_temperature_low_day(day),
ATTR_API_FORECAST_TIME: dt_util.as_utc(date).isoformat(),
ATTR_API_FORECAST_WIND_SPEED: self._get_wind_speed_day(day),
ATTR_API_FORECAST_WIND_BEARING: self._get_wind_bearing_day(day),
}
def _convert_forecast_hour(self, date, day, hour):
if not (condition := self._get_condition(day, hour)):
return None
forecast_dt = date.replace(hour=hour, minute=0, second=0)
return {
ATTR_API_FORECAST_CONDITION: condition,
ATTR_API_FORECAST_PRECIPITATION: self._calc_precipitation(day, hour),
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY: self._calc_precipitation_prob(
day, hour
),
ATTR_API_FORECAST_TEMP: self._get_temperature(day, hour),
ATTR_API_FORECAST_TIME: dt_util.as_utc(forecast_dt).isoformat(),
ATTR_API_FORECAST_WIND_MAX_SPEED: self._get_wind_max_speed(day, hour),
ATTR_API_FORECAST_WIND_SPEED: self._get_wind_speed(day, hour),
ATTR_API_FORECAST_WIND_BEARING: self._get_wind_bearing(day, hour),
}
def _calc_precipitation(self, day, hour):
"""Calculate the precipitation."""
rain_value = self._get_rain(day, hour) or 0
snow_value = self._get_snow(day, hour) or 0
if round(rain_value + snow_value, 1) == 0:
return None
return round(rain_value + snow_value, 1)
def _calc_precipitation_prob(self, day, hour):
"""Calculate the precipitation probability (hour)."""
rain_value = self._get_rain_prob(day, hour) or 0
snow_value = self._get_snow_prob(day, hour) or 0
if rain_value == 0 and snow_value == 0:
return None
return max(rain_value, snow_value)
@staticmethod
def _get_condition(day_data, hour):
"""Get weather condition (hour) from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_SKY_STATE], hour)
if val:
return format_condition(val)
return None
@staticmethod
def _get_condition_day(day_data):
"""Get weather condition (day) from weather data."""
val = get_forecast_day_value(day_data[AEMET_ATTR_SKY_STATE])
if val:
return format_condition(val)
return None
@staticmethod
def _get_humidity(day_data, hour):
"""Get humidity from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_HUMIDITY], hour)
if val:
return format_int(val)
return None
@staticmethod
def _get_precipitation_prob_day(day_data):
"""Get humidity from weather data."""
val = get_forecast_day_value(day_data[AEMET_ATTR_PRECIPITATION_PROBABILITY])
if val:
return format_int(val)
return None
@staticmethod
def _get_rain(day_data, hour):
"""Get rain from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_PRECIPITATION], hour)
if val:
return format_float(val)
return None
@staticmethod
def _get_rain_prob(day_data, hour):
"""Get rain probability from weather data."""
val = get_forecast_interval_value(
day_data[AEMET_ATTR_PRECIPITATION_PROBABILITY], hour
)
if val:
return format_int(val)
return None
@staticmethod
def _get_snow(day_data, hour):
"""Get snow from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_SNOW], hour)
if val:
return format_float(val)
return None
@staticmethod
def _get_snow_prob(day_data, hour):
"""Get snow probability from weather data."""
val = get_forecast_interval_value(day_data[AEMET_ATTR_SNOW_PROBABILITY], hour)
if val:
return format_int(val)
return None
def _get_station_id(self):
"""Get station ID from weather data."""
if self.aemet.station:
return self.aemet.station.get_id()
return None
def _get_station_name(self):
"""Get station name from weather data."""
if self.aemet.station:
return self.aemet.station.get_name()
return None
@staticmethod
def _get_storm_prob(day_data, hour):
"""Get storm probability from weather data."""
val = get_forecast_interval_value(day_data[AEMET_ATTR_STORM_PROBABILITY], hour)
if val:
return format_int(val)
return None
@staticmethod
def _get_temperature(day_data, hour):
"""Get temperature (hour) from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_TEMPERATURE], hour)
return format_int(val)
@staticmethod
def _get_temperature_day(day_data):
"""Get temperature (day) from weather data."""
val = get_forecast_day_value(
day_data[AEMET_ATTR_TEMPERATURE], key=AEMET_ATTR_MAX
)
return format_int(val)
@staticmethod
def _get_temperature_low_day(day_data):
"""Get temperature (day) from weather data."""
val = get_forecast_day_value(
day_data[AEMET_ATTR_TEMPERATURE], key=AEMET_ATTR_MIN
)
return format_int(val)
@staticmethod
def _get_temperature_feeling(day_data, hour):
"""Get temperature from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_FEEL_TEMPERATURE], hour)
return format_int(val)
def _get_town_id(self):
"""Get town ID from weather data."""
if self.aemet.town:
return self.aemet.town.get_id()
return None
def _get_town_name(self):
"""Get town name from weather data."""
if self.aemet.town:
return self.aemet.town.get_name()
return None
@staticmethod
def _get_wind_bearing(day_data, hour):
"""Get wind bearing (hour) from weather data."""
val = get_forecast_hour_value(
day_data[AEMET_ATTR_WIND_GUST], hour, key=AEMET_ATTR_DIRECTION
)[0]
return ForecastValue.parse_wind_direction(val)
@staticmethod
def _get_wind_bearing_day(day_data):
"""Get wind bearing (day) from weather data."""
val = get_forecast_day_value(
day_data[AEMET_ATTR_WIND], key=AEMET_ATTR_DIRECTION
)
return ForecastValue.parse_wind_direction(val)
@staticmethod
def _get_wind_max_speed(day_data, hour):
"""Get wind max speed from weather data."""
val = get_forecast_hour_value(day_data[AEMET_ATTR_WIND_GUST], hour)
if val:
return format_int(val)
return None
@staticmethod
def _get_wind_speed(day_data, hour):
"""Get wind speed (hour) from weather data."""
val = get_forecast_hour_value(
day_data[AEMET_ATTR_WIND_GUST], hour, key=AEMET_ATTR_SPEED
)[0]
if val:
return format_int(val)
return None
@staticmethod
def _get_wind_speed_day(day_data):
"""Get wind speed (day) from weather data."""
val = get_forecast_day_value(day_data[AEMET_ATTR_WIND], key=AEMET_ATTR_SPEED)
if val:
return format_int(val)
return None

8
tests/components/aemet/test_sensor.py
View File

@ -93,7 +93,7 @@ async def test_aemet_weather_create_sensors(
assert state.state == "1004.4"
state = hass.states.get("sensor.aemet_rain")
assert state.state == "1.8"
assert state.state == "7.0"
state = hass.states.get("sensor.aemet_rain_probability")
assert state.state == "100"
@ -132,10 +132,10 @@ async def test_aemet_weather_create_sensors(
assert state.state == "2021-01-09T11:47:45+00:00"
state = hass.states.get("sensor.aemet_wind_bearing")
assert state.state == "90.0"
assert state.state == "122.0"
state = hass.states.get("sensor.aemet_wind_max_speed")
assert state.state == "24"
assert state.state == "12.2"
state = hass.states.get("sensor.aemet_wind_speed")
assert state.state == "15"
assert state.state == "3.2"