Fix smhi typing (#50690)

This commit is contained in:
Martin Hjelmare 2021-05-15 21:38:12 +02:00 committed by GitHub
parent cad41cd4ed
commit 5da64d01e2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 58 additions and 40 deletions

View file

@ -1,10 +1,15 @@
"""Config flow to configure SMHI component."""
from __future__ import annotations
from typing import Any
from smhi.smhi_lib import Smhi, SmhiForecastException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
import homeassistant.helpers.config_validation as cv
from homeassistant.util import slugify
@ -13,7 +18,7 @@ from .const import DOMAIN, HOME_LOCATION_NAME
@callback
def smhi_locations(hass: HomeAssistant):
def smhi_locations(hass: HomeAssistant) -> set[str]:
"""Return configurations of SMHI component."""
return {
(slugify(entry.data[CONF_NAME]))
@ -28,9 +33,11 @@ class SmhiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize SMHI forecast configuration flow."""
self._errors = {}
self._errors: dict[str, str] = {}
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
self._errors = {}
@ -79,8 +86,11 @@ class SmhiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return name in smhi_locations(self.hass)
async def _show_config_form(
self, name: str = None, latitude: str = None, longitude: str = None
):
self,
name: str | None = None,
latitude: float | None = None,
longitude: float | None = None,
) -> FlowResult:
"""Show the configuration form to edit location data."""
return self.async_show_form(
step_id="user",
@ -94,7 +104,7 @@ class SmhiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=self._errors,
)
async def _check_location(self, longitude: str, latitude: str) -> bool:
async def _check_location(self, longitude: float, latitude: float) -> bool:
"""Return true if location is ok."""
try:
session = aiohttp_client.async_get_clientsession(self.hass)

View file

@ -1,9 +1,11 @@
"""Constants in smhi component."""
from typing import Final
from homeassistant.components.weather import DOMAIN as WEATHER_DOMAIN
ATTR_SMHI_CLOUDINESS = "cloudiness"
ATTR_SMHI_WIND_GUST_SPEED = "wind_gust_speed"
ATTR_SMHI_THUNDER_PROBABILITY = "thunder_probability"
ATTR_SMHI_CLOUDINESS: Final = "cloudiness"
ATTR_SMHI_WIND_GUST_SPEED: Final = "wind_gust_speed"
ATTR_SMHI_THUNDER_PROBABILITY: Final = "thunder_probability"
DOMAIN = "smhi"

View file

@ -2,13 +2,14 @@
from __future__ import annotations
import asyncio
from datetime import timedelta
from datetime import datetime, timedelta
import logging
from typing import Any, Final, TypedDict
import aiohttp
import async_timeout
from smhi import Smhi
from smhi.smhi_lib import SmhiForecastException
from smhi.smhi_lib import SmhiForecast, SmhiForecastException
from homeassistant.components.weather import (
ATTR_CONDITION_CLOUDY,
@ -36,6 +37,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later
from homeassistant.util import Throttle, slugify
from .const import (
@ -48,7 +51,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
# Used to map condition from API results
CONDITION_CLASSES = {
CONDITION_CLASSES: Final[dict[str, list[int]]] = {
ATTR_CONDITION_CLOUDY: [5, 6],
ATTR_CONDITION_FOG: [7],
ATTR_CONDITION_HAIL: [],
@ -72,8 +75,10 @@ MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=31)
async def async_setup_entry(
hass: HomeAssistant, config_entry: ConfigEntry, config_entries
) -> bool:
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add a weather entity from map location."""
location = config_entry.data
name = slugify(location[CONF_NAME])
@ -88,8 +93,7 @@ async def async_setup_entry(
)
entity.entity_id = ENTITY_ID_SENSOR_FORMAT.format(name)
config_entries([entity], True)
return True
async_add_entities([entity], True)
class SmhiWeather(WeatherEntity):
@ -100,14 +104,14 @@ class SmhiWeather(WeatherEntity):
name: str,
latitude: str,
longitude: str,
session: aiohttp.ClientSession = None,
session: aiohttp.ClientSession,
) -> None:
"""Initialize the SMHI weather entity."""
self._name = name
self._latitude = latitude
self._longitude = longitude
self._forecasts = None
self._forecasts: list[SmhiForecast] | None = None
self._fail_count = 0
self._smhi_api = Smhi(self._longitude, self._latitude, session=session)
@ -128,17 +132,15 @@ class SmhiWeather(WeatherEntity):
_LOGGER.error("Failed to connect to SMHI API, retry in 5 minutes")
self._fail_count += 1
if self._fail_count < 3:
self.hass.helpers.event.async_call_later(
RETRY_TIMEOUT, self.retry_update
)
async_call_later(self.hass, RETRY_TIMEOUT, self.retry_update)
async def retry_update(self, _):
async def retry_update(self, _: datetime) -> None:
"""Retry refresh weather forecast."""
await self.async_update( # pylint: disable=unexpected-keyword-arg
no_throttle=True
)
async def get_weather_forecast(self) -> []:
async def get_weather_forecast(self) -> list[SmhiForecast]:
"""Return the current forecasts from SMHI API."""
return await self._smhi_api.async_get_forecast()
@ -148,7 +150,7 @@ class SmhiWeather(WeatherEntity):
return self._name
@property
def temperature(self) -> int:
def temperature(self) -> int | None:
"""Return the temperature."""
if self._forecasts is not None:
return self._forecasts[0].temperature
@ -160,14 +162,14 @@ class SmhiWeather(WeatherEntity):
return TEMP_CELSIUS
@property
def humidity(self) -> int:
def humidity(self) -> int | None:
"""Return the humidity."""
if self._forecasts is not None:
return self._forecasts[0].humidity
return None
@property
def wind_speed(self) -> float:
def wind_speed(self) -> float | None:
"""Return the wind speed."""
if self._forecasts is not None:
# Convert from m/s to km/h
@ -175,7 +177,7 @@ class SmhiWeather(WeatherEntity):
return None
@property
def wind_gust_speed(self) -> float:
def wind_gust_speed(self) -> float | None:
"""Return the wind gust speed."""
if self._forecasts is not None:
# Convert from m/s to km/h
@ -183,42 +185,42 @@ class SmhiWeather(WeatherEntity):
return None
@property
def wind_bearing(self) -> int:
def wind_bearing(self) -> int | None:
"""Return the wind bearing."""
if self._forecasts is not None:
return self._forecasts[0].wind_direction
return None
@property
def visibility(self) -> float:
def visibility(self) -> float | None:
"""Return the visibility."""
if self._forecasts is not None:
return self._forecasts[0].horizontal_visibility
return None
@property
def pressure(self) -> int:
def pressure(self) -> int | None:
"""Return the pressure."""
if self._forecasts is not None:
return self._forecasts[0].pressure
return None
@property
def cloudiness(self) -> int:
def cloudiness(self) -> int | None:
"""Return the cloudiness."""
if self._forecasts is not None:
return self._forecasts[0].cloudiness
return None
@property
def thunder_probability(self) -> int:
def thunder_probability(self) -> int | None:
"""Return the chance of thunder, unit Percent."""
if self._forecasts is not None:
return self._forecasts[0].thunder
return None
@property
def condition(self) -> str:
def condition(self) -> str | None:
"""Return the weather condition."""
if self._forecasts is None:
return None
@ -233,7 +235,7 @@ class SmhiWeather(WeatherEntity):
return "Swedish weather institute (SMHI)"
@property
def forecast(self) -> list:
def forecast(self) -> list[dict[str, Any]] | None:
"""Return the forecast."""
if self._forecasts is None or len(self._forecasts) < 2:
return None
@ -258,9 +260,9 @@ class SmhiWeather(WeatherEntity):
return data
@property
def extra_state_attributes(self) -> dict:
def extra_state_attributes(self) -> ExtraAttributes:
"""Return SMHI specific attributes."""
extra_attributes = {}
extra_attributes: ExtraAttributes = {}
if self.cloudiness is not None:
extra_attributes[ATTR_SMHI_CLOUDINESS] = self.cloudiness
if self.wind_gust_speed is not None:
@ -268,3 +270,11 @@ class SmhiWeather(WeatherEntity):
if self.thunder_probability is not None:
extra_attributes[ATTR_SMHI_THUNDER_PROBABILITY] = self.thunder_probability
return extra_attributes
class ExtraAttributes(TypedDict, total=False):
"""Represent the extra state attribute types."""
cloudiness: int
thunder_probability: int
wind_gust_speed: float

View file

@ -1160,9 +1160,6 @@ ignore_errors = true
[mypy-homeassistant.components.smarty.*]
ignore_errors = true
[mypy-homeassistant.components.smhi.*]
ignore_errors = true
[mypy-homeassistant.components.solaredge.*]
ignore_errors = true

View file

@ -188,7 +188,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.smartthings.*",
"homeassistant.components.smarttub.*",
"homeassistant.components.smarty.*",
"homeassistant.components.smhi.*",
"homeassistant.components.solaredge.*",
"homeassistant.components.solarlog.*",
"homeassistant.components.somfy.*",