mirror of
https://github.com/home-assistant/core.git
synced 2025-12-21 03:20:01 +00:00
822 lines
31 KiB
Python
822 lines
31 KiB
Python
"""Template platform that aggregates meteorological data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import asdict, dataclass
|
|
from functools import partial
|
|
from typing import TYPE_CHECKING, Any, Literal, Self
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.weather import (
|
|
ATTR_CONDITION_CLEAR_NIGHT,
|
|
ATTR_CONDITION_CLOUDY,
|
|
ATTR_CONDITION_EXCEPTIONAL,
|
|
ATTR_CONDITION_FOG,
|
|
ATTR_CONDITION_HAIL,
|
|
ATTR_CONDITION_LIGHTNING,
|
|
ATTR_CONDITION_LIGHTNING_RAINY,
|
|
ATTR_CONDITION_PARTLYCLOUDY,
|
|
ATTR_CONDITION_POURING,
|
|
ATTR_CONDITION_RAINY,
|
|
ATTR_CONDITION_SNOWY,
|
|
ATTR_CONDITION_SNOWY_RAINY,
|
|
ATTR_CONDITION_SUNNY,
|
|
ATTR_CONDITION_WINDY,
|
|
ATTR_CONDITION_WINDY_VARIANT,
|
|
DOMAIN as WEATHER_DOMAIN,
|
|
ENTITY_ID_FORMAT,
|
|
PLATFORM_SCHEMA as WEATHER_PLATFORM_SCHEMA,
|
|
Forecast,
|
|
WeatherEntity,
|
|
WeatherEntityFeature,
|
|
)
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import (
|
|
CONF_NAME,
|
|
CONF_TEMPERATURE_UNIT,
|
|
CONF_UNIQUE_ID,
|
|
STATE_UNAVAILABLE,
|
|
STATE_UNKNOWN,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import TemplateError
|
|
from homeassistant.helpers import config_validation as cv, template
|
|
from homeassistant.helpers.entity_platform import (
|
|
AddConfigEntryEntitiesCallback,
|
|
AddEntitiesCallback,
|
|
)
|
|
from homeassistant.helpers.restore_state import ExtraStoredData, RestoreEntity
|
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
|
from homeassistant.util.unit_conversion import (
|
|
DistanceConverter,
|
|
PressureConverter,
|
|
SpeedConverter,
|
|
TemperatureConverter,
|
|
)
|
|
|
|
from .coordinator import TriggerUpdateCoordinator
|
|
from .entity import AbstractTemplateEntity
|
|
from .helpers import (
|
|
async_setup_template_entry,
|
|
async_setup_template_platform,
|
|
async_setup_template_preview,
|
|
rewrite_legacy_to_modern_config,
|
|
)
|
|
from .schemas import (
|
|
TEMPLATE_ENTITY_COMMON_CONFIG_ENTRY_SCHEMA,
|
|
make_template_entity_common_modern_schema,
|
|
)
|
|
from .template_entity import TemplateEntity
|
|
from .trigger_entity import TriggerEntity
|
|
|
|
CHECK_FORECAST_KEYS = (
|
|
set()
|
|
.union(Forecast.__annotations__.keys())
|
|
# Manually add the forecast resulting attributes that only exists
|
|
# as native_* in the Forecast definition
|
|
.union(("apparent_temperature", "wind_gust_speed", "dew_point"))
|
|
)
|
|
|
|
CONDITION_CLASSES = {
|
|
ATTR_CONDITION_CLEAR_NIGHT,
|
|
ATTR_CONDITION_CLOUDY,
|
|
ATTR_CONDITION_FOG,
|
|
ATTR_CONDITION_HAIL,
|
|
ATTR_CONDITION_LIGHTNING,
|
|
ATTR_CONDITION_LIGHTNING_RAINY,
|
|
ATTR_CONDITION_PARTLYCLOUDY,
|
|
ATTR_CONDITION_POURING,
|
|
ATTR_CONDITION_RAINY,
|
|
ATTR_CONDITION_SNOWY,
|
|
ATTR_CONDITION_SNOWY_RAINY,
|
|
ATTR_CONDITION_SUNNY,
|
|
ATTR_CONDITION_WINDY,
|
|
ATTR_CONDITION_WINDY_VARIANT,
|
|
ATTR_CONDITION_EXCEPTIONAL,
|
|
}
|
|
|
|
CONF_APPARENT_TEMPERATURE = "apparent_temperature"
|
|
CONF_APPARENT_TEMPERATURE_TEMPLATE = "apparent_temperature_template"
|
|
CONF_ATTRIBUTION = "attribution"
|
|
CONF_ATTRIBUTION_TEMPLATE = "attribution_template"
|
|
CONF_CLOUD_COVERAGE = "cloud_coverage"
|
|
CONF_CLOUD_COVERAGE_TEMPLATE = "cloud_coverage_template"
|
|
CONF_CONDITION = "condition"
|
|
CONF_CONDITION_TEMPLATE = "condition_template"
|
|
CONF_DEW_POINT = "dew_point"
|
|
CONF_DEW_POINT_TEMPLATE = "dew_point_template"
|
|
CONF_FORECAST_DAILY = "forecast_daily"
|
|
CONF_FORECAST_DAILY_TEMPLATE = "forecast_daily_template"
|
|
CONF_FORECAST_HOURLY = "forecast_hourly"
|
|
CONF_FORECAST_HOURLY_TEMPLATE = "forecast_hourly_template"
|
|
CONF_FORECAST_TWICE_DAILY = "forecast_twice_daily"
|
|
CONF_FORECAST_TWICE_DAILY_TEMPLATE = "forecast_twice_daily_template"
|
|
CONF_HUMIDITY = "humidity"
|
|
CONF_HUMIDITY_TEMPLATE = "humidity_template"
|
|
CONF_OZONE = "ozone"
|
|
CONF_OZONE_TEMPLATE = "ozone_template"
|
|
CONF_PRECIPITATION_UNIT = "precipitation_unit"
|
|
CONF_PRESSURE = "pressure"
|
|
CONF_PRESSURE_TEMPLATE = "pressure_template"
|
|
CONF_PRESSURE_UNIT = "pressure_unit"
|
|
CONF_TEMPERATURE = "temperature"
|
|
CONF_TEMPERATURE_TEMPLATE = "temperature_template"
|
|
CONF_UV_INDEX = "uv_index"
|
|
CONF_UV_INDEX_TEMPLATE = "uv_index_template"
|
|
CONF_VISIBILITY = "visibility"
|
|
CONF_VISIBILITY_TEMPLATE = "visibility_template"
|
|
CONF_VISIBILITY_UNIT = "visibility_unit"
|
|
CONF_WEATHER = "weather"
|
|
CONF_WIND_BEARING = "wind_bearing"
|
|
CONF_WIND_BEARING_TEMPLATE = "wind_bearing_template"
|
|
CONF_WIND_GUST_SPEED = "wind_gust_speed"
|
|
CONF_WIND_GUST_SPEED_TEMPLATE = "wind_gust_speed_template"
|
|
CONF_WIND_SPEED = "wind_speed"
|
|
CONF_WIND_SPEED_TEMPLATE = "wind_speed_template"
|
|
CONF_WIND_SPEED_UNIT = "wind_speed_unit"
|
|
|
|
DEFAULT_NAME = "Template Weather"
|
|
|
|
LEGACY_FIELDS = {
|
|
CONF_APPARENT_TEMPERATURE_TEMPLATE: CONF_APPARENT_TEMPERATURE,
|
|
CONF_ATTRIBUTION_TEMPLATE: CONF_ATTRIBUTION,
|
|
CONF_CLOUD_COVERAGE_TEMPLATE: CONF_CLOUD_COVERAGE,
|
|
CONF_CONDITION_TEMPLATE: CONF_CONDITION,
|
|
CONF_DEW_POINT_TEMPLATE: CONF_DEW_POINT,
|
|
CONF_FORECAST_DAILY_TEMPLATE: CONF_FORECAST_DAILY,
|
|
CONF_FORECAST_HOURLY_TEMPLATE: CONF_FORECAST_HOURLY,
|
|
CONF_FORECAST_TWICE_DAILY_TEMPLATE: CONF_FORECAST_TWICE_DAILY,
|
|
CONF_HUMIDITY_TEMPLATE: CONF_HUMIDITY,
|
|
CONF_OZONE_TEMPLATE: CONF_OZONE,
|
|
CONF_PRESSURE_TEMPLATE: CONF_PRESSURE,
|
|
CONF_TEMPERATURE_TEMPLATE: CONF_TEMPERATURE,
|
|
CONF_UV_INDEX_TEMPLATE: CONF_UV_INDEX,
|
|
CONF_VISIBILITY_TEMPLATE: CONF_VISIBILITY,
|
|
CONF_WIND_BEARING_TEMPLATE: CONF_WIND_BEARING,
|
|
CONF_WIND_GUST_SPEED_TEMPLATE: CONF_WIND_GUST_SPEED,
|
|
CONF_WIND_SPEED_TEMPLATE: CONF_WIND_SPEED,
|
|
}
|
|
|
|
|
|
WEATHER_COMMON_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_APPARENT_TEMPERATURE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_ATTRIBUTION_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_CLOUD_COVERAGE_TEMPLATE): cv.template,
|
|
vol.Required(CONF_CONDITION_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_DEW_POINT_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_FORECAST_DAILY_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_FORECAST_HOURLY_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_FORECAST_TWICE_DAILY_TEMPLATE): cv.template,
|
|
vol.Required(CONF_HUMIDITY_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_OZONE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_PRECIPITATION_UNIT): vol.In(DistanceConverter.VALID_UNITS),
|
|
vol.Optional(CONF_PRESSURE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_PRESSURE_UNIT): vol.In(PressureConverter.VALID_UNITS),
|
|
vol.Required(CONF_TEMPERATURE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_TEMPERATURE_UNIT): vol.In(TemperatureConverter.VALID_UNITS),
|
|
vol.Optional(CONF_VISIBILITY_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_VISIBILITY_UNIT): vol.In(DistanceConverter.VALID_UNITS),
|
|
vol.Optional(CONF_WIND_BEARING_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_WIND_GUST_SPEED_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_WIND_SPEED_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_WIND_SPEED_UNIT): vol.In(SpeedConverter.VALID_UNITS),
|
|
}
|
|
)
|
|
|
|
|
|
WEATHER_YAML_SCHEMA = (
|
|
vol.Schema(
|
|
{
|
|
vol.Optional(CONF_UV_INDEX_TEMPLATE): cv.template,
|
|
}
|
|
)
|
|
.extend(WEATHER_COMMON_SCHEMA.schema)
|
|
.extend(
|
|
make_template_entity_common_modern_schema(WEATHER_DOMAIN, DEFAULT_NAME).schema
|
|
)
|
|
)
|
|
|
|
PLATFORM_SCHEMA = (
|
|
vol.Schema(
|
|
{
|
|
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.template,
|
|
vol.Optional(CONF_UNIQUE_ID): cv.string,
|
|
}
|
|
)
|
|
.extend(WEATHER_COMMON_SCHEMA.schema)
|
|
.extend(WEATHER_PLATFORM_SCHEMA.schema)
|
|
)
|
|
|
|
|
|
WEATHER_CONFIG_ENTRY_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_APPARENT_TEMPERATURE): cv.template,
|
|
vol.Optional(CONF_ATTRIBUTION): cv.template,
|
|
vol.Optional(CONF_CLOUD_COVERAGE): cv.template,
|
|
vol.Required(CONF_CONDITION): cv.template,
|
|
vol.Optional(CONF_DEW_POINT): cv.template,
|
|
vol.Optional(CONF_FORECAST_DAILY): cv.template,
|
|
vol.Optional(CONF_FORECAST_HOURLY): cv.template,
|
|
vol.Optional(CONF_FORECAST_TWICE_DAILY): cv.template,
|
|
vol.Required(CONF_HUMIDITY): cv.template,
|
|
vol.Optional(CONF_OZONE): cv.template,
|
|
vol.Optional(CONF_PRECIPITATION_UNIT): vol.In(DistanceConverter.VALID_UNITS),
|
|
vol.Optional(CONF_PRESSURE): cv.template,
|
|
vol.Optional(CONF_PRESSURE_UNIT): vol.In(PressureConverter.VALID_UNITS),
|
|
vol.Required(CONF_TEMPERATURE): cv.template,
|
|
vol.Optional(CONF_TEMPERATURE_UNIT): vol.In(TemperatureConverter.VALID_UNITS),
|
|
vol.Optional(CONF_UV_INDEX): cv.template,
|
|
vol.Optional(CONF_VISIBILITY): cv.template,
|
|
vol.Optional(CONF_VISIBILITY_UNIT): vol.In(DistanceConverter.VALID_UNITS),
|
|
vol.Optional(CONF_WIND_BEARING): cv.template,
|
|
vol.Optional(CONF_WIND_GUST_SPEED): cv.template,
|
|
vol.Optional(CONF_WIND_SPEED): cv.template,
|
|
vol.Optional(CONF_WIND_SPEED_UNIT): vol.In(SpeedConverter.VALID_UNITS),
|
|
}
|
|
).extend(TEMPLATE_ENTITY_COMMON_CONFIG_ENTRY_SCHEMA.schema)
|
|
|
|
|
|
async def async_setup_platform(
|
|
hass: HomeAssistant,
|
|
config: ConfigType,
|
|
async_add_entities: AddEntitiesCallback,
|
|
discovery_info: DiscoveryInfoType | None = None,
|
|
) -> None:
|
|
"""Set up the Template weather."""
|
|
|
|
# Rewrite the configuration options to modern keys.
|
|
if discovery_info is None:
|
|
# Legacy
|
|
config = rewrite_legacy_to_modern_config(hass, config, LEGACY_FIELDS)
|
|
else:
|
|
# Modern and Trigger
|
|
entity_configs: list[ConfigType] = discovery_info["entities"]
|
|
modified_entity_configs = []
|
|
for entity_config in entity_configs:
|
|
entity_config = rewrite_legacy_to_modern_config(
|
|
hass, entity_config, LEGACY_FIELDS
|
|
)
|
|
|
|
modified_entity_configs.append(entity_config)
|
|
|
|
if modified_entity_configs:
|
|
discovery_info["entities"] = modified_entity_configs
|
|
|
|
await async_setup_template_platform(
|
|
hass,
|
|
WEATHER_DOMAIN,
|
|
config,
|
|
StateWeatherEntity,
|
|
TriggerWeatherEntity,
|
|
async_add_entities,
|
|
discovery_info,
|
|
{},
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Initialize config entry."""
|
|
await async_setup_template_entry(
|
|
hass,
|
|
config_entry,
|
|
async_add_entities,
|
|
StateWeatherEntity,
|
|
WEATHER_CONFIG_ENTRY_SCHEMA,
|
|
)
|
|
|
|
|
|
@callback
|
|
def async_create_preview_weather(
|
|
hass: HomeAssistant, name: str, config: dict[str, Any]
|
|
) -> StateWeatherEntity:
|
|
"""Create a preview."""
|
|
return async_setup_template_preview(
|
|
hass,
|
|
name,
|
|
config,
|
|
StateWeatherEntity,
|
|
WEATHER_CONFIG_ENTRY_SCHEMA,
|
|
)
|
|
|
|
|
|
class AbstractTemplateWeather(AbstractTemplateEntity, WeatherEntity):
|
|
"""Representation of a template weathers features."""
|
|
|
|
_entity_id_format = ENTITY_ID_FORMAT
|
|
_optimistic_entity = True
|
|
|
|
# The super init is not called because TemplateEntity and TriggerEntity will call AbstractTemplateEntity.__init__.
|
|
# This ensures that the __init__ on AbstractTemplateEntity is not called twice.
|
|
def __init__( # pylint: disable=super-init-not-called
|
|
self, config: dict[str, Any], initial_state: bool | None = False
|
|
) -> None:
|
|
"""Initialize the features."""
|
|
|
|
# Templates
|
|
self._apparent_temperature_template = config.get(CONF_APPARENT_TEMPERATURE)
|
|
self._attribution_template = config.get(CONF_ATTRIBUTION)
|
|
self._cloud_coverage_template = config.get(CONF_CLOUD_COVERAGE)
|
|
self._condition_template = config[CONF_CONDITION]
|
|
self._dew_point_template = config.get(CONF_DEW_POINT)
|
|
self._forecast_daily_template = config.get(CONF_FORECAST_DAILY)
|
|
self._forecast_hourly_template = config.get(CONF_FORECAST_HOURLY)
|
|
self._forecast_twice_daily_template = config.get(CONF_FORECAST_TWICE_DAILY)
|
|
self._humidity_template = config[CONF_HUMIDITY]
|
|
self._ozone_template = config.get(CONF_OZONE)
|
|
self._pressure_template = config.get(CONF_PRESSURE)
|
|
self._temperature_template = config[CONF_TEMPERATURE]
|
|
self._uv_index_template = config.get(CONF_UV_INDEX)
|
|
self._visibility_template = config.get(CONF_VISIBILITY)
|
|
self._wind_bearing_template = config.get(CONF_WIND_BEARING)
|
|
self._wind_gust_speed_template = config.get(CONF_WIND_GUST_SPEED)
|
|
self._wind_speed_template = config.get(CONF_WIND_SPEED)
|
|
|
|
# Legacy support
|
|
self._attribution: str | None = None
|
|
|
|
# Native units
|
|
self._attr_native_precipitation_unit = config.get(CONF_PRECIPITATION_UNIT)
|
|
self._attr_native_pressure_unit = config.get(CONF_PRESSURE_UNIT)
|
|
self._attr_native_temperature_unit = config.get(CONF_TEMPERATURE_UNIT)
|
|
self._attr_native_visibility_unit = config.get(CONF_VISIBILITY_UNIT)
|
|
self._attr_native_wind_speed_unit = config.get(CONF_WIND_SPEED_UNIT)
|
|
|
|
# Supported Features
|
|
self._attr_supported_features = 0
|
|
if self._forecast_daily_template:
|
|
self._attr_supported_features |= WeatherEntityFeature.FORECAST_DAILY
|
|
if self._forecast_hourly_template:
|
|
self._attr_supported_features |= WeatherEntityFeature.FORECAST_HOURLY
|
|
if self._forecast_twice_daily_template:
|
|
self._attr_supported_features |= WeatherEntityFeature.FORECAST_TWICE_DAILY
|
|
|
|
@property
|
|
def attribution(self) -> str | None:
|
|
"""Return the attribution."""
|
|
if self._attribution is None:
|
|
return "Powered by Home Assistant"
|
|
return self._attribution
|
|
|
|
def _validate[T](
|
|
self,
|
|
validator: Callable[[Any], T],
|
|
result: Any,
|
|
) -> T | None:
|
|
try:
|
|
return validator(result)
|
|
except vol.Invalid:
|
|
return None
|
|
|
|
@callback
|
|
def _update_apparent_temperature(self, result: Any) -> None:
|
|
self._attr_native_apparent_temperature = self._validate(
|
|
vol.Coerce(float), result
|
|
)
|
|
|
|
@callback
|
|
def _update_attribution(self, result: Any) -> None:
|
|
self._attribution = vol.Coerce(str)(result)
|
|
|
|
@callback
|
|
def _update_condition(self, result: Any) -> None:
|
|
self._attr_condition = result if result in CONDITION_CLASSES else None
|
|
|
|
@callback
|
|
def _update_coverage(self, result: Any) -> None:
|
|
self._attr_cloud_coverage = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_dew_point(self, result: Any) -> None:
|
|
self._attr_native_dew_point = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_humidity(self, result: Any) -> None:
|
|
self._attr_humidity = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_ozone(self, result: Any) -> None:
|
|
self._attr_ozone = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_pressure(self, result: Any) -> None:
|
|
self._attr_native_pressure = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_temperature(self, result: Any) -> None:
|
|
self._attr_native_temperature = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_uv_index(self, result: Any) -> None:
|
|
self._attr_uv_index = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_visibility(self, result: Any) -> None:
|
|
self._attr_native_visibility = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_wind_bearing(self, result: Any) -> None:
|
|
try:
|
|
self._attr_wind_bearing = vol.Coerce(float)(result)
|
|
except vol.Invalid:
|
|
self._attr_wind_bearing = vol.Coerce(str)(result)
|
|
|
|
@callback
|
|
def _update_wind_gust_speed(self, result: Any) -> None:
|
|
self._attr_native_wind_gust_speed = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _update_wind_speed(self, result: Any) -> None:
|
|
self._attr_native_wind_speed = self._validate(vol.Coerce(float), result)
|
|
|
|
@callback
|
|
def _validate_forecast(
|
|
self,
|
|
forecast_type: Literal["daily", "hourly", "twice_daily"],
|
|
result: Any,
|
|
) -> list[Forecast] | None:
|
|
"""Validate the forecasts."""
|
|
if result is None:
|
|
return None
|
|
|
|
if not isinstance(result, list):
|
|
raise vol.Invalid(
|
|
"Forecasts is not a list, see Weather documentation https://www.home-assistant.io/integrations/weather/"
|
|
)
|
|
for forecast in result:
|
|
if not isinstance(forecast, dict):
|
|
raise vol.Invalid(
|
|
"Forecast in list is not a dict, see Weather documentation https://www.home-assistant.io/integrations/weather/"
|
|
)
|
|
diff_result = set().union(forecast.keys()).difference(CHECK_FORECAST_KEYS)
|
|
if diff_result:
|
|
raise vol.Invalid(
|
|
f"Only valid keys in Forecast are allowed, unallowed keys: ({diff_result}), "
|
|
"see Weather documentation https://www.home-assistant.io/integrations/weather/"
|
|
)
|
|
if forecast_type == "twice_daily" and "is_daytime" not in forecast:
|
|
raise vol.Invalid(
|
|
"`is_daytime` is missing in twice_daily forecast, see Weather documentation https://www.home-assistant.io/integrations/weather/"
|
|
)
|
|
if "datetime" not in forecast:
|
|
raise vol.Invalid(
|
|
"`datetime` is required in forecasts, see Weather documentation https://www.home-assistant.io/integrations/weather/"
|
|
)
|
|
continue
|
|
return result
|
|
|
|
|
|
class StateWeatherEntity(TemplateEntity, AbstractTemplateWeather):
|
|
"""Representation of a Template weather."""
|
|
|
|
_attr_should_poll = False
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
config: ConfigType,
|
|
unique_id: str | None,
|
|
) -> None:
|
|
"""Initialize the Template weather."""
|
|
TemplateEntity.__init__(self, hass, config, unique_id)
|
|
AbstractTemplateWeather.__init__(self, config)
|
|
|
|
name = self._attr_name
|
|
if TYPE_CHECKING:
|
|
assert name is not None
|
|
|
|
# Forecasts
|
|
self._forecast_daily: list[Forecast] | None = []
|
|
self._forecast_hourly: list[Forecast] | None = []
|
|
self._forecast_twice_daily: list[Forecast] | None = []
|
|
|
|
@callback
|
|
def _async_setup_templates(self) -> None:
|
|
"""Set up templates."""
|
|
|
|
if self._apparent_temperature_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_apparent_temperature",
|
|
self._apparent_temperature_template,
|
|
on_update=self._update_apparent_temperature,
|
|
)
|
|
if self._attribution_template:
|
|
self.add_template_attribute(
|
|
"_attribution",
|
|
self._attribution_template,
|
|
on_update=self._update_attribution,
|
|
)
|
|
if self._cloud_coverage_template:
|
|
self.add_template_attribute(
|
|
"_attr_cloud_coverage",
|
|
self._cloud_coverage_template,
|
|
on_update=self._update_coverage,
|
|
)
|
|
if self._condition_template:
|
|
self.add_template_attribute(
|
|
"_attr_condition",
|
|
self._condition_template,
|
|
on_update=self._update_condition,
|
|
)
|
|
if self._dew_point_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_dew_point",
|
|
self._dew_point_template,
|
|
on_update=self._update_dew_point,
|
|
)
|
|
if self._forecast_daily_template:
|
|
self.add_template_attribute(
|
|
"_forecast_daily",
|
|
self._forecast_daily_template,
|
|
on_update=partial(self._update_forecast, "daily"),
|
|
validator=partial(self._validate_forecast, "daily"),
|
|
)
|
|
if self._forecast_hourly_template:
|
|
self.add_template_attribute(
|
|
"_forecast_hourly",
|
|
self._forecast_hourly_template,
|
|
on_update=partial(self._update_forecast, "hourly"),
|
|
validator=partial(self._validate_forecast, "hourly"),
|
|
)
|
|
if self._forecast_twice_daily_template:
|
|
self.add_template_attribute(
|
|
"_forecast_twice_daily",
|
|
self._forecast_twice_daily_template,
|
|
on_update=partial(self._update_forecast, "twice_daily"),
|
|
validator=partial(self._validate_forecast, "twice_daily"),
|
|
)
|
|
if self._humidity_template:
|
|
self.add_template_attribute(
|
|
"_attr_humidity",
|
|
self._humidity_template,
|
|
on_update=self._update_humidity,
|
|
)
|
|
if self._ozone_template:
|
|
self.add_template_attribute(
|
|
"_attr_ozone",
|
|
self._ozone_template,
|
|
on_update=self._update_ozone,
|
|
)
|
|
if self._pressure_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_pressure",
|
|
self._pressure_template,
|
|
on_update=self._update_pressure,
|
|
)
|
|
if self._temperature_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_temperature",
|
|
self._temperature_template,
|
|
on_update=self._update_temperature,
|
|
)
|
|
if self._uv_index_template:
|
|
self.add_template_attribute(
|
|
"_attr_uv_index",
|
|
self._uv_index_template,
|
|
on_update=self._update_uv_index,
|
|
)
|
|
if self._visibility_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_visibility",
|
|
self._visibility_template,
|
|
on_update=self._update_visibility,
|
|
)
|
|
if self._wind_bearing_template:
|
|
self.add_template_attribute(
|
|
"_attr_wind_bearing",
|
|
self._wind_bearing_template,
|
|
on_update=self._update_wind_bearing,
|
|
)
|
|
if self._wind_gust_speed_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_wind_gust_speed",
|
|
self._wind_gust_speed_template,
|
|
on_update=self._update_wind_gust_speed,
|
|
)
|
|
if self._wind_speed_template:
|
|
self.add_template_attribute(
|
|
"_attr_native_wind_speed",
|
|
self._wind_speed_template,
|
|
on_update=self._update_wind_speed,
|
|
)
|
|
|
|
super()._async_setup_templates()
|
|
|
|
async def async_forecast_daily(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return self._forecast_daily or []
|
|
|
|
async def async_forecast_hourly(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return self._forecast_hourly or []
|
|
|
|
async def async_forecast_twice_daily(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return self._forecast_twice_daily or []
|
|
|
|
@callback
|
|
def _update_forecast(
|
|
self,
|
|
forecast_type: Literal["daily", "hourly", "twice_daily"],
|
|
result: list[Forecast] | TemplateError,
|
|
) -> None:
|
|
"""Save template result and trigger forecast listener."""
|
|
attr_result = None if isinstance(result, TemplateError) else result
|
|
setattr(self, f"_forecast_{forecast_type}", attr_result)
|
|
self.hass.async_create_task(
|
|
self.async_update_listeners([forecast_type]), eager_start=True
|
|
)
|
|
|
|
|
|
@dataclass(kw_only=True)
|
|
class WeatherExtraStoredData(ExtraStoredData):
|
|
"""Object to hold extra stored data."""
|
|
|
|
last_apparent_temperature: float | None
|
|
last_cloud_coverage: int | None
|
|
last_dew_point: float | None
|
|
last_humidity: float | None
|
|
last_ozone: float | None
|
|
last_pressure: float | None
|
|
last_temperature: float | None
|
|
last_uv_index: float | None
|
|
last_visibility: float | None
|
|
last_wind_bearing: float | str | None
|
|
last_wind_gust_speed: float | None
|
|
last_wind_speed: float | None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
"""Return a dict representation of the event data."""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, restored: dict[str, Any]) -> Self | None:
|
|
"""Initialize a stored event state from a dict."""
|
|
try:
|
|
return cls(
|
|
last_apparent_temperature=restored["last_apparent_temperature"],
|
|
last_cloud_coverage=restored["last_cloud_coverage"],
|
|
last_dew_point=restored["last_dew_point"],
|
|
last_humidity=restored["last_humidity"],
|
|
last_ozone=restored["last_ozone"],
|
|
last_pressure=restored["last_pressure"],
|
|
last_temperature=restored["last_temperature"],
|
|
last_uv_index=restored["last_uv_index"],
|
|
last_visibility=restored["last_visibility"],
|
|
last_wind_bearing=restored["last_wind_bearing"],
|
|
last_wind_gust_speed=restored["last_wind_gust_speed"],
|
|
last_wind_speed=restored["last_wind_speed"],
|
|
)
|
|
except KeyError:
|
|
return None
|
|
|
|
|
|
class TriggerWeatherEntity(TriggerEntity, AbstractTemplateWeather, RestoreEntity):
|
|
"""Weather entity based on trigger data."""
|
|
|
|
domain = WEATHER_DOMAIN
|
|
extra_template_keys = (
|
|
CONF_CONDITION,
|
|
CONF_TEMPERATURE,
|
|
CONF_HUMIDITY,
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
coordinator: TriggerUpdateCoordinator,
|
|
config: ConfigType,
|
|
) -> None:
|
|
"""Initialize."""
|
|
TriggerEntity.__init__(self, hass, coordinator, config)
|
|
AbstractTemplateWeather.__init__(self, config, None)
|
|
|
|
for key in (
|
|
CONF_APPARENT_TEMPERATURE,
|
|
CONF_CLOUD_COVERAGE,
|
|
CONF_DEW_POINT,
|
|
CONF_FORECAST_DAILY,
|
|
CONF_FORECAST_HOURLY,
|
|
CONF_FORECAST_TWICE_DAILY,
|
|
CONF_OZONE,
|
|
CONF_PRESSURE,
|
|
CONF_UV_INDEX,
|
|
CONF_VISIBILITY,
|
|
CONF_WIND_BEARING,
|
|
CONF_WIND_GUST_SPEED,
|
|
CONF_WIND_SPEED,
|
|
):
|
|
if isinstance(config.get(key), template.Template):
|
|
self._to_render_simple.append(key)
|
|
self._parse_result.add(key)
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
"""Restore last state."""
|
|
await super().async_added_to_hass()
|
|
if (
|
|
(state := await self.async_get_last_state())
|
|
and state.state is not None
|
|
and state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
|
|
and (weather_data := await self.async_get_last_weather_data())
|
|
):
|
|
self._attr_native_apparent_temperature = (
|
|
weather_data.last_apparent_temperature
|
|
)
|
|
self._attr_cloud_coverage = weather_data.last_cloud_coverage
|
|
self._attr_condition = state.state
|
|
self._attr_native_dew_point = weather_data.last_dew_point
|
|
self._attr_humidity = weather_data.last_humidity
|
|
self._attr_ozone = weather_data.last_ozone
|
|
self._attr_native_pressure = weather_data.last_pressure
|
|
self._attr_native_temperature = weather_data.last_temperature
|
|
self._attr_uv_index = weather_data.last_uv_index
|
|
self._attr_native_visibility = weather_data.last_visibility
|
|
self._attr_wind_bearing = weather_data.last_wind_bearing
|
|
self._attr_native_wind_gust_speed = weather_data.last_wind_gust_speed
|
|
self._attr_native_wind_speed = weather_data.last_wind_speed
|
|
|
|
@callback
|
|
def _handle_coordinator_update(self) -> None:
|
|
"""Handle update of the data."""
|
|
self._process_data()
|
|
|
|
if not self.available:
|
|
self.async_write_ha_state()
|
|
return
|
|
|
|
write_ha_state = False
|
|
for key, updater in (
|
|
(CONF_APPARENT_TEMPERATURE, self._update_apparent_temperature),
|
|
(CONF_CLOUD_COVERAGE, self._update_coverage),
|
|
(CONF_CONDITION, self._update_condition),
|
|
(CONF_DEW_POINT, self._update_dew_point),
|
|
(CONF_HUMIDITY, self._update_humidity),
|
|
(CONF_OZONE, self._update_ozone),
|
|
(CONF_PRESSURE, self._update_pressure),
|
|
(CONF_TEMPERATURE, self._update_temperature),
|
|
(CONF_UV_INDEX, self._update_uv_index),
|
|
(CONF_VISIBILITY, self._update_visibility),
|
|
(CONF_WIND_BEARING, self._update_wind_bearing),
|
|
(CONF_WIND_GUST_SPEED, self._update_wind_gust_speed),
|
|
(CONF_WIND_SPEED, self._update_wind_speed),
|
|
):
|
|
if (rendered := self._rendered.get(key)) is not None:
|
|
updater(rendered)
|
|
write_ha_state = True
|
|
|
|
if write_ha_state:
|
|
self.async_write_ha_state()
|
|
|
|
async def async_forecast_daily(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return (
|
|
self._validate_forecast("daily", self._rendered.get(CONF_FORECAST_DAILY))
|
|
or []
|
|
)
|
|
|
|
async def async_forecast_hourly(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return (
|
|
self._validate_forecast("hourly", self._rendered.get(CONF_FORECAST_HOURLY))
|
|
or []
|
|
)
|
|
|
|
async def async_forecast_twice_daily(self) -> list[Forecast]:
|
|
"""Return the daily forecast in native units."""
|
|
return (
|
|
self._validate_forecast(
|
|
"twice_daily", self._rendered.get(CONF_FORECAST_TWICE_DAILY)
|
|
)
|
|
or []
|
|
)
|
|
|
|
@property
|
|
def extra_restore_state_data(self) -> WeatherExtraStoredData:
|
|
"""Return weather specific state data to be restored."""
|
|
return WeatherExtraStoredData(
|
|
last_apparent_temperature=self._rendered.get(CONF_APPARENT_TEMPERATURE),
|
|
last_cloud_coverage=self._rendered.get(CONF_CLOUD_COVERAGE),
|
|
last_dew_point=self._rendered.get(CONF_DEW_POINT),
|
|
last_humidity=self._rendered.get(CONF_HUMIDITY),
|
|
last_ozone=self._rendered.get(CONF_OZONE),
|
|
last_pressure=self._rendered.get(CONF_PRESSURE),
|
|
last_temperature=self._rendered.get(CONF_TEMPERATURE),
|
|
last_uv_index=self._rendered.get(CONF_UV_INDEX),
|
|
last_visibility=self._rendered.get(CONF_VISIBILITY),
|
|
last_wind_bearing=self._rendered.get(CONF_WIND_BEARING),
|
|
last_wind_gust_speed=self._rendered.get(CONF_WIND_GUST_SPEED),
|
|
last_wind_speed=self._rendered.get(CONF_WIND_SPEED),
|
|
)
|
|
|
|
async def async_get_last_weather_data(self) -> WeatherExtraStoredData | None:
|
|
"""Restore weather specific state data."""
|
|
if (restored_last_extra_data := await self.async_get_last_extra_data()) is None:
|
|
return None
|
|
return WeatherExtraStoredData.from_dict(restored_last_extra_data.as_dict())
|