Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Modernize ipma weather (#98062) 

* Modernize ipma weather

* Add test snapshots

* Don't include forecast mode in weather entity unique_id for new config entries

* Remove old migration code

* Remove outdated test
pull/98304/head
Erik Montnemery 2023-08-12 15:15:09 +02:00 committed by GitHub
parent be9afd7eae
commit ae8f9dcb77
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 282 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/ipma/config_flow.py
View File

@ -2,10 +2,10 @@
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE, CONF_NAME
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
import homeassistant.helpers.config_validation as cv
from .const import DOMAIN, FORECAST_MODE, HOME_LOCATION_NAME
from .const import DOMAIN, HOME_LOCATION_NAME
class IpmaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@ -47,7 +47,6 @@ class IpmaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
vol.Required(CONF_NAME, default=name): str,
vol.Required(CONF_LATITUDE, default=latitude): cv.latitude,
vol.Required(CONF_LONGITUDE, default=longitude): cv.longitude,
vol.Required(CONF_MODE, default="daily"): vol.In(FORECAST_MODE),
}
),
errors=self._errors,

2
homeassistant/components/ipma/const.py
View File

@ -49,6 +49,4 @@ CONDITION_CLASSES = {
ATTR_CONDITION_CLEAR_NIGHT: [-1],
}
FORECAST_MODE = ["hourly", "daily"]
ATTRIBUTION = "Instituto Português do Mar e Atmosfera"

115
homeassistant/components/ipma/weather.py
View File

