jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add new Forecasting to Weather (#75219)

Browse Source 
* Add new Forecasting to Weather

* Add is_daytime for forecast_twice_daily

* Fix test

* Fix demo test

* Adjust tests

* Fix typing

* Add demo

* Mod demo more realistic

* Fix test

* Remove one weather

* Fix weather example

* kitchen_sink

* Reverse demo partially

* mod kitchen sink

* Fix twice_daily

* kitchen_sink

* Add test weathers

* Add twice daily to demo

* dt_util

* Fix names

* Expose forecast via WS instead of as state attributes

* Regularly update demo + kitchen_sink weather forecasts

* Run linters

* Fix rebase mistake

* Improve demo test coverage

* Improve weather test coverage

* Exclude kitchen_sink weather from test coverage

* Rename async_update_forecast to async_update_listeners

* Add async_has_listeners helper

* Revert "Add async_has_listeners helper"

This reverts commit 52af3664bb06d9feac2c5ff963ee0022077c23ba.

* Fix rebase mistake

---------

Co-authored-by: Erik <erik@montnemery.com>
This commit is contained in:
G Johansson 2023-07-21 17:30:48 +02:00 committed by GitHub
parent 9f98a418cd
commit 4e30056830
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 1347 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -596,6 +596,7 @@ omit =
homeassistant/components/keymitt_ble/entity.py
homeassistant/components/keymitt_ble/switch.py
homeassistant/components/keymitt_ble/coordinator.py
homeassistant/components/kitchen_sink/weather.py
homeassistant/components/kiwi/lock.py
homeassistant/components/kodi/__init__.py
homeassistant/components/kodi/browse_media.py

1
homeassistant/components/demo/__init__.py
View File

@ -56,6 +56,7 @@ COMPONENTS_WITH_DEMO_PLATFORM = [
Platform.IMAGE_PROCESSING,
Platform.CALENDAR,
Platform.DEVICE_TRACKER,
Platform.WEATHER,
]
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)

117
homeassistant/components/demo/weather.py
View File

@ -1,7 +1,7 @@
"""Demo platform that offers fake meteorological data."""
from __future__ import annotations
from datetime import timedelta
from datetime import datetime, timedelta
from homeassistant.components.weather import (
ATTR_CONDITION_CLOUDY,
@ -20,11 +20,13 @@ from homeassistant.components.weather import (
ATTR_CONDITION_WINDY_VARIANT,
Forecast,
WeatherEntity,
WeatherEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfPressure, UnitOfSpeed, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.dt as dt_util
@ -45,6 +47,8 @@ CONDITION_CLASSES: dict[str, list[str]] = {
ATTR_CONDITION_EXCEPTIONAL: [],
}
WEATHER_UPDATE_INTERVAL = timedelta(minutes=30)
async def async_setup_entry(
hass: HomeAssistant,
@ -83,6 +87,8 @@ def setup_platform(
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100],
],
None,
None,
),
DemoWeather(
"North",
@ -103,6 +109,24 @@ def setup_platform(
[ATTR_CONDITION_SUNNY, 0.3, -14, -19, 0],
[ATTR_CONDITION_SUNNY, 0, -9, -12, 0],
],
[
[ATTR_CONDITION_SUNNY, 2, -10, -15, 60],
[ATTR_CONDITION_SUNNY, 1, -13, -14, 25],
[ATTR_CONDITION_SUNNY, 0, -18, -22, 70],
[ATTR_CONDITION_SUNNY, 0.1, -23, -23, 90],
[ATTR_CONDITION_SUNNY, 4, -19, -20, 40],
[ATTR_CONDITION_SUNNY, 0.3, -14, -19, 0],
[ATTR_CONDITION_SUNNY, 0, -9, -12, 0],
],
[
[ATTR_CONDITION_SNOWY, 2, -10, -15, 60, True],
[ATTR_CONDITION_PARTLYCLOUDY, 1, -13, -14, 25, False],
[ATTR_CONDITION_SUNNY, 0, -18, -22, 70, True],
[ATTR_CONDITION_SUNNY, 0.1, -23, -23, 90, False],
[ATTR_CONDITION_SNOWY, 4, -19, -20, 40, True],
[ATTR_CONDITION_SUNNY, 0.3, -14, -19, 0, False],
[ATTR_CONDITION_SUNNY, 0, -9, -12, 0, True],
],
),
]
)
@ -125,10 +149,13 @@ class DemoWeather(WeatherEntity):
temperature_unit: str,
pressure_unit: str,
wind_speed_unit: str,
forecast: list[list],
forecast_daily: list[list] | None,
forecast_hourly: list[list] | None,
forecast_twice_daily: list[list] | None,
) -> None:
"""Initialize the Demo weather."""
self._attr_name = f"Demo Weather {name}"
self._attr_unique_id = f"demo-weather-{name.lower()}"
self._condition = condition
self._native_temperature = temperature
self._native_temperature_unit = temperature_unit
@ -137,7 +164,40 @@ class DemoWeather(WeatherEntity):
self._native_pressure_unit = pressure_unit
self._native_wind_speed = wind_speed
self._native_wind_speed_unit = wind_speed_unit
self._forecast = forecast
self._forecast_daily = forecast_daily
self._forecast_hourly = forecast_hourly
self._forecast_twice_daily = forecast_twice_daily
self._attr_supported_features = 0
if self._forecast_daily:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_DAILY
if self._forecast_hourly:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_HOURLY
if self._forecast_twice_daily:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_TWICE_DAILY
async def async_added_to_hass(self) -> None:
"""Set up a timer updating the forecasts."""
async def update_forecasts(_: datetime) -> None:
if self._forecast_daily:
self._forecast_daily = (
self._forecast_daily[1:] + self._forecast_daily[:1]
)
if self._forecast_hourly:
self._forecast_hourly = (
self._forecast_hourly[1:] + self._forecast_hourly[:1]
)
if self._forecast_twice_daily:
self._forecast_twice_daily = (
self._forecast_twice_daily[1:] + self._forecast_twice_daily[:1]
)
await self.async_update_listeners(None)
self.async_on_remove(
async_track_time_interval(
self.hass, update_forecasts, WEATHER_UPDATE_INTERVAL
)
)
@property
def native_temperature(self) -> float:
@ -181,13 +241,13 @@ class DemoWeather(WeatherEntity):
k for k, v in CONDITION_CLASSES.items() if self._condition.lower() in v
][0]
@property
def forecast(self) -> list[Forecast]:
"""Return the forecast."""
async def async_forecast_daily(self) -> list[Forecast]:
"""Return the daily forecast."""
reftime = dt_util.now().replace(hour=16, minute=00)
forecast_data = []
for entry in self._forecast:
assert self._forecast_daily is not None
for entry in self._forecast_daily:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
@ -196,7 +256,48 @@ class DemoWeather(WeatherEntity):
templow=entry[3],
precipitation_probability=entry[4],
)
reftime = reftime + timedelta(hours=4)
reftime = reftime + timedelta(hours=24)
forecast_data.append(data_dict)
return forecast_data
async def async_forecast_hourly(self) -> list[Forecast]:
"""Return the hourly forecast."""
reftime = dt_util.now().replace(hour=16, minute=00)
forecast_data = []
assert self._forecast_hourly is not None
for entry in self._forecast_hourly:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
)
reftime = reftime + timedelta(hours=1)
forecast_data.append(data_dict)
return forecast_data
async def async_forecast_twice_daily(self) -> list[Forecast]:
"""Return the twice daily forecast."""
reftime = dt_util.now().replace(hour=11, minute=00)
forecast_data = []
assert self._forecast_twice_daily is not None
for entry in self._forecast_twice_daily:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
is_daytime=entry[5],
)
reftime = reftime + timedelta(hours=12)
forecast_data.append(data_dict)
return forecast_data

