Add type annotations for MET (#58804)

* Add Typing

* Add missing types

* define w/o Null

* specify # type: ignore
This commit is contained in:
carstenschroeder 2021-11-01 19:37:03 +01:00 committed by GitHub
parent 388cdf4e94
commit 63c9cfdbc8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 76 additions and 40 deletions

View File

@ -1,10 +1,15 @@
"""The met component.""" """The met component."""
from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
from random import randrange from random import randrange
from types import MappingProxyType
from typing import Any, Callable
import metno import metno
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
CONF_ELEVATION, CONF_ELEVATION,
CONF_LATITUDE, CONF_LATITUDE,
@ -13,6 +18,7 @@ from homeassistant.const import (
LENGTH_FEET, LENGTH_FEET,
LENGTH_METERS, LENGTH_METERS,
) )
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.distance import convert as convert_distance from homeassistant.util.distance import convert as convert_distance
@ -32,7 +38,7 @@ PLATFORMS = ["weather"]
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry): async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up Met as config entry.""" """Set up Met as config entry."""
# Don't setup if tracking home location and latitude or longitude isn't set. # Don't setup if tracking home location and latitude or longitude isn't set.
# Also, filters out our onboarding default location. # Also, filters out our onboarding default location.
@ -62,7 +68,7 @@ async def async_setup_entry(hass, config_entry):
return True return True
async def async_unload_entry(hass, config_entry): async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload a config entry.""" """Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms( unload_ok = await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS config_entry, PLATFORMS
@ -77,9 +83,9 @@ async def async_unload_entry(hass, config_entry):
class MetDataUpdateCoordinator(DataUpdateCoordinator): class MetDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching Met data.""" """Class to manage fetching Met data."""
def __init__(self, hass, config_entry): def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Initialize global Met data updater.""" """Initialize global Met data updater."""
self._unsub_track_home = None self._unsub_track_home: Callable | None = None
self.weather = MetWeatherData( self.weather = MetWeatherData(
hass, config_entry.data, hass.config.units.is_metric hass, config_entry.data, hass.config.units.is_metric
) )
@ -89,19 +95,19 @@ class MetDataUpdateCoordinator(DataUpdateCoordinator):
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval) super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
async def _async_update_data(self): async def _async_update_data(self) -> MetWeatherData:
"""Fetch data from Met.""" """Fetch data from Met."""
try: try:
return await self.weather.fetch_data() return await self.weather.fetch_data()
except Exception as err: except Exception as err:
raise UpdateFailed(f"Update failed: {err}") from err raise UpdateFailed(f"Update failed: {err}") from err
def track_home(self): def track_home(self) -> None:
"""Start tracking changes to HA home setting.""" """Start tracking changes to HA home setting."""
if self._unsub_track_home: if self._unsub_track_home:
return return
async def _async_update_weather_data(_event=None): async def _async_update_weather_data(_event: str | None = None) -> None:
"""Update weather data.""" """Update weather data."""
if self.weather.set_coordinates(): if self.weather.set_coordinates():
await self.async_refresh() await self.async_refresh()
@ -110,7 +116,7 @@ class MetDataUpdateCoordinator(DataUpdateCoordinator):
EVENT_CORE_CONFIG_UPDATE, _async_update_weather_data EVENT_CORE_CONFIG_UPDATE, _async_update_weather_data
) )
def untrack_home(self): def untrack_home(self) -> None:
"""Stop tracking changes to HA home setting.""" """Stop tracking changes to HA home setting."""
if self._unsub_track_home: if self._unsub_track_home:
self._unsub_track_home() self._unsub_track_home()
@ -120,18 +126,20 @@ class MetDataUpdateCoordinator(DataUpdateCoordinator):
class MetWeatherData: class MetWeatherData:
"""Keep data for Met.no weather entities.""" """Keep data for Met.no weather entities."""
def __init__(self, hass, config, is_metric): def __init__(
self, hass: HomeAssistant, config: MappingProxyType[str, Any], is_metric: bool
) -> None:
"""Initialise the weather entity data.""" """Initialise the weather entity data."""
self.hass = hass self.hass = hass
self._config = config self._config = config
self._is_metric = is_metric self._is_metric = is_metric
self._weather_data = None self._weather_data: metno.MetWeatherData
self.current_weather_data = {} self.current_weather_data: dict = {}
self.daily_forecast = None self.daily_forecast = None
self.hourly_forecast = None self.hourly_forecast = None
self._coordinates = None self._coordinates: dict[str, str] | None = None
def set_coordinates(self): def set_coordinates(self) -> bool:
"""Weather data inialization - set the coordinates.""" """Weather data inialization - set the coordinates."""
if self._config.get(CONF_TRACK_HOME, False): if self._config.get(CONF_TRACK_HOME, False):
latitude = self.hass.config.latitude latitude = self.hass.config.latitude
@ -161,7 +169,7 @@ class MetWeatherData:
) )
return True return True
async def fetch_data(self): async def fetch_data(self) -> MetWeatherData:
"""Fetch data from API - (current weather and forecast).""" """Fetch data from API - (current weather and forecast)."""
await self._weather_data.fetching_data() await self._weather_data.fetching_data()
self.current_weather_data = self._weather_data.get_current_weather() self.current_weather_data = self._weather_data.get_current_weather()

View File

@ -1,5 +1,9 @@
"""Support for Met.no weather service.""" """Support for Met.no weather service."""
from __future__ import annotations
import logging import logging
from types import MappingProxyType
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -13,9 +17,10 @@ from homeassistant.components.weather import (
ATTR_WEATHER_WIND_BEARING, ATTR_WEATHER_WIND_BEARING,
ATTR_WEATHER_WIND_SPEED, ATTR_WEATHER_WIND_SPEED,
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
Forecast,
WeatherEntity, WeatherEntity,
) )
from homeassistant.config_entries import SOURCE_IMPORT from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
CONF_ELEVATION, CONF_ELEVATION,
CONF_LATITUDE, CONF_LATITUDE,
@ -29,9 +34,16 @@ from homeassistant.const import (
PRESSURE_INHG, PRESSURE_INHG,
TEMP_CELSIUS, TEMP_CELSIUS,
) )
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
T,
)
from homeassistant.util.distance import convert as convert_distance from homeassistant.util.distance import convert as convert_distance
from homeassistant.util.pressure import convert as convert_pressure from homeassistant.util.pressure import convert as convert_pressure
@ -67,7 +79,12 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
) )
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Met.no weather platform.""" """Set up the Met.no weather platform."""
_LOGGER.warning("Loading Met.no via platform config is deprecated") _LOGGER.warning("Loading Met.no via platform config is deprecated")
@ -84,7 +101,11 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
) )
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add a weather entity from a config_entry.""" """Add a weather entity from a config_entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id] coordinator = hass.data[DOMAIN][config_entry.entry_id]
async_add_entities( async_add_entities(
@ -110,7 +131,13 @@ def format_condition(condition: str) -> str:
class MetWeather(CoordinatorEntity, WeatherEntity): class MetWeather(CoordinatorEntity, WeatherEntity):
"""Implementation of a Met.no weather condition.""" """Implementation of a Met.no weather condition."""
def __init__(self, coordinator, config, is_metric, hourly): def __init__(
self,
coordinator: DataUpdateCoordinator[T],
config: MappingProxyType[str, Any],
is_metric: bool,
hourly: bool,
) -> None:
"""Initialise the platform with a data instance and site.""" """Initialise the platform with a data instance and site."""
super().__init__(coordinator) super().__init__(coordinator)
self._config = config self._config = config
@ -118,12 +145,12 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
self._hourly = hourly self._hourly = hourly
@property @property
def track_home(self): def track_home(self) -> (Any | bool):
"""Return if we are tracking home.""" """Return if we are tracking home."""
return self._config.get(CONF_TRACK_HOME, False) return self._config.get(CONF_TRACK_HOME, False)
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return unique ID.""" """Return unique ID."""
name_appendix = "" name_appendix = ""
if self._hourly: if self._hourly:
@ -134,7 +161,7 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
return f"{self._config[CONF_LATITUDE]}-{self._config[CONF_LONGITUDE]}{name_appendix}" return f"{self._config[CONF_LATITUDE]}-{self._config[CONF_LONGITUDE]}{name_appendix}"
@property @property
def name(self): def name(self) -> str:
"""Return the name of the sensor.""" """Return the name of the sensor."""
name = self._config.get(CONF_NAME) name = self._config.get(CONF_NAME)
name_appendix = "" name_appendix = ""
@ -155,25 +182,25 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
return not self._hourly return not self._hourly
@property @property
def condition(self): def condition(self) -> str | None:
"""Return the current condition.""" """Return the current condition."""
condition = self.coordinator.data.current_weather_data.get("condition") condition = self.coordinator.data.current_weather_data.get("condition")
return format_condition(condition) return format_condition(condition)
@property @property
def temperature(self): def temperature(self) -> float | None:
"""Return the temperature.""" """Return the temperature."""
return self.coordinator.data.current_weather_data.get( return self.coordinator.data.current_weather_data.get(
ATTR_MAP[ATTR_WEATHER_TEMPERATURE] ATTR_MAP[ATTR_WEATHER_TEMPERATURE]
) )
@property @property
def temperature_unit(self): def temperature_unit(self) -> str:
"""Return the unit of measurement.""" """Return the unit of measurement."""
return TEMP_CELSIUS return TEMP_CELSIUS
@property @property
def pressure(self): def pressure(self) -> float | None:
"""Return the pressure.""" """Return the pressure."""
pressure_hpa = self.coordinator.data.current_weather_data.get( pressure_hpa = self.coordinator.data.current_weather_data.get(
ATTR_MAP[ATTR_WEATHER_PRESSURE] ATTR_MAP[ATTR_WEATHER_PRESSURE]
@ -184,14 +211,14 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
return round(convert_pressure(pressure_hpa, PRESSURE_HPA, PRESSURE_INHG), 2) return round(convert_pressure(pressure_hpa, PRESSURE_HPA, PRESSURE_INHG), 2)
@property @property
def humidity(self): def humidity(self) -> float | None:
"""Return the humidity.""" """Return the humidity."""
return self.coordinator.data.current_weather_data.get( return self.coordinator.data.current_weather_data.get(
ATTR_MAP[ATTR_WEATHER_HUMIDITY] ATTR_MAP[ATTR_WEATHER_HUMIDITY]
) )
@property @property
def wind_speed(self): def wind_speed(self) -> float | None:
"""Return the wind speed.""" """Return the wind speed."""
speed_km_h = self.coordinator.data.current_weather_data.get( speed_km_h = self.coordinator.data.current_weather_data.get(
ATTR_MAP[ATTR_WEATHER_WIND_SPEED] ATTR_MAP[ATTR_WEATHER_WIND_SPEED]
@ -203,26 +230,26 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
return int(round(speed_mi_h)) return int(round(speed_mi_h))
@property @property
def wind_bearing(self): def wind_bearing(self) -> float | str | None:
"""Return the wind direction.""" """Return the wind direction."""
return self.coordinator.data.current_weather_data.get( return self.coordinator.data.current_weather_data.get(
ATTR_MAP[ATTR_WEATHER_WIND_BEARING] ATTR_MAP[ATTR_WEATHER_WIND_BEARING]
) )
@property @property
def attribution(self): def attribution(self) -> str:
"""Return the attribution.""" """Return the attribution."""
return ATTRIBUTION return ATTRIBUTION
@property @property
def forecast(self): def forecast(self) -> list[Forecast] | None:
"""Return the forecast array.""" """Return the forecast array."""
if self._hourly: if self._hourly:
met_forecast = self.coordinator.data.hourly_forecast met_forecast = self.coordinator.data.hourly_forecast
else: else:
met_forecast = self.coordinator.data.daily_forecast met_forecast = self.coordinator.data.daily_forecast
required_keys = {ATTR_FORECAST_TEMP, ATTR_FORECAST_TIME} required_keys = {ATTR_FORECAST_TEMP, ATTR_FORECAST_TIME}
ha_forecast = [] ha_forecast: list[Forecast] = []
for met_item in met_forecast: for met_item in met_forecast:
if not set(met_item).issuperset(required_keys): if not set(met_item).issuperset(required_keys):
continue continue
@ -232,26 +259,27 @@ class MetWeather(CoordinatorEntity, WeatherEntity):
if met_item.get(v) is not None if met_item.get(v) is not None
} }
if not self._is_metric and ATTR_FORECAST_PRECIPITATION in ha_item: if not self._is_metric and ATTR_FORECAST_PRECIPITATION in ha_item:
precip_inches = convert_distance( if ha_item[ATTR_FORECAST_PRECIPITATION] is not None:
ha_item[ATTR_FORECAST_PRECIPITATION], precip_inches = convert_distance(
LENGTH_MILLIMETERS, ha_item[ATTR_FORECAST_PRECIPITATION],
LENGTH_INCHES, LENGTH_MILLIMETERS,
) LENGTH_INCHES,
)
ha_item[ATTR_FORECAST_PRECIPITATION] = round(precip_inches, 2) ha_item[ATTR_FORECAST_PRECIPITATION] = round(precip_inches, 2)
if ha_item.get(ATTR_FORECAST_CONDITION): if ha_item.get(ATTR_FORECAST_CONDITION):
ha_item[ATTR_FORECAST_CONDITION] = format_condition( ha_item[ATTR_FORECAST_CONDITION] = format_condition(
ha_item[ATTR_FORECAST_CONDITION] ha_item[ATTR_FORECAST_CONDITION]
) )
ha_forecast.append(ha_item) ha_forecast.append(ha_item) # type: ignore[arg-type]
return ha_forecast return ha_forecast
@property @property
def device_info(self): def device_info(self) -> DeviceInfo:
"""Device info.""" """Device info."""
return DeviceInfo( return DeviceInfo(
default_name="Forecast", default_name="Forecast",
entry_type="service", entry_type="service",
identifiers={(DOMAIN,)}, identifiers={(DOMAIN,)}, # type: ignore[arg-type]
manufacturer="Met.no", manufacturer="Met.no",
model="Forecast", model="Forecast",
) )