@ -1,11 +1,14 @@
"""Support for IPMA weather service."""
from __future__ import annotations
import asyncio
import contextlib
import logging
from typing import Literal
import async_timeout
from pyipma.api import IPMA_API
from pyipma.forecast import Forecast
from pyipma.forecast import Forecast as IPMAForecast
from pyipma.location import Location
from homeassistant.components.weather import (
@ -16,7 +19,9 @@ from homeassistant.components.weather import (
ATTR_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
Forecast,
WeatherEntity,
WeatherEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -26,8 +31,7 @@ from homeassistant.const import (
UnitOfSpeed,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.sun import is_up
from homeassistant.util import Throttle
@ -53,39 +57,19 @@ async def async_setup_entry(
"""Add a weather entity from a config_entry."""
api = hass.data[DOMAIN][config_entry.entry_id][DATA_API]
location = hass.data[DOMAIN][config_entry.entry_id][DATA_LOCATION]
mode = config_entry.data[CONF_MODE]
# Migrate old unique_id
@callback
def _async_migrator(entity_entry: er.RegistryEntry):
# Reject if new unique_id
if entity_entry.unique_id.count(",") == 2:
return None
new_unique_id = (
f"{location.station_latitude}, {location.station_longitude}, {mode}"
)
_LOGGER.info(
"Migrating unique_id from [%s] to [%s]",
entity_entry.unique_id,
new_unique_id,
)
return {"new_unique_id": new_unique_id}
await er.async_migrate_entries(hass, config_entry.entry_id, _async_migrator)
async_add_entities([IPMAWeather(location, api, config_entry.data)], True)
class IPMAWeather(WeatherEntity, IPMADevice):
"""Representation of a weather condition."""
_attr_attribution = ATTRIBUTION
_attr_native_pressure_unit = UnitOfPressure.HPA
_attr_native_temperature_unit = UnitOfTemperature.CELSIUS
_attr_native_wind_speed_unit = UnitOfSpeed.KILOMETERS_PER_HOUR
_attr_attribution = ATTRIBUTION
_attr_supported_features = (
WeatherEntityFeature.FORECAST_DAILY | WeatherEntityFeature.FORECAST_HOURLY
)
def __init__(self, location: Location, api: IPMA_API, config) -> None:
"""Initialise the platform with a data instance and station name."""
@ -95,25 +79,35 @@ class IPMAWeather(WeatherEntity, IPMADevice):
self._mode = config.get(CONF_MODE)
self._period = 1 if config.get(CONF_MODE) == "hourly" else 24
self._observation = None
self._forecast: list[Forecast] = []
self._attr_unique_id = f"{self._location.station_latitude}, {self._location.station_longitude}, {self._mode}"
self._daily_forecast: list[IPMAForecast] | None = None
self._hourly_forecast: list[IPMAForecast] | None = None
if self._mode is not None:
self._attr_unique_id = f"{self._location.station_latitude}, {self._location.station_longitude}, {self._mode}"
else:
self._attr_unique_id = (
f"{self._location.station_latitude}, {self._location.station_longitude}"
)
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self) -> None:
"""Update Condition and Forecast."""
async with async_timeout.timeout(10):
new_observation = await self._location.observation(self._api)
new_forecast = await self._location.forecast(self._api, self._period)
if new_observation:
self._observation = new_observation
else:
_LOGGER.warning("Could not update weather observation")
if new_forecast:
self._forecast = new_forecast
if self._period == 24 or self._forecast_listeners["daily"]:
await self._update_forecast("daily", 24, True)
else:
_LOGGER.warning("Could not update weather forecast")
self._daily_forecast = None
if self._period == 1 or self._forecast_listeners["hourly"]:
await self._update_forecast("hourly", 1, True)
else:
self._hourly_forecast = None
_LOGGER.debug(
"Updated location %s based on %s, current observation %s",
@ -122,6 +116,21 @@ class IPMAWeather(WeatherEntity, IPMADevice):
self._observation,
)
async def _update_forecast(
self,
forecast_type: Literal["daily", "hourly"],
period: int,
update_listeners: bool,
) -> None:
"""Update weather forecast."""
new_forecast = await self._location.forecast(self._api, period)
if new_forecast:
setattr(self, f"_{forecast_type}_forecast", new_forecast)
if update_listeners:
await self.async_update_listeners((forecast_type,))
else:
_LOGGER.warning("Could not update %s weather forecast", forecast_type)
def _condition_conversion(self, identifier, forecast_dt):
"""Convert from IPMA weather_type id to HA."""
if identifier == 1 and not is_up(self.hass, forecast_dt):
@ -135,10 +144,12 @@ class IPMAWeather(WeatherEntity, IPMADevice):
@property
def condition(self):
"""Return the current condition."""
if not self._forecast:
forecast = self._hourly_forecast or self._daily_forecast
if not forecast:
return
return self._condition_conversion(self._forecast[0].weather_type.id, None)
return self._condition_conversion(forecast[0].weather_type.id, None)
@property
def native_temperature(self):
@ -180,10 +191,9 @@ class IPMAWeather(WeatherEntity, IPMADevice):
return self._observation.wind_direction
@property
def forecast(self):
def _forecast(self, forecast: list[IPMAForecast] | None) -> list[Forecast]:
"""Return the forecast array."""
if not self._forecast:
if not forecast:
return []
return [
@ -198,5 +208,32 @@ class IPMAWeather(WeatherEntity, IPMADevice):
ATTR_FORECAST_NATIVE_WIND_SPEED: data_in.wind_strength,
ATTR_FORECAST_WIND_BEARING: data_in.wind_direction,
}
for data_in in self._forecast
for data_in in forecast
]
@property
def forecast(self) -> list[Forecast]:
"""Return the forecast array."""
return self._forecast(
self._hourly_forecast if self._period == 1 else self._daily_forecast
)
async def _try_update_forecast(
self,
forecast_type: Literal["daily", "hourly"],
period: int,
) -> None:
"""Try to update weather forecast."""
with contextlib.suppress(asyncio.TimeoutError):
async with async_timeout.timeout(10):
await self._update_forecast(forecast_type, period, False)
async def async_forecast_daily(self) -> list[Forecast]:
"""Return the daily forecast in native units."""
await self._try_update_forecast("daily", 24)
return self._forecast(self._daily_forecast)
async def async_forecast_hourly(self) -> list[Forecast]:
"""Return the hourly forecast in native units."""
await self._try_update_forecast("hourly", 1)
return self._forecast(self._hourly_forecast)

104
tests/components/ipma/snapshots/test_weather.ambr Normal file
View File

@ -0,0 +1,104 @@
# serializer version: 1
# name: test_forecast_service
dict({
'forecast': list([
dict({
'condition': 'rainy',
'datetime': datetime.datetime(2020, 1, 16, 0, 0),
'precipitation_probability': '100.0',
'temperature': 16.2,
'templow': 10.6,
'wind_bearing': 'S',
'wind_speed': 10.0,
}),
]),
})
# ---
# name: test_forecast_service.1
dict({
'forecast': list([
dict({
'condition': 'rainy',
'datetime': datetime.datetime(2020, 1, 15, 1, 0, tzinfo=datetime.timezone.utc),
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
dict({
'condition': 'clear-night',
'datetime': datetime.datetime(2020, 1, 15, 2, 0, tzinfo=datetime.timezone.utc),
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
]),
})
# ---
# name: test_forecast_subscription[daily]
list([
dict({
'condition': 'rainy',
'datetime': '2020-01-16T00:00:00',
'precipitation_probability': '100.0',
'temperature': 16.2,
'templow': 10.6,
'wind_bearing': 'S',
'wind_speed': 10.0,
}),
])
# ---
# name: test_forecast_subscription[daily].1
list([
dict({
'condition': 'rainy',
'datetime': '2020-01-16T00:00:00',
'precipitation_probability': '100.0',
'temperature': 16.2,
'templow': 10.6,
'wind_bearing': 'S',
'wind_speed': 10.0,
}),
])
# ---
# name: test_forecast_subscription[hourly]
list([
dict({
'condition': 'rainy',
'datetime': '2020-01-15T01:00:00+00:00',
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
dict({
'condition': 'clear-night',
'datetime': '2020-01-15T02:00:00+00:00',
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
])
# ---
# name: test_forecast_subscription[hourly].1
list([
dict({
'condition': 'rainy',
'datetime': '2020-01-15T01:00:00+00:00',
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
dict({
'condition': 'clear-night',
'datetime': '2020-01-15T02:00:00+00:00',
'precipitation_probability': 80.0,
'temperature': 12.0,
'wind_bearing': 'S',
'wind_speed': 32.7,
}),
])
# ---

60
tests/components/ipma/test_config_flow.py
View File

@ -1,15 +1,9 @@
"""Tests for IPMA config flow."""
from unittest.mock import Mock, patch
from homeassistant.components.ipma import DOMAIN, config_flow
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE
from homeassistant.components.ipma import config_flow
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from . import MockLocation
from tests.common import MockConfigEntry, mock_registry
async def test_show_config_form() -> None:
@ -120,53 +114,3 @@ async def test_flow_entry_config_entry_already_exists() -> None:
assert len(config_form.mock_calls) == 1
assert len(config_entries.mock_calls) == 1
assert len(flow._errors) == 1
async def test_config_entry_migration(hass: HomeAssistant) -> None:
"""Tests config entry without mode in unique_id can be migrated."""
ipma_entry = MockConfigEntry(
domain=DOMAIN,
title="Home",
data={CONF_LATITUDE: 0, CONF_LONGITUDE: 0, CONF_MODE: "daily"},
)
ipma_entry.add_to_hass(hass)
ipma_entry2 = MockConfigEntry(
domain=DOMAIN,
title="Home",
data={CONF_LATITUDE: 0, CONF_LONGITUDE: 0, CONF_MODE: "hourly"},
)
ipma_entry2.add_to_hass(hass)
mock_registry(
hass,
{
"weather.hometown": er.RegistryEntry(
entity_id="weather.hometown",
unique_id="0, 0",
platform="ipma",
config_entry_id=ipma_entry.entry_id,
),
"weather.hometown_2": er.RegistryEntry(
entity_id="weather.hometown_2",
unique_id="0, 0, hourly",
platform="ipma",
config_entry_id=ipma_entry.entry_id,
),
},
)
with patch(
"pyipma.location.Location.get",
return_value=MockLocation(),
):
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
ent_reg = er.async_get(hass)
weather_home = ent_reg.async_get("weather.hometown")
assert weather_home.unique_id == "0, 0, daily"
weather_home2 = ent_reg.async_get("weather.hometown_2")
assert weather_home2.unique_id == "0, 0, hourly"

100
tests/components/ipma/test_weather.py
View File

@ -1,9 +1,12 @@
"""The tests for the IPMA weather component."""
from datetime import datetime
import datetime
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.ipma.const import MIN_TIME_BETWEEN_UPDATES
from homeassistant.components.weather import (
ATTR_FORECAST,
ATTR_FORECAST_CONDITION,
@ -18,6 +21,8 @@ from homeassistant.components.weather import (
ATTR_WEATHER_TEMPERATURE,
ATTR_WEATHER_WIND_BEARING,
ATTR_WEATHER_WIND_SPEED,
DOMAIN as WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
)
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
@ -25,6 +30,7 @@ from homeassistant.core import HomeAssistant
from . import MockLocation
from tests.common import MockConfigEntry
from tests.typing import WebSocketGenerator
TEST_CONFIG = {
"name": "HomeTown",
@ -91,7 +97,7 @@ async def test_daily_forecast(hass: HomeAssistant) -> None:
assert state.state == "rainy"
forecast = state.attributes.get(ATTR_FORECAST)[0]
assert forecast.get(ATTR_FORECAST_TIME) == datetime(2020, 1, 16, 0, 0, 0)
assert forecast.get(ATTR_FORECAST_TIME) == datetime.datetime(2020, 1, 16, 0, 0, 0)
assert forecast.get(ATTR_FORECAST_CONDITION) == "rainy"
assert forecast.get(ATTR_FORECAST_TEMP) == 16.2
assert forecast.get(ATTR_FORECAST_TEMP_LOW) == 10.6
@ -144,3 +150,93 @@ async def test_failed_get_observation_forecast(hass: HomeAssistant) -> None:
assert data.get(ATTR_WEATHER_WIND_SPEED) is None
assert data.get(ATTR_WEATHER_WIND_BEARING) is None
assert state.attributes.get("friendly_name") == "HomeTown"
async def test_forecast_service(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
) -> None:
"""Test multiple forecast."""
with patch(
"pyipma.location.Location.get",
return_value=MockLocation(),
):
entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
response = await hass.services.async_call(
WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
{
"entity_id": "weather.hometown",
"type": "daily",
},
blocking=True,
return_response=True,
)
assert response == snapshot
response = await hass.services.async_call(
WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
{
"entity_id": "weather.hometown",
"type": "hourly",
},
blocking=True,
return_response=True,
)
assert response == snapshot
@pytest.mark.parametrize("forecast_type", ["daily", "hourly"])
async def test_forecast_subscription(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
forecast_type: str,
) -> None:
"""Test multiple forecast."""
client = await hass_ws_client(hass)
with patch(
"pyipma.location.Location.get",
return_value=MockLocation(),
):
entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await client.send_json_auto_id(
{
"type": "weather/subscribe_forecast",
"forecast_type": forecast_type,
"entity_id": "weather.hometown",
}
)
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 forecast1 == snapshot
freezer.tick(MIN_TIME_BETWEEN_UPDATES + 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 == snapshot