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

Bump pyipma to 3.0.2 (#76332) 

* upgrade to pyipma 3.0.0

* bump to support python3.9

* remove deprecated async_setup_platform

* full coverage

* add migrate
pull/77597/head
Diogo Gomes 2022-08-31 12:00:42 +01:00 committed by GitHub
parent e5eddba223
commit f98e86d3a6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 205 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -18,6 +18,13 @@ class IpmaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Init IpmaFlowHandler.""" """Init IpmaFlowHandler."""
self._errors = {} self._errors = {}
async def async_step_import(self, config):
"""Import a configuration from config.yaml."""
self._async_abort_entries_match(config)
config[CONF_MODE] = "daily"
return await self.async_step_user(user_input=config)
async def async_step_user(self, user_input=None): async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user.""" """Handle a flow initialized by the user."""
self._errors = {} self._errors = {}

2
homeassistant/components/ipma/manifest.json
View File

@ -3,7 +3,7 @@
"name": "Instituto Portugu\u00eas do Mar e Atmosfera (IPMA)", "name": "Instituto Portugu\u00eas do Mar e Atmosfera (IPMA)",
"config_flow": true, "config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/ipma", "documentation": "https://www.home-assistant.io/integrations/ipma",
"requirements": ["pyipma==2.0.5"], "requirements": ["pyipma==3.0.2"],
"codeowners": ["@dgomes", "@abmantis"], "codeowners": ["@dgomes", "@abmantis"],
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"loggers": ["geopy", "pyipma"] "loggers": ["geopy", "pyipma"]

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

@ -6,10 +6,12 @@ import logging
import async_timeout import async_timeout
from pyipma.api import IPMA_API from pyipma.api import IPMA_API
from pyipma.forecast import Forecast
from pyipma.location import Location from pyipma.location import Location
import voluptuous as vol import voluptuous as vol
from homeassistant.components.weather import ( from homeassistant.components.weather import (
ATTR_CONDITION_CLEAR_NIGHT,
ATTR_CONDITION_CLOUDY, ATTR_CONDITION_CLOUDY,
ATTR_CONDITION_EXCEPTIONAL, ATTR_CONDITION_EXCEPTIONAL,
ATTR_CONDITION_FOG, ATTR_CONDITION_FOG,
@ -48,9 +50,8 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, entity_registry from homeassistant.helpers import config_validation as cv, entity_registry
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.sun import is_up
from homeassistant.util import Throttle from homeassistant.util import Throttle
from homeassistant.util.dt import now, parse_datetime
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -73,6 +74,7 @@ CONDITION_CLASSES = {
ATTR_CONDITION_WINDY: [], ATTR_CONDITION_WINDY: [],
ATTR_CONDITION_WINDY_VARIANT: [], ATTR_CONDITION_WINDY_VARIANT: [],
ATTR_CONDITION_EXCEPTIONAL: [], ATTR_CONDITION_EXCEPTIONAL: [],
ATTR_CONDITION_CLEAR_NIGHT: [-1],
} }
FORECAST_MODE = ["hourly", "daily"] FORECAST_MODE = ["hourly", "daily"]
@ -87,31 +89,6 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
) )
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the ipma platform.
Deprecated.
"""
_LOGGER.warning("Loading IPMA via platform config is deprecated")
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
if None in (latitude, longitude):
_LOGGER.error("Latitude or longitude not set in Home Assistant config")
return
api = await async_get_api(hass)
location = await async_get_location(hass, api, latitude, longitude)
async_add_entities([IPMAWeather(location, api, config)], True)
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: ConfigEntry, config_entry: ConfigEntry,
@ -180,21 +157,24 @@ class IPMAWeather(WeatherEntity):
_attr_native_temperature_unit = TEMP_CELSIUS _attr_native_temperature_unit = TEMP_CELSIUS
_attr_native_wind_speed_unit = SPEED_KILOMETERS_PER_HOUR _attr_native_wind_speed_unit = SPEED_KILOMETERS_PER_HOUR
_attr_attribution = ATTRIBUTION
def __init__(self, location: Location, api: IPMA_API, config): def __init__(self, location: Location, api: IPMA_API, config):
"""Initialise the platform with a data instance and station name.""" """Initialise the platform with a data instance and station name."""
self._api = api self._api = api
self._location_name = config.get(CONF_NAME, location.name) self._location_name = config.get(CONF_NAME, location.name)
self._mode = config.get(CONF_MODE) self._mode = config.get(CONF_MODE)
self._period = 1 if config.get(CONF_MODE) == "hourly" else 24
self._location = location self._location = location
self._observation = None self._observation = None
self._forecast = None self._forecast: list[Forecast] = []
@Throttle(MIN_TIME_BETWEEN_UPDATES) @Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self) -> None: async def async_update(self) -> None:
"""Update Condition and Forecast.""" """Update Condition and Forecast."""
async with async_timeout.timeout(10): async with async_timeout.timeout(10):
new_observation = await self._location.observation(self._api) new_observation = await self._location.observation(self._api)
new_forecast = await self._location.forecast(self._api) new_forecast = await self._location.forecast(self._api, self._period)
if new_observation: if new_observation:
self._observation = new_observation self._observation = new_observation
@ -207,8 +187,9 @@ class IPMAWeather(WeatherEntity):
_LOGGER.warning("Could not update weather forecast") _LOGGER.warning("Could not update weather forecast")
_LOGGER.debug( _LOGGER.debug(
"Updated location %s, observation %s", "Updated location %s based on %s, current observation %s",
self._location.name, self._location.name,
self._location.station,
self._observation, self._observation,
) )
@ -217,30 +198,28 @@ class IPMAWeather(WeatherEntity):
"""Return a unique id.""" """Return a unique id."""
return f"{self._location.station_latitude}, {self._location.station_longitude}, {self._mode}" return f"{self._location.station_latitude}, {self._location.station_longitude}, {self._mode}"
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property @property
def name(self): def name(self):
"""Return the name of the station.""" """Return the name of the station."""
return self._location_name return self._location_name
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):
identifier = -identifier
return next(
(k for k, v in CONDITION_CLASSES.items() if identifier in v),
None,
)
@property @property
def condition(self): def condition(self):
"""Return the current condition.""" """Return the current condition."""
if not self._forecast: if not self._forecast:
return return
return next( return self._condition_conversion(self._forecast[0].weather_type.id, None)
(
k
for k, v in CONDITION_CLASSES.items()
if self._forecast[0].weather_type in v
),
None,
)
@property @property
def native_temperature(self): def native_temperature(self):
@ -288,57 +267,17 @@ class IPMAWeather(WeatherEntity):
if not self._forecast: if not self._forecast:
return [] return []
if self._mode == "hourly": return [
forecast_filtered = [ {
x ATTR_FORECAST_TIME: data_in.forecast_date,
for x in self._forecast ATTR_FORECAST_CONDITION: self._condition_conversion(
if x.forecasted_hours == 1 data_in.weather_type.id, data_in.forecast_date
and parse_datetime(x.forecast_date) ),
> (now().utcnow() - timedelta(hours=1)) ATTR_FORECAST_NATIVE_TEMP_LOW: data_in.min_temperature,
] ATTR_FORECAST_NATIVE_TEMP: data_in.max_temperature,
ATTR_FORECAST_PRECIPITATION_PROBABILITY: data_in.precipitation_probability,
fcdata_out = [ ATTR_FORECAST_NATIVE_WIND_SPEED: data_in.wind_strength,
{ ATTR_FORECAST_WIND_BEARING: data_in.wind_direction,
ATTR_FORECAST_TIME: data_in.forecast_date, }
ATTR_FORECAST_CONDITION: next( for data_in in self._forecast
( ]
k
for k, v in CONDITION_CLASSES.items()
if int(data_in.weather_type) in v
),
None,
),
ATTR_FORECAST_NATIVE_TEMP: float(data_in.feels_like_temperature),
ATTR_FORECAST_PRECIPITATION_PROBABILITY: (
int(float(data_in.precipitation_probability))
if int(float(data_in.precipitation_probability)) >= 0
else None
),
ATTR_FORECAST_NATIVE_WIND_SPEED: data_in.wind_strength,
ATTR_FORECAST_WIND_BEARING: data_in.wind_direction,
}
for data_in in forecast_filtered
]
else:
forecast_filtered = [f for f in self._forecast if f.forecasted_hours == 24]
fcdata_out = [
{
ATTR_FORECAST_TIME: data_in.forecast_date,
ATTR_FORECAST_CONDITION: next(
(
k
for k, v in CONDITION_CLASSES.items()
if int(data_in.weather_type) in v
),
None,
),
ATTR_FORECAST_NATIVE_TEMP_LOW: data_in.min_temperature,
ATTR_FORECAST_NATIVE_TEMP: data_in.max_temperature,
ATTR_FORECAST_PRECIPITATION_PROBABILITY: data_in.precipitation_probability,
ATTR_FORECAST_NATIVE_WIND_SPEED: data_in.wind_strength,
ATTR_FORECAST_WIND_BEARING: data_in.wind_direction,
}
for data_in in forecast_filtered
]
return fcdata_out

2
requirements_all.txt
View File

@ -1602,7 +1602,7 @@ pyinsteon==1.2.0
pyintesishome==1.8.0 pyintesishome==1.8.0
# homeassistant.components.ipma # homeassistant.components.ipma
pyipma==2.0.5 pyipma==3.0.2
# homeassistant.components.ipp # homeassistant.components.ipp
pyipp==0.11.0 pyipp==0.11.0

2
requirements_test_all.txt
View File

@ -1118,7 +1118,7 @@ pyicloud==1.0.0
pyinsteon==1.2.0 pyinsteon==1.2.0
# homeassistant.components.ipma # homeassistant.components.ipma
pyipma==2.0.5 pyipma==3.0.2
# homeassistant.components.ipp # homeassistant.components.ipp
pyipp==0.11.0 pyipp==0.11.0

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

@ -2,8 +2,10 @@
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from homeassistant import config_entries
from homeassistant.components.ipma import DOMAIN, config_flow from homeassistant.components.ipma import DOMAIN, config_flow
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE, CONF_NAME
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@ -11,6 +13,13 @@ from .test_weather import MockLocation
from tests.common import MockConfigEntry, mock_registry from tests.common import MockConfigEntry, mock_registry
ENTRY_CONFIG = {
CONF_NAME: "Home Town",
CONF_LATITUDE: "1",
CONF_LONGITUDE: "2",
CONF_MODE: "hourly",
}
async def test_show_config_form(): async def test_show_config_form():
"""Test show configuration form.""" """Test show configuration form."""
@ -172,3 +181,45 @@ async def test_config_entry_migration(hass):
weather_home2 = ent_reg.async_get("weather.hometown_2") weather_home2 = ent_reg.async_get("weather.hometown_2")
assert weather_home2.unique_id == "0, 0, hourly" assert weather_home2.unique_id == "0, 0, hourly"
async def test_import_flow_success(hass):
"""Test a successful import of yaml."""
with patch(
"homeassistant.components.ipma.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=ENTRY_CONFIG,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Home Town"
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_flow_already_exist(hass):
"""Test import of yaml already exist."""
MockConfigEntry(
domain=DOMAIN,
data=ENTRY_CONFIG,
).add_to_hass(hass)
with patch(
"homeassistant.components.ipma.async_setup_entry",
return_value=True,
):
result3 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=ENTRY_CONFIG,
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.ABORT
assert result3["reason"] == "already_configured"

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

@ -1,8 +1,10 @@
"""The tests for the IPMA weather component.""" """The tests for the IPMA weather component."""
from collections import namedtuple from collections import namedtuple
from datetime import datetime, timezone
from unittest.mock import patch from unittest.mock import patch
from homeassistant.components import weather from freezegun import freeze_time
from homeassistant.components.weather import ( from homeassistant.components.weather import (
ATTR_FORECAST, ATTR_FORECAST,
ATTR_FORECAST_CONDITION, ATTR_FORECAST_CONDITION,
@ -19,8 +21,7 @@ from homeassistant.components.weather import (
ATTR_WEATHER_WIND_SPEED, ATTR_WEATHER_WIND_SPEED,
DOMAIN as WEATHER_DOMAIN, DOMAIN as WEATHER_DOMAIN,
) )
from homeassistant.setup import async_setup_component from homeassistant.const import STATE_UNKNOWN
from homeassistant.util.dt import now
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -31,6 +32,13 @@ TEST_CONFIG = {
"mode": "daily", "mode": "daily",
} }
TEST_CONFIG_HOURLY = {
"name": "HomeTown",
"latitude": "40.00",
"longitude": "-8.00",
"mode": "hourly",
}
class MockLocation: class MockLocation:
"""Mock Location from pyipma.""" """Mock Location from pyipma."""
@ -52,7 +60,7 @@ class MockLocation:
return Observation(0.0, 71.0, 1000.0, 0.0, 18.0, "NW", 3.94) return Observation(0.0, 71.0, 1000.0, 0.0, 18.0, "NW", 3.94)
async def forecast(self, api): async def forecast(self, api, period):
"""Mock Forecast.""" """Mock Forecast."""
Forecast = namedtuple( Forecast = namedtuple(
"Forecast", "Forecast",
@ -72,42 +80,67 @@ class MockLocation:
], ],
) )
return [ WeatherType = namedtuple("WeatherType", ["id", "en", "pt"])
Forecast(
None, if period == 24:
"2020-01-15T00:00:00", return [
24, Forecast(
None, None,
16.2, datetime(2020, 1, 16, 0, 0, 0),
10.6, 24,
"100.0", None,
13.4, 16.2,
"2020-01-15T07:51:00", 10.6,
9, "100.0",
"S", 13.4,
"10", "2020-01-15T07:51:00",
), WeatherType(9, "Rain/showers", "Chuva/aguaceiros"),
Forecast( "S",
"7.7", "10",
now().utcnow().strftime("%Y-%m-%dT%H:%M:%S"), ),
1, ]
"86.9", if period == 1:
None, return [
None, Forecast(
"80.0", "7.7",
10.6, datetime(2020, 1, 15, 1, 0, 0, tzinfo=timezone.utc),
"2020-01-15T07:51:00", 1,
10, "86.9",
"S", 12.0,
"32.7", None,
), 80.0,
] 10.6,
"2020-01-15T02:51:00",
WeatherType(10, "Light rain", "Chuva fraca ou chuvisco"),
"S",
"32.7",
),
Forecast(
"5.7",
datetime(2020, 1, 15, 2, 0, 0, tzinfo=timezone.utc),
1,
"86.9",
12.0,
None,
80.0,
10.6,
"2020-01-15T02:51:00",
WeatherType(1, "Clear sky", "C\u00e9u limpo"),
"S",
"32.7",
),
]
@property @property
def name(self): def name(self):
"""Mock location.""" """Mock location."""
return "HomeTown" return "HomeTown"
@property
def station(self):
"""Mock station."""
return "HomeTown Station"
@property @property
def station_latitude(self): def station_latitude(self):
"""Mock latitude.""" """Mock latitude."""
@ -129,35 +162,22 @@ class MockLocation:
return 0 return 0
async def test_setup_configuration(hass): class MockBadLocation(MockLocation):
"""Test for successfully setting up the IPMA platform.""" """Mock Location with unresponsive api."""
with patch(
"homeassistant.components.ipma.weather.async_get_location",
return_value=MockLocation(),
):
assert await async_setup_component(
hass,
weather.DOMAIN,
{"weather": {"name": "HomeTown", "platform": "ipma", "mode": "hourly"}},
)
await hass.async_block_till_done()
state = hass.states.get("weather.hometown") async def observation(self, api):
assert state.state == "rainy" """Mock Observation."""
return None
data = state.attributes async def forecast(self, api, period):
assert data.get(ATTR_WEATHER_TEMPERATURE) == 18.0 """Mock Forecast."""
assert data.get(ATTR_WEATHER_HUMIDITY) == 71 return []
assert data.get(ATTR_WEATHER_PRESSURE) == 1000.0
assert data.get(ATTR_WEATHER_WIND_SPEED) == 3.94
assert data.get(ATTR_WEATHER_WIND_BEARING) == "NW"
assert state.attributes.get("friendly_name") == "HomeTown"
async def test_setup_config_flow(hass): async def test_setup_config_flow(hass):
"""Test for successfully setting up the IPMA platform.""" """Test for successfully setting up the IPMA platform."""
with patch( with patch(
"homeassistant.components.ipma.weather.async_get_location", "pyipma.location.Location.get",
return_value=MockLocation(), return_value=MockLocation(),
): ):
entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG) entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG)
@ -179,21 +199,18 @@ async def test_setup_config_flow(hass):
async def test_daily_forecast(hass): async def test_daily_forecast(hass):
"""Test for successfully getting daily forecast.""" """Test for successfully getting daily forecast."""
with patch( with patch(
"homeassistant.components.ipma.weather.async_get_location", "pyipma.location.Location.get",
return_value=MockLocation(), return_value=MockLocation(),
): ):
assert await async_setup_component( entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG)
hass, await hass.config_entries.async_forward_entry_setup(entry, WEATHER_DOMAIN)
weather.DOMAIN,
{"weather": {"name": "HomeTown", "platform": "ipma", "mode": "daily"}},
)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("weather.hometown") state = hass.states.get("weather.hometown")
assert state.state == "rainy" assert state.state == "rainy"
forecast = state.attributes.get(ATTR_FORECAST)[0] forecast = state.attributes.get(ATTR_FORECAST)[0]
assert forecast.get(ATTR_FORECAST_TIME) == "2020-01-15T00:00:00" assert forecast.get(ATTR_FORECAST_TIME) == datetime(2020, 1, 16, 0, 0, 0)
assert forecast.get(ATTR_FORECAST_CONDITION) == "rainy" assert forecast.get(ATTR_FORECAST_CONDITION) == "rainy"
assert forecast.get(ATTR_FORECAST_TEMP) == 16.2 assert forecast.get(ATTR_FORECAST_TEMP) == 16.2
assert forecast.get(ATTR_FORECAST_TEMP_LOW) == 10.6 assert forecast.get(ATTR_FORECAST_TEMP_LOW) == 10.6
@ -202,17 +219,15 @@ async def test_daily_forecast(hass):
assert forecast.get(ATTR_FORECAST_WIND_BEARING) == "S" assert forecast.get(ATTR_FORECAST_WIND_BEARING) == "S"
@freeze_time("2020-01-14 23:00:00")
async def test_hourly_forecast(hass): async def test_hourly_forecast(hass):
"""Test for successfully getting daily forecast.""" """Test for successfully getting daily forecast."""
with patch( with patch(
"homeassistant.components.ipma.weather.async_get_location", "pyipma.location.Location.get",
return_value=MockLocation(), return_value=MockLocation(),
): ):
assert await async_setup_component( entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG_HOURLY)
hass, await hass.config_entries.async_forward_entry_setup(entry, WEATHER_DOMAIN)
weather.DOMAIN,
{"weather": {"name": "HomeTown", "platform": "ipma", "mode": "hourly"}},
)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("weather.hometown") state = hass.states.get("weather.hometown")
@ -220,7 +235,29 @@ async def test_hourly_forecast(hass):
forecast = state.attributes.get(ATTR_FORECAST)[0] forecast = state.attributes.get(ATTR_FORECAST)[0]
assert forecast.get(ATTR_FORECAST_CONDITION) == "rainy" assert forecast.get(ATTR_FORECAST_CONDITION) == "rainy"
assert forecast.get(ATTR_FORECAST_TEMP) == 7.7 assert forecast.get(ATTR_FORECAST_TEMP) == 12.0
assert forecast.get(ATTR_FORECAST_PRECIPITATION_PROBABILITY) == 80.0 assert forecast.get(ATTR_FORECAST_PRECIPITATION_PROBABILITY) == 80.0
assert forecast.get(ATTR_FORECAST_WIND_SPEED) == 32.7 assert forecast.get(ATTR_FORECAST_WIND_SPEED) == 32.7
assert forecast.get(ATTR_FORECAST_WIND_BEARING) == "S" assert forecast.get(ATTR_FORECAST_WIND_BEARING) == "S"
async def test_failed_get_observation_forecast(hass):
"""Test for successfully setting up the IPMA platform."""
with patch(
"pyipma.location.Location.get",
return_value=MockBadLocation(),
):
entry = MockConfigEntry(domain="ipma", data=TEST_CONFIG)
await hass.config_entries.async_forward_entry_setup(entry, WEATHER_DOMAIN)
await hass.async_block_till_done()
state = hass.states.get("weather.hometown")
assert state.state == STATE_UNKNOWN
data = state.attributes
assert data.get(ATTR_WEATHER_TEMPERATURE) is None
assert data.get(ATTR_WEATHER_HUMIDITY) is None
assert data.get(ATTR_WEATHER_PRESSURE) is 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"