7
homeassistant/components/kitchen_sink/__init__.py
View File

@ -26,7 +26,12 @@ import homeassistant.util.dt as dt_util
DOMAIN = "kitchen_sink"
COMPONENTS_WITH_DEMO_PLATFORM = [Platform.SENSOR, Platform.LOCK, Platform.IMAGE]
COMPONENTS_WITH_DEMO_PLATFORM = [
Platform.SENSOR,
Platform.LOCK,
Platform.IMAGE,
Platform.WEATHER,
]
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)

446
homeassistant/components/kitchen_sink/weather.py Normal file
View File

@ -0,0 +1,446 @@
"""Demo platform that offers fake meteorological data."""
from __future__ import annotations
from datetime import timedelta
from homeassistant.components.weather import (
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,
Forecast,
WeatherEntity,
WeatherEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfPressure, UnitOfSpeed, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
import homeassistant.util.dt as dt_util
CONDITION_CLASSES: dict[str, list[str]] = {
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: ["shower rain"],
ATTR_CONDITION_SNOWY: [],
ATTR_CONDITION_SNOWY_RAINY: [],
ATTR_CONDITION_SUNNY: ["sunshine"],
ATTR_CONDITION_WINDY: [],
ATTR_CONDITION_WINDY_VARIANT: [],
ATTR_CONDITION_EXCEPTIONAL: [],
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Demo config entry."""
async_add_entities(
[
DemoWeather(
"Legacy weather",
"Sunshine",
21.6414,
92,
1099,
0.5,
UnitOfTemperature.CELSIUS,
UnitOfPressure.HPA,
UnitOfSpeed.METERS_PER_SECOND,
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100],
],
None,
None,
None,
),
DemoWeather(
"Legacy + daily weather",
"Sunshine",
21.6414,
92,
1099,
0.5,
UnitOfTemperature.CELSIUS,
UnitOfPressure.HPA,
UnitOfSpeed.METERS_PER_SECOND,
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100],
],
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100],
],
None,
None,
),
DemoWeather(
"Daily + hourly weather",
"Shower rain",
-12,
54,
987,
4.8,
UnitOfTemperature.FAHRENHEIT,
UnitOfPressure.INHG,
UnitOfSpeed.MILES_PER_HOUR,
None,
[
[ATTR_CONDITION_SNOWY, 2, -10, -15, 60],
[ATTR_CONDITION_PARTLYCLOUDY, 1, -13, -14, 25],
[ATTR_CONDITION_SUNNY, 0, -18, -22, 70],
[ATTR_CONDITION_SUNNY, 0.1, -23, -23, 90],
[ATTR_CONDITION_SNOWY, 4, -19, -20, 40],
[ATTR_CONDITION_SUNNY, 0.3, -14, -19, 0],
[ATTR_CONDITION_SUNNY, 0, -9, -12, 0],
],
[
[ATTR_CONDITION_SUNNY, 2, -10, -15, 60],
[ATTR_CONDITION_SUNNY, 1, -13, -14, 25],
[ATTR_CONDITION_SUNNY, 0, -18, -22, 70],
[ATTR_CONDITION_SUNNY, 0.1, -23, -23, 90],
[ATTR_CONDITION_SUNNY, 4, -19, -20, 40],
[ATTR_CONDITION_SUNNY, 0.3, -14, -19, 0],
[ATTR_CONDITION_SUNNY, 0, -9, -12, 0],
],
None,
),
DemoWeather(
"Daily + bi-daily + hourly weather",
"Sunshine",
21.6414,
92,
1099,
0.5,
UnitOfTemperature.CELSIUS,
UnitOfPressure.HPA,
UnitOfSpeed.METERS_PER_SECOND,
None,
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_RAINY, 0, 15, 9, 10],
[ATTR_CONDITION_RAINY, 0, 12, 6, 0],
[ATTR_CONDITION_RAINY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_RAINY, 0.2, 21, 12, 100],
],
[
[ATTR_CONDITION_CLOUDY, 1, 22, 15, 60],
[ATTR_CONDITION_CLOUDY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_CLOUDY, 0, 12, 6, 0],
[ATTR_CONDITION_CLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_CLOUDY, 15, 18, 7, 0],
[ATTR_CONDITION_CLOUDY, 0.2, 21, 12, 100],
],
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60, True],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30, False],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10, True],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0, False],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20, True],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0, False],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100, True],
],
),
DemoWeather(
"Hourly + bi-daily weather",
"Sunshine",
21.6414,
92,
1099,
0.5,
UnitOfTemperature.CELSIUS,
UnitOfPressure.HPA,
UnitOfSpeed.METERS_PER_SECOND,
None,
None,
[
[ATTR_CONDITION_CLOUDY, 1, 22, 15, 60],
[ATTR_CONDITION_CLOUDY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_CLOUDY, 0, 12, 6, 0],
[ATTR_CONDITION_CLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_CLOUDY, 15, 18, 7, 0],
[ATTR_CONDITION_CLOUDY, 0.2, 21, 12, 100],
],
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60, True],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30, False],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10, True],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0, False],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20, True],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0, False],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100, True],
],
),
DemoWeather(
"Daily + broken bi-daily weather",
"Sunshine",
21.6414,
92,
1099,
0.5,
UnitOfTemperature.CELSIUS,
UnitOfPressure.HPA,
UnitOfSpeed.METERS_PER_SECOND,
None,
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_RAINY, 0, 15, 9, 10],
[ATTR_CONDITION_RAINY, 0, 12, 6, 0],
[ATTR_CONDITION_RAINY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_RAINY, 0.2, 21, 12, 100],
],
None,
[
[ATTR_CONDITION_RAINY, 1, 22, 15, 60],
[ATTR_CONDITION_RAINY, 5, 19, 8, 30],
[ATTR_CONDITION_CLOUDY, 0, 15, 9, 10],
[ATTR_CONDITION_SUNNY, 0, 12, 6, 0],
[ATTR_CONDITION_PARTLYCLOUDY, 2, 14, 7, 20],
[ATTR_CONDITION_RAINY, 15, 18, 7, 0],
[ATTR_CONDITION_FOG, 0.2, 21, 12, 100],
],
),
]
)
class DemoWeather(WeatherEntity):
"""Representation of a weather condition."""
_attr_attribution = "Powered by Home Assistant"
_attr_should_poll = False
def __init__(
self,
name: str,
condition: str,
temperature: float,
humidity: float,
pressure: float,
wind_speed: float,
temperature_unit: str,
pressure_unit: str,
wind_speed_unit: str,
forecast: list[list] | None,
forecast_daily: list[list] | None,
forecast_hourly: list[list] | None,
forecast_twice_daily: list[list] | None,
) -> None:
"""Initialize the Demo weather."""
self._attr_name = f"Test Weather {name}"
self._attr_unique_id = f"test-weather-{name.lower()}"
self._condition = condition
self._native_temperature = temperature
self._native_temperature_unit = temperature_unit
self._humidity = humidity
self._native_pressure = pressure
self._native_pressure_unit = pressure_unit
self._native_wind_speed = wind_speed
self._native_wind_speed_unit = wind_speed_unit
self._forecast = forecast
self._forecast_daily = forecast_daily
self._forecast_hourly = forecast_hourly
self._forecast_twice_daily = forecast_twice_daily
self._attr_supported_features = 0
if self._forecast_daily:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_DAILY
if self._forecast_hourly:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_HOURLY
if self._forecast_twice_daily:
self._attr_supported_features |= WeatherEntityFeature.FORECAST_TWICE_DAILY
async def async_added_to_hass(self) -> None:
"""Set up a timer updating the forecasts."""
async def update_forecasts(_) -> None:
if self._forecast_daily:
self._forecast_daily = (
self._forecast_daily[1:] + self._forecast_daily[:1]
)
if self._forecast_hourly:
self._forecast_hourly = (
self._forecast_hourly[1:] + self._forecast_hourly[:1]
)
if self._forecast_twice_daily:
self._forecast_twice_daily = (
self._forecast_twice_daily[1:] + self._forecast_twice_daily[:1]
)
await self.async_update_listeners(None)
self.async_on_remove(
async_track_time_interval(
self.hass, update_forecasts, timedelta(seconds=30)
)
)
@property
def native_temperature(self) -> float:
"""Return the temperature."""
return self._native_temperature
@property
def native_temperature_unit(self) -> str:
"""Return the unit of measurement."""
return self._native_temperature_unit
@property
def humidity(self) -> float:
"""Return the humidity."""
return self._humidity
@property
def native_wind_speed(self) -> float:
"""Return the wind speed."""
return self._native_wind_speed
@property
def native_wind_speed_unit(self) -> str:
"""Return the wind speed."""
return self._native_wind_speed_unit
@property
def native_pressure(self) -> float:
"""Return the pressure."""
return self._native_pressure
@property
def native_pressure_unit(self) -> str:
"""Return the pressure."""
return self._native_pressure_unit
@property
def condition(self) -> str:
"""Return the weather condition."""
return [
k for k, v in CONDITION_CLASSES.items() if self._condition.lower() in v
][0]
@property
def forecast(self) -> list[Forecast]:
"""Return legacy forecast."""
if self._forecast is None:
return []
reftime = dt_util.now().replace(hour=16, minute=00)
forecast_data = []
for entry in self._forecast:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
)
reftime = reftime + timedelta(hours=24)
forecast_data.append(data_dict)
return forecast_data
async def async_forecast_daily(self) -> list[Forecast]:
"""Return the daily forecast."""
if self._forecast_daily is None:
return []
reftime = dt_util.now().replace(hour=16, minute=00)
forecast_data = []
for entry in self._forecast_daily:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
)
reftime = reftime + timedelta(hours=24)
forecast_data.append(data_dict)
return forecast_data
async def async_forecast_hourly(self) -> list[Forecast]:
"""Return the hourly forecast."""
if self._forecast_hourly is None:
return []
reftime = dt_util.now().replace(hour=16, minute=00)
forecast_data = []
for entry in self._forecast_hourly:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
)
reftime = reftime + timedelta(hours=1)
forecast_data.append(data_dict)
return forecast_data
async def async_forecast_twice_daily(self) -> list[Forecast]:
"""Return the twice daily forecast."""
if self._forecast_twice_daily is None:
return []
reftime = dt_util.now().replace(hour=11, minute=00)
forecast_data = []
for entry in self._forecast_twice_daily:
try:
data_dict = Forecast(
datetime=reftime.isoformat(),
condition=entry[0],
precipitation=entry[1],
temperature=entry[2],
templow=entry[3],
precipitation_probability=entry[4],
is_daytime=entry[5],
)
reftime = reftime + timedelta(hours=12)
forecast_data.append(data_dict)
except IndexError:
continue
return forecast_data

465
homeassistant/components/weather/__init__.py
View File

@ -1,12 +1,13 @@
"""Weather component that handles meteorological data for your location."""
from __future__ import annotations
from collections.abc import Callable, Iterable
from contextlib import suppress
from dataclasses import dataclass
from datetime import timedelta
import inspect
import logging
from typing import Any, Final, TypedDict, final
from typing import Any, Final, Literal, TypedDict, final
from typing_extensions import Required
@ -19,7 +20,7 @@ from homeassistant.const import (
UnitOfSpeed,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.config_validation import ( # noqa: F401
PLATFORM_SCHEMA,
PLATFORM_SCHEMA_BASE,
@ -29,7 +30,7 @@ from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.unit_system import US_CUSTOMARY_SYSTEM
from .const import (
from .const import ( # noqa: F401
ATTR_WEATHER_APPARENT_TEMPERATURE,
ATTR_WEATHER_CLOUD_COVERAGE,
ATTR_WEATHER_DEW_POINT,
@ -50,6 +51,7 @@ from .const import (
DOMAIN,
UNIT_CONVERSIONS,
VALID_UNITS,
WeatherEntityFeature,
)
from .websocket_api import async_setup as async_setup_ws_api
@ -72,6 +74,7 @@ ATTR_CONDITION_SUNNY = "sunny"
ATTR_CONDITION_WINDY = "windy"
ATTR_CONDITION_WINDY_VARIANT = "windy-variant"
ATTR_FORECAST = "forecast"
ATTR_FORECAST_IS_DAYTIME: Final = "is_daytime"
ATTR_FORECAST_CONDITION: Final = "condition"
ATTR_FORECAST_HUMIDITY: Final = "humidity"
ATTR_FORECAST_NATIVE_PRECIPITATION: Final = "native_precipitation"
@ -149,6 +152,7 @@ class Forecast(TypedDict, total=False):
wind_speed: None
native_dew_point: float | None
uv_index: float | None
is_daytime: bool | None # Mandatory to use with forecast_twice_daily
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@ -183,6 +187,8 @@ class WeatherEntity(Entity):
entity_description: WeatherEntityDescription
_attr_condition: str | None
# _attr_forecast is deprecated, implement async_forecast_daily,
# async_forecast_hourly or async_forecast_twice daily instead
_attr_forecast: list[Forecast] | None = None
_attr_humidity: float | None = None
_attr_ozone: float | None = None
@ -232,6 +238,11 @@ class WeatherEntity(Entity):
_attr_native_wind_speed_unit: str | None = None
_attr_native_dew_point: float | None = None
_forecast_listeners: dict[
Literal["daily", "hourly", "twice_daily"],
list[Callable[[list[dict[str, Any]] | None], None]],
]
_weather_option_temperature_unit: str | None = None
_weather_option_pressure_unit: str | None = None
_weather_option_visibility_unit: str | None = None
@ -263,6 +274,8 @@ class WeatherEntity(Entity):
"visibility_unit",
"_attr_precipitation_unit",
"precipitation_unit",
"_attr_forecast",
"forecast",
)
):
if _reported is False:
@ -291,8 +304,9 @@ class WeatherEntity(Entity):
)
async def async_internal_added_to_hass(self) -> None:
"""Call when the sensor entity is added to hass."""
"""Call when the weather entity is added to hass."""
await super().async_internal_added_to_hass()
self._forecast_listeners = {"daily": [], "hourly": [], "twice_daily": []}
if not self.registry_entry:
return
self.async_registry_entry_updated()
@ -571,9 +585,24 @@ class WeatherEntity(Entity):
@property
def forecast(self) -> list[Forecast] | None:
"""Return the forecast in native units."""
"""Return the forecast in native units.
Should not be overridden by integrations. Kept for backwards compatibility.
"""
return self._attr_forecast
async def async_forecast_daily(self) -> list[Forecast] | None:
"""Return the daily forecast in native units."""
raise NotImplementedError
async def async_forecast_twice_daily(self) -> list[Forecast] | None:
"""Return the daily forecast in native units."""
raise NotImplementedError
async def async_forecast_hourly(self) -> list[Forecast] | None:
"""Return the hourly forecast in native units."""
raise NotImplementedError
@property
def native_precipitation_unit(self) -> str | None:
"""Return the native unit of measurement for accumulated precipitation."""
@ -756,197 +785,197 @@ class WeatherEntity(Entity):
data[ATTR_WEATHER_VISIBILITY_UNIT] = self._visibility_unit
data[ATTR_WEATHER_PRECIPITATION_UNIT] = self._precipitation_unit
if self.forecast is not None:
forecast: list[dict[str, Any]] = []
for existing_forecast_entry in self.forecast:
forecast_entry: dict[str, Any] = dict(existing_forecast_entry)
temperature = forecast_entry.pop(
ATTR_FORECAST_NATIVE_TEMP, forecast_entry.get(ATTR_FORECAST_TEMP)
)
from_temp_unit = (
self.native_temperature_unit or self._default_temperature_unit
)
to_temp_unit = self._temperature_unit
if temperature is None:
forecast_entry[ATTR_FORECAST_TEMP] = None
else:
with suppress(TypeError, ValueError):
temperature_f = float(temperature)
value_temp = UNIT_CONVERSIONS[ATTR_WEATHER_TEMPERATURE_UNIT](
temperature_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_TEMP] = round_temperature(
value_temp, precision
)
if (
forecast_apparent_temp := forecast_entry.pop(
ATTR_FORECAST_NATIVE_APPARENT_TEMP,
forecast_entry.get(ATTR_FORECAST_NATIVE_APPARENT_TEMP),
)
) is not None:
with suppress(TypeError, ValueError):
forecast_apparent_temp = float(forecast_apparent_temp)
value_apparent_temp = UNIT_CONVERSIONS[
ATTR_WEATHER_TEMPERATURE_UNIT
](
forecast_apparent_temp,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_APPARENT_TEMP] = round_temperature(
value_apparent_temp, precision
)
if (
forecast_temp_low := forecast_entry.pop(
ATTR_FORECAST_NATIVE_TEMP_LOW,
forecast_entry.get(ATTR_FORECAST_TEMP_LOW),
)
) is not None:
with suppress(TypeError, ValueError):
forecast_temp_low_f = float(forecast_temp_low)
value_temp_low = UNIT_CONVERSIONS[
ATTR_WEATHER_TEMPERATURE_UNIT
](
forecast_temp_low_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_TEMP_LOW] = round_temperature(
value_temp_low, precision
)
if (
forecast_dew_point := forecast_entry.pop(
ATTR_FORECAST_NATIVE_DEW_POINT,
None,
)
) is not None:
with suppress(TypeError, ValueError):
forecast_dew_point_f = float(forecast_dew_point)
value_dew_point = UNIT_CONVERSIONS[
ATTR_WEATHER_TEMPERATURE_UNIT
](
forecast_dew_point_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_DEW_POINT] = round_temperature(
value_dew_point, precision
)
if (
forecast_pressure := forecast_entry.pop(
ATTR_FORECAST_NATIVE_PRESSURE,
forecast_entry.get(ATTR_FORECAST_PRESSURE),
)
) is not None:
from_pressure_unit = (
self.native_pressure_unit or self._default_pressure_unit
)
to_pressure_unit = self._pressure_unit
with suppress(TypeError, ValueError):
forecast_pressure_f = float(forecast_pressure)
forecast_entry[ATTR_FORECAST_PRESSURE] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_PRESSURE_UNIT](
forecast_pressure_f,
from_pressure_unit,
to_pressure_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_wind_gust_speed := forecast_entry.pop(
ATTR_FORECAST_NATIVE_WIND_GUST_SPEED,
None,
)
) is not None:
from_wind_speed_unit = (
self.native_wind_speed_unit or self._default_wind_speed_unit
)
to_wind_speed_unit = self._wind_speed_unit
with suppress(TypeError, ValueError):
forecast_wind_gust_speed_f = float(forecast_wind_gust_speed)
forecast_entry[ATTR_FORECAST_WIND_GUST_SPEED] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_WIND_SPEED_UNIT](
forecast_wind_gust_speed_f,
from_wind_speed_unit,
to_wind_speed_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_wind_speed := forecast_entry.pop(
ATTR_FORECAST_NATIVE_WIND_SPEED,
forecast_entry.get(ATTR_FORECAST_WIND_SPEED),
)
) is not None:
from_wind_speed_unit = (
self.native_wind_speed_unit or self._default_wind_speed_unit
)
to_wind_speed_unit = self._wind_speed_unit
with suppress(TypeError, ValueError):
forecast_wind_speed_f = float(forecast_wind_speed)
forecast_entry[ATTR_FORECAST_WIND_SPEED] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_WIND_SPEED_UNIT](
forecast_wind_speed_f,
from_wind_speed_unit,
to_wind_speed_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_precipitation := forecast_entry.pop(
ATTR_FORECAST_NATIVE_PRECIPITATION,
forecast_entry.get(ATTR_FORECAST_PRECIPITATION),
)
) is not None:
from_precipitation_unit = (
self.native_precipitation_unit
or self._default_precipitation_unit
)
to_precipitation_unit = self._precipitation_unit
with suppress(TypeError, ValueError):
forecast_precipitation_f = float(forecast_precipitation)
forecast_entry[ATTR_FORECAST_PRECIPITATION] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_PRECIPITATION_UNIT](
forecast_precipitation_f,
from_precipitation_unit,
to_precipitation_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_humidity := forecast_entry.pop(
ATTR_FORECAST_HUMIDITY,
None,
)
) is not None:
with suppress(TypeError, ValueError):
forecast_humidity_f = float(forecast_humidity)
forecast_entry[ATTR_FORECAST_HUMIDITY] = round(
forecast_humidity_f
)
forecast.append(forecast_entry)
data[ATTR_FORECAST] = forecast
if self.forecast:
data[ATTR_FORECAST] = self._convert_forecast(self.forecast)
return data
@final
def _convert_forecast(
self, native_forecast_list: list[Forecast]
) -> list[dict[str, Any]]:
"""Convert a forecast in native units to the unit configured by the user."""
converted_forecast_list: list[dict[str, Any]] = []
precision = self.precision
from_temp_unit = self.native_temperature_unit or self._default_temperature_unit
to_temp_unit = self._temperature_unit
for _forecast_entry in native_forecast_list:
forecast_entry: dict[str, Any] = dict(_forecast_entry)
temperature = forecast_entry.pop(
ATTR_FORECAST_NATIVE_TEMP, forecast_entry.get(ATTR_FORECAST_TEMP)
)
if temperature is None:
forecast_entry[ATTR_FORECAST_TEMP] = None
else:
with suppress(TypeError, ValueError):
temperature_f = float(temperature)
value_temp = UNIT_CONVERSIONS[ATTR_WEATHER_TEMPERATURE_UNIT](
temperature_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_TEMP] = round_temperature(
value_temp, precision
)
if (
forecast_apparent_temp := forecast_entry.pop(
ATTR_FORECAST_NATIVE_APPARENT_TEMP,
forecast_entry.get(ATTR_FORECAST_NATIVE_APPARENT_TEMP),
)
) is not None:
with suppress(TypeError, ValueError):
forecast_apparent_temp = float(forecast_apparent_temp)
value_apparent_temp = UNIT_CONVERSIONS[
ATTR_WEATHER_TEMPERATURE_UNIT
](
forecast_apparent_temp,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_APPARENT_TEMP] = round_temperature(
value_apparent_temp, precision
)
if (
forecast_temp_low := forecast_entry.pop(
ATTR_FORECAST_NATIVE_TEMP_LOW,
forecast_entry.get(ATTR_FORECAST_TEMP_LOW),
)
) is not None:
with suppress(TypeError, ValueError):
forecast_temp_low_f = float(forecast_temp_low)
value_temp_low = UNIT_CONVERSIONS[ATTR_WEATHER_TEMPERATURE_UNIT](
forecast_temp_low_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_TEMP_LOW] = round_temperature(
value_temp_low, precision
)
if (
forecast_dew_point := forecast_entry.pop(
ATTR_FORECAST_NATIVE_DEW_POINT,
None,
)
) is not None:
with suppress(TypeError, ValueError):
forecast_dew_point_f = float(forecast_dew_point)
value_dew_point = UNIT_CONVERSIONS[ATTR_WEATHER_TEMPERATURE_UNIT](
forecast_dew_point_f,
from_temp_unit,
to_temp_unit,
)
forecast_entry[ATTR_FORECAST_DEW_POINT] = round_temperature(
value_dew_point, precision
)
if (
forecast_pressure := forecast_entry.pop(
ATTR_FORECAST_NATIVE_PRESSURE,
forecast_entry.get(ATTR_FORECAST_PRESSURE),
)
) is not None:
from_pressure_unit = (
self.native_pressure_unit or self._default_pressure_unit
)
to_pressure_unit = self._pressure_unit
with suppress(TypeError, ValueError):
forecast_pressure_f = float(forecast_pressure)
forecast_entry[ATTR_FORECAST_PRESSURE] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_PRESSURE_UNIT](
forecast_pressure_f,
from_pressure_unit,
to_pressure_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_wind_gust_speed := forecast_entry.pop(
ATTR_FORECAST_NATIVE_WIND_GUST_SPEED,
None,
)
) is not None:
from_wind_speed_unit = (
self.native_wind_speed_unit or self._default_wind_speed_unit
)
to_wind_speed_unit = self._wind_speed_unit
with suppress(TypeError, ValueError):
forecast_wind_gust_speed_f = float(forecast_wind_gust_speed)
forecast_entry[ATTR_FORECAST_WIND_GUST_SPEED] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_WIND_SPEED_UNIT](
forecast_wind_gust_speed_f,
from_wind_speed_unit,
to_wind_speed_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_wind_speed := forecast_entry.pop(
ATTR_FORECAST_NATIVE_WIND_SPEED,
forecast_entry.get(ATTR_FORECAST_WIND_SPEED),
)
) is not None:
from_wind_speed_unit = (
self.native_wind_speed_unit or self._default_wind_speed_unit
)
to_wind_speed_unit = self._wind_speed_unit
with suppress(TypeError, ValueError):
forecast_wind_speed_f = float(forecast_wind_speed)
forecast_entry[ATTR_FORECAST_WIND_SPEED] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_WIND_SPEED_UNIT](
forecast_wind_speed_f,
from_wind_speed_unit,
to_wind_speed_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_precipitation := forecast_entry.pop(
ATTR_FORECAST_NATIVE_PRECIPITATION,
forecast_entry.get(ATTR_FORECAST_PRECIPITATION),
)
) is not None:
from_precipitation_unit = (
self.native_precipitation_unit or self._default_precipitation_unit
)
to_precipitation_unit = self._precipitation_unit
with suppress(TypeError, ValueError):
forecast_precipitation_f = float(forecast_precipitation)
forecast_entry[ATTR_FORECAST_PRECIPITATION] = round(
UNIT_CONVERSIONS[ATTR_WEATHER_PRECIPITATION_UNIT](
forecast_precipitation_f,
from_precipitation_unit,
to_precipitation_unit,
),
ROUNDING_PRECISION,
)
if (
forecast_humidity := forecast_entry.pop(
ATTR_FORECAST_HUMIDITY,
None,
)
) is not None:
with suppress(TypeError, ValueError):
forecast_humidity_f = float(forecast_humidity)
forecast_entry[ATTR_FORECAST_HUMIDITY] = round(forecast_humidity_f)
converted_forecast_list.append(forecast_entry)
return converted_forecast_list
@property
@final
def state(self) -> str | None:
@ -998,3 +1027,53 @@ class WeatherEntity(Entity):
)
) and custom_unit_visibility in VALID_UNITS[ATTR_WEATHER_VISIBILITY_UNIT]:
self._weather_option_visibility_unit = custom_unit_visibility
@final
@callback
def async_subscribe_forecast(
self,
forecast_type: Literal["daily", "hourly", "twice_daily"],
forecast_listener: Callable[[list[dict[str, Any]] | None], None],
) -> CALLBACK_TYPE:
"""Subscribe to forecast updates.
Called by websocket API.
"""
self._forecast_listeners[forecast_type].append(forecast_listener)
@callback
def unsubscribe() -> None:
self._forecast_listeners[forecast_type].remove(forecast_listener)
return unsubscribe
@final
async def async_update_listeners(
self, forecast_types: Iterable[Literal["daily", "hourly", "twice_daily"]] | None
) -> None:
"""Push updated forecast to all listeners."""
if forecast_types is None:
forecast_types = {"daily", "hourly", "twice_daily"}
for forecast_type in forecast_types:
if not self._forecast_listeners[forecast_type]:
continue
native_forecast_list: list[Forecast] | None = await getattr(
self, f"async_forecast_{forecast_type}"
)()
if native_forecast_list is None:
for listener in self._forecast_listeners[forecast_type]:
listener(None)
continue
if forecast_type == "twice_daily":
for fc_twice_daily in native_forecast_list:
if fc_twice_daily.get(ATTR_FORECAST_IS_DAYTIME) is None:
raise ValueError(
"is_daytime mandatory attribute for forecast_twice_daily is missing"
)
converted_forecast_list = self._convert_forecast(native_forecast_list)
for listener in self._forecast_listeners[forecast_type]:
listener(converted_forecast_list)

10
homeassistant/components/weather/const.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Callable
from enum import IntFlag
from typing import Final
from homeassistant.const import (
@ -18,6 +19,15 @@ from homeassistant.util.unit_conversion import (
TemperatureConverter,
)
class WeatherEntityFeature(IntFlag):
"""Supported features of the update entity."""
FORECAST_DAILY = 1
FORECAST_HOURLY = 2
FORECAST_TWICE_DAILY = 4
ATTR_WEATHER_HUMIDITY = "humidity"
ATTR_WEATHER_OZONE = "ozone"
ATTR_WEATHER_DEW_POINT = "dew_point"

72
homeassistant/components/weather/websocket_api.py
View File

@ -1,20 +1,29 @@
"""The weather websocket API."""
from __future__ import annotations
from typing import Any
from typing import Any, Literal
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_component import EntityComponent
from .const import VALID_UNITS
from .const import DOMAIN, VALID_UNITS, WeatherEntityFeature
FORECAST_TYPE_TO_FLAG = {
"daily": WeatherEntityFeature.FORECAST_DAILY,
"hourly": WeatherEntityFeature.FORECAST_HOURLY,
"twice_daily": WeatherEntityFeature.FORECAST_TWICE_DAILY,
}
@callback
def async_setup(hass: HomeAssistant) -> None:
"""Set up the weather websocket API."""
websocket_api.async_register_command(hass, ws_convertible_units)
websocket_api.async_register_command(hass, ws_subscribe_forecast)
@callback
@ -31,3 +40,62 @@ def ws_convertible_units(
key: sorted(units, key=str.casefold) for key, units in VALID_UNITS.items()
}
connection.send_result(msg["id"], {"units": sorted_units})
@websocket_api.websocket_command(
{
vol.Required("type"): "weather/subscribe_forecast",
vol.Required("entity_id"): cv.entity_domain(DOMAIN),
vol.Required("forecast_type"): vol.In(["daily", "hourly", "twice_daily"]),
}
)
@websocket_api.async_response
async def ws_subscribe_forecast(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Subscribe to weather forecasts."""
from . import WeatherEntity # pylint: disable=import-outside-toplevel
component: EntityComponent[WeatherEntity] = hass.data[DOMAIN]
entity_id: str = msg["entity_id"]
forecast_type: Literal["daily", "hourly", "twice_daily"] = msg["forecast_type"]
if not (entity := component.get_entity(msg["entity_id"])):
connection.send_error(
msg["id"],
"invalid_entity_id",
f"Weather entity not found: {entity_id}",
)
return
if (
entity.supported_features is None
or not entity.supported_features & FORECAST_TYPE_TO_FLAG[forecast_type]
):
connection.send_error(
msg["id"],
"forecast_not_supported",
f"The weather entity does not support forecast type: {forecast_type}",
)
return
@callback
def forecast_listener(forecast: list[dict[str, Any]] | None) -> None:
"""Push a new forecast to websocket."""
connection.send_message(
websocket_api.event_message(
msg["id"],
{
"type": forecast_type,
"forecast": forecast,
},
)
)
connection.subscriptions[msg["id"]] = entity.async_subscribe_forecast(
forecast_type, forecast_listener
)
connection.send_message(websocket_api.result_message(msg["id"]))
# Push an initial forecast update
await entity.async_update_listeners({forecast_type})

143
tests/components/demo/test_weather.py
View File

@ -1,12 +1,13 @@
"""The tests for the demo weather component."""
import datetime
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import weather
from homeassistant.components.demo.weather import WEATHER_UPDATE_INTERVAL
from homeassistant.components.weather import (
ATTR_FORECAST,
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_WEATHER_HUMIDITY,
ATTR_WEATHER_OZONE,
ATTR_WEATHER_PRESSURE,
@ -19,6 +20,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.unit_system import METRIC_SYSTEM
from tests.typing import WebSocketGenerator
async def test_attributes(hass: HomeAssistant, disable_platforms) -> None:
"""Test weather attributes."""
@ -41,16 +44,120 @@ async def test_attributes(hass: HomeAssistant, disable_platforms) -> None:
assert data.get(ATTR_WEATHER_WIND_BEARING) is None
assert data.get(ATTR_WEATHER_OZONE) is None
assert data.get(ATTR_ATTRIBUTION) == "Powered by Home Assistant"
assert data.get(ATTR_FORECAST)[0].get(ATTR_FORECAST_CONDITION) == "rainy"
assert data.get(ATTR_FORECAST)[0].get(ATTR_FORECAST_PRECIPITATION) == 1
assert data.get(ATTR_FORECAST)[0].get(ATTR_FORECAST_PRECIPITATION_PROBABILITY) == 60
assert data.get(ATTR_FORECAST)[0].get(ATTR_FORECAST_TEMP) == 22
assert data.get(ATTR_FORECAST)[0].get(ATTR_FORECAST_TEMP_LOW) == 15
assert data.get(ATTR_FORECAST)[6].get(ATTR_FORECAST_CONDITION) == "fog"
assert data.get(ATTR_FORECAST)[6].get(ATTR_FORECAST_PRECIPITATION) == 0.2
assert data.get(ATTR_FORECAST)[6].get(ATTR_FORECAST_TEMP) == 21
assert data.get(ATTR_FORECAST)[6].get(ATTR_FORECAST_TEMP_LOW) == 12
assert (
data.get(ATTR_FORECAST)[6].get(ATTR_FORECAST_PRECIPITATION_PROBABILITY) == 100
TEST_TIME_ADVANCE_INTERVAL = datetime.timedelta(seconds=5 + 1)
@pytest.mark.parametrize(
("forecast_type", "expected_forecast"),
[
(
"daily",
[
{
"condition": "snowy",
"precipitation": 2.0,
"temperature": -23.3,
"templow": -26.1,
"precipitation_probability": 60,
},
{
"condition": "sunny",
"precipitation": 0.0,
"temperature": -22.8,
"templow": -24.4,
"precipitation_probability": 0,
},
],
),
(
"hourly",
[
{
"condition": "sunny",
"precipitation": 2.0,
"temperature": -23.3,
"templow": -26.1,
"precipitation_probability": 60,
},
{
"condition": "sunny",
"precipitation": 0.0,
"temperature": -22.8,
"templow": -24.4,
"precipitation_probability": 0,
},
],
),
(
"twice_daily",
[
{
"condition": "snowy",
"precipitation": 2.0,
"temperature": -23.3,
"templow": -26.1,
"precipitation_probability": 60,
},
{
"condition": "sunny",
"precipitation": 0.0,
"temperature": -22.8,
"templow": -24.4,
"precipitation_probability": 0,
},
],
),
],
)
async def test_forecast(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
freezer: FrozenDateTimeFactory,
disable_platforms: None,
forecast_type: str,
expected_forecast: list[dict[str, Any]],
) -> None:
"""Test multiple forecast."""
assert await async_setup_component(
hass, weather.DOMAIN, {"weather": {"platform": "demo"}}
)
assert len(data.get(ATTR_FORECAST)) == 7
hass.config.units = METRIC_SYSTEM
await hass.async_block_till_done()
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": forecast_type,
"entity_id": "weather.demo_weather_north",
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
subscription_id = msg["id"]
msg = await client.receive_json()
assert msg["id"] == subscription_id
assert msg["type"] == "event"
forecast1 = msg["event"]["forecast"]
assert len(forecast1) == 7
for key, val in expected_forecast[0].items():
assert forecast1[0][key] == val
for key, val in expected_forecast[1].items():
assert forecast1[6][key] == val
freezer.tick(WEATHER_UPDATE_INTERVAL + datetime.timedelta(seconds=1))
await hass.async_block_till_done()
msg = await client.receive_json()
assert msg["id"] == subscription_id
assert msg["type"] == "event"
forecast2 = msg["event"]["forecast"]
assert forecast2 != forecast1
assert len(forecast2) == 7

31
tests/components/weather/__init__.py
View File

@ -1 +1,32 @@
"""The tests for Weather platforms."""
from homeassistant.components.weather import ATTR_CONDITION_SUNNY
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.testing_config.custom_components.test import weather as WeatherPlatform
async def create_entity(hass: HomeAssistant, **kwargs):
"""Create the weather entity to run tests on."""
kwargs = {
"native_temperature": None,
"native_temperature_unit": None,
"is_daytime": True,
**kwargs,
}
platform: WeatherPlatform = getattr(hass.components, "test.weather")
platform.init(empty=True)
platform.ENTITIES.append(
platform.MockWeatherMockForecast(
name="Test", condition=ATTR_CONDITION_SUNNY, **kwargs
)
)
entity0 = platform.ENTITIES[0]
assert await async_setup_component(
hass, "weather", {"weather": {"platform": "test"}}
)
await hass.async_block_till_done()
return entity0

90
tests/components/weather/test_init.py
View File

@ -34,6 +34,7 @@ from homeassistant.components.weather import (
ROUNDING_PRECISION,
Forecast,
WeatherEntity,
WeatherEntityFeature,
round_temperature,
)
from homeassistant.components.weather.const import (
@ -54,6 +55,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_conversion import (
DistanceConverter,
PressureConverter,
@ -62,7 +64,10 @@ from homeassistant.util.unit_conversion import (
)
from homeassistant.util.unit_system import METRIC_SYSTEM, US_CUSTOMARY_SYSTEM
from . import create_entity
from tests.testing_config.custom_components.test import weather as WeatherPlatform
from tests.typing import WebSocketGenerator
class MockWeatherEntity(WeatherEntity):
@ -86,12 +91,19 @@ class MockWeatherEntity(WeatherEntity):
self._attr_native_wind_speed_unit = UnitOfSpeed.METERS_PER_SECOND
self._attr_forecast = [
Forecast(
datetime=datetime(2022, 6, 20, 20, 00, 00),
datetime=datetime(2022, 6, 20, 00, 00, 00, tzinfo=dt_util.UTC),
native_precipitation=1,
native_temperature=20,
native_dew_point=2,
)
]
self._attr_forecast_twice_daily = [
Forecast(
datetime=datetime(2022, 6, 20, 8, 00, 00, tzinfo=dt_util.UTC),
native_precipitation=10,
native_temperature=25,
)
]
class MockWeatherEntityPrecision(WeatherEntity):
@ -126,32 +138,13 @@ class MockWeatherEntityCompat(WeatherEntity):
self._attr_wind_speed_unit = UnitOfSpeed.METERS_PER_SECOND
self._attr_forecast = [
Forecast(
datetime=datetime(2022, 6, 20, 20, 00, 00),
datetime=datetime(2022, 6, 20, 0, 00, 00, tzinfo=dt_util.UTC),
precipitation=1,
temperature=20,
)
]
async def create_entity(hass: HomeAssistant, **kwargs):
"""Create the weather entity to run tests on."""
kwargs = {"native_temperature": None, "native_temperature_unit": None, **kwargs}
platform: WeatherPlatform = getattr(hass.components, "test.weather")
platform.init(empty=True)
platform.ENTITIES.append(
platform.MockWeatherMockForecast(
name="Test", condition=ATTR_CONDITION_SUNNY, **kwargs
)
)
entity0 = platform.ENTITIES[0]
assert await async_setup_component(
hass, "weather", {"weather": {"platform": "test"}}
)
await hass.async_block_till_done()
return entity0
@pytest.mark.parametrize(
"native_unit", (UnitOfTemperature.FAHRENHEIT, UnitOfTemperature.CELSIUS)
)
@ -192,7 +185,7 @@ async def test_temperature(
)
state = hass.states.get(entity0.entity_id)
forecast = state.attributes[ATTR_FORECAST][0]
forecast_daily = state.attributes[ATTR_FORECAST][0]
expected = state_value
apparent_expected = apparent_state_value
@ -207,14 +200,20 @@ async def test_temperature(
dew_point_expected, rel=0.1
)
assert state.attributes[ATTR_WEATHER_TEMPERATURE_UNIT] == state_unit
assert float(forecast[ATTR_FORECAST_TEMP]) == pytest.approx(expected, rel=0.1)
assert float(forecast[ATTR_FORECAST_APPARENT_TEMP]) == pytest.approx(
assert float(forecast_daily[ATTR_FORECAST_TEMP]) == pytest.approx(expected, rel=0.1)
assert float(forecast_daily[ATTR_FORECAST_APPARENT_TEMP]) == pytest.approx(
apparent_expected, rel=0.1
)
assert float(forecast[ATTR_FORECAST_DEW_POINT]) == pytest.approx(
assert float(forecast_daily[ATTR_FORECAST_DEW_POINT]) == pytest.approx(
dew_point_expected, rel=0.1
)
assert float(forecast[ATTR_FORECAST_TEMP_LOW]) == pytest.approx(expected, rel=0.1)
assert float(forecast_daily[ATTR_FORECAST_TEMP_LOW]) == pytest.approx(
expected, rel=0.1
)
assert float(forecast_daily[ATTR_FORECAST_TEMP]) == pytest.approx(expected, rel=0.1)
assert float(forecast_daily[ATTR_FORECAST_TEMP_LOW]) == pytest.approx(
expected, rel=0.1
)
@pytest.mark.parametrize("native_unit", (None,))
@ -695,6 +694,7 @@ async def test_custom_units(
native_visibility_unit=visibility_unit,
native_precipitation=precipitation_value,
native_precipitation_unit=precipitation_unit,
is_daytime=True,
unique_id="very_unique",
)
)
@ -1031,7 +1031,7 @@ async def test_attr_compatibility(hass: HomeAssistant) -> None:
forecast_entry = [
Forecast(
datetime=datetime(2022, 6, 20, 20, 00, 00),
datetime=datetime(2022, 6, 20, 0, 00, 00, tzinfo=dt_util.UTC),
precipitation=1,
temperature=20,
)
@ -1067,3 +1067,39 @@ async def test_precision_for_temperature(hass: HomeAssistant) -> None:
assert weather.state_attributes[ATTR_WEATHER_TEMPERATURE] == 20.5
assert weather.state_attributes[ATTR_WEATHER_DEW_POINT] == 2.5
async def test_forecast_twice_daily_missing_is_daytime(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
enable_custom_integrations: None,
) -> None:
"""Test forecast_twice_daily missing mandatory attribute is_daytime."""
entity0 = await create_entity(
hass,
native_temperature=38,
native_temperature_unit=UnitOfTemperature.CELSIUS,
is_daytime=None,
supported_features=WeatherEntityFeature.FORECAST_TWICE_DAILY,
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": "twice_daily",
"entity_id": entity0.entity_id,
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
subscription_id = msg["id"]
msg = await client.receive_json()
assert msg["id"] == subscription_id
assert msg["error"] == {"code": "unknown_error", "message": "Unknown error"}
assert not msg["success"]
assert msg["type"] == "result"

44
tests/components/weather/test_recorder.py
View File

@ -5,7 +5,11 @@ from datetime import timedelta
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.history import get_significant_states
from homeassistant.components.weather import ATTR_FORECAST, DOMAIN
from homeassistant.components.weather import (
ATTR_CONDITION_SUNNY,
ATTR_FORECAST,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
@ -13,17 +17,47 @@ from homeassistant.util.unit_system import METRIC_SYSTEM
from tests.common import async_fire_time_changed
from tests.components.recorder.common import async_wait_recording_done
from tests.testing_config.custom_components.test import weather as WeatherPlatform
async def test_exclude_attributes(recorder_mock: Recorder, hass: HomeAssistant) -> None:
async def create_entity(hass: HomeAssistant, **kwargs):
"""Create the weather entity to run tests on."""
kwargs = {
"native_temperature": None,
"native_temperature_unit": None,
"is_daytime": True,
**kwargs,
}
platform: WeatherPlatform = getattr(hass.components, "test.weather")
platform.init(empty=True)
platform.ENTITIES.append(
platform.MockWeatherMockForecast(
name="Test", condition=ATTR_CONDITION_SUNNY, **kwargs
)
)
entity0 = platform.ENTITIES[0]
assert await async_setup_component(
hass, "weather", {"weather": {"platform": "test"}}
)
await hass.async_block_till_done()
return entity0
async def test_exclude_attributes(
recorder_mock: Recorder, hass: HomeAssistant, enable_custom_integrations: None
) -> None:
"""Test weather attributes to be excluded."""
now = dt_util.utcnow()
await async_setup_component(hass, "homeassistant", {})
await async_setup_component(hass, DOMAIN, {DOMAIN: {"platform": "demo"}})
entity0 = await create_entity(
hass,
native_temperature=38,
native_temperature_unit=UnitOfTemperature.CELSIUS,
)
hass.config.units = METRIC_SYSTEM
await hass.async_block_till_done()
state = hass.states.get("weather.demo_weather_south")
state = hass.states.get(entity0.entity_id)
assert state.attributes[ATTR_FORECAST]
await hass.async_block_till_done()

119
tests/components/weather/test_websocket_api.py
View File

@ -1,8 +1,12 @@
"""Test the weather websocket API."""
from homeassistant.components.weather import WeatherEntityFeature
from homeassistant.components.weather.const import DOMAIN
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from . import create_entity
from tests.typing import WebSocketGenerator
@ -31,3 +35,118 @@ async def test_device_class_units(
"wind_speed_unit": ["ft/s", "km/h", "kn", "m/s", "mph"],
}
}
async def test_subscribe_forecast(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
enable_custom_integrations: None,
) -> None:
"""Test multiple forecast."""
entity0 = await create_entity(
hass,
native_temperature=38,
native_temperature_unit=UnitOfTemperature.CELSIUS,
supported_features=WeatherEntityFeature.FORECAST_DAILY,
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": "daily",
"entity_id": entity0.entity_id,
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
subscription_id = msg["id"]
msg = await client.receive_json()
assert msg["id"] == subscription_id
assert msg["type"] == "event"
forecast = msg["event"]
assert forecast == {
"type": "daily",
"forecast": [
{
"cloud_coverage": None,
"temperature": 38.0,
"templow": 38.0,
"uv_index": None,
"wind_bearing": None,
}
],
}
await entity0.async_update_listeners(None)
msg = await client.receive_json()
assert msg["event"] == forecast
await entity0.async_update_listeners(["daily"])
msg = await client.receive_json()
assert msg["event"] == forecast
entity0.forecast_list = None
await entity0.async_update_listeners(None)
msg = await client.receive_json()
assert msg["event"] == {"type": "daily", "forecast": None}
async def test_subscribe_forecast_unknown_entity(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
enable_custom_integrations: None,
) -> None:
"""Test multiple forecast."""
assert await async_setup_component(hass, DOMAIN, {})
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": "daily",
"entity_id": "weather.unknown",
}
)
msg = await client.receive_json()
assert not msg["success"]
assert msg["error"] == {
"code": "invalid_entity_id",
"message": "Weather entity not found: weather.unknown",
}
async def test_subscribe_forecast_unsupported(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
enable_custom_integrations: None,
) -> None:
"""Test multiple forecast."""
entity0 = await create_entity(
hass,
native_temperature=38,
native_temperature_unit=UnitOfTemperature.CELSIUS,
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": "daily",
"entity_id": entity0.entity_id,
}
)
msg = await client.receive_json()
assert not msg["success"]
assert msg["error"] == {
"code": "forecast_not_supported",
"message": "The weather entity does not support forecast type: daily",
}

55
tests/testing_config/custom_components/test/weather.py
View File

@ -4,9 +4,12 @@ Call init before using it in your tests to ensure clean test data.
"""
from __future__ import annotations
from typing import Any
from homeassistant.components.weather import (
ATTR_FORECAST_CLOUD_COVERAGE,
ATTR_FORECAST_HUMIDITY,
ATTR_FORECAST_IS_DAYTIME,
ATTR_FORECAST_NATIVE_APPARENT_TEMP,
ATTR_FORECAST_NATIVE_DEW_POINT,
ATTR_FORECAST_NATIVE_PRECIPITATION,
@ -220,9 +223,61 @@ class MockWeatherCompat(MockEntity, WeatherEntity):
class MockWeatherMockForecast(MockWeather):
"""Mock weather class with mocked forecast."""
def __init__(self, **values: Any) -> None:
"""Initialize."""
super().__init__(**values)
self.forecast_list: list[Forecast] | None = [
{
ATTR_FORECAST_NATIVE_TEMP: self.native_temperature,
ATTR_FORECAST_NATIVE_APPARENT_TEMP: self.native_apparent_temperature,
ATTR_FORECAST_NATIVE_TEMP_LOW: self.native_temperature,
ATTR_FORECAST_NATIVE_DEW_POINT: self.native_dew_point,
ATTR_FORECAST_CLOUD_COVERAGE: self.cloud_coverage,
ATTR_FORECAST_NATIVE_PRESSURE: self.native_pressure,
ATTR_FORECAST_NATIVE_WIND_GUST_SPEED: self.native_wind_gust_speed,
ATTR_FORECAST_NATIVE_WIND_SPEED: self.native_wind_speed,
ATTR_FORECAST_WIND_BEARING: self.wind_bearing,
ATTR_FORECAST_UV_INDEX: self.uv_index,
ATTR_FORECAST_NATIVE_PRECIPITATION: self._values.get(
"native_precipitation"
),
ATTR_FORECAST_HUMIDITY: self.humidity,
}
]
@property
def forecast(self) -> list[Forecast] | None:
"""Return the forecast."""
return self.forecast_list
async def async_forecast_daily(self) -> list[Forecast] | None:
"""Return the forecast_daily."""
return self.forecast_list
async def async_forecast_twice_daily(self) -> list[Forecast] | None:
"""Return the forecast_twice_daily."""
return [
{
ATTR_FORECAST_NATIVE_TEMP: self.native_temperature,
ATTR_FORECAST_NATIVE_APPARENT_TEMP: self.native_apparent_temperature,
ATTR_FORECAST_NATIVE_TEMP_LOW: self.native_temperature,
ATTR_FORECAST_NATIVE_DEW_POINT: self.native_dew_point,
ATTR_FORECAST_CLOUD_COVERAGE: self.cloud_coverage,
ATTR_FORECAST_NATIVE_PRESSURE: self.native_pressure,
ATTR_FORECAST_NATIVE_WIND_GUST_SPEED: self.native_wind_gust_speed,
ATTR_FORECAST_NATIVE_WIND_SPEED: self.native_wind_speed,
ATTR_FORECAST_WIND_BEARING: self.wind_bearing,
ATTR_FORECAST_UV_INDEX: self.uv_index,
ATTR_FORECAST_NATIVE_PRECIPITATION: self._values.get(
"native_precipitation"
),
ATTR_FORECAST_HUMIDITY: self.humidity,
ATTR_FORECAST_IS_DAYTIME: self._values.get("is_daytime"),
}
]
async def async_forecast_hourly(self) -> list[Forecast] | None:
"""Return the forecast_hourly."""
return [
{
ATTR_FORECAST_NATIVE_TEMP: self.native_temperature,