mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 21:27:38 +00:00
Improve type hints in nws (#83173)
This commit is contained in:
parent
a818ef6411
commit
c3c9ed6835
@ -37,7 +37,7 @@ FAILED_SCAN_INTERVAL = datetime.timedelta(minutes=1)
|
|||||||
DEBOUNCE_TIME = 60 # in seconds
|
DEBOUNCE_TIME = 60 # in seconds
|
||||||
|
|
||||||
|
|
||||||
def base_unique_id(latitude, longitude):
|
def base_unique_id(latitude: float, longitude: float) -> str:
|
||||||
"""Return unique id for entries in configuration."""
|
"""Return unique id for entries in configuration."""
|
||||||
return f"{latitude}_{longitude}"
|
return f"{latitude}_{longitude}"
|
||||||
|
|
||||||
@ -174,7 +174,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
return unload_ok
|
return unload_ok
|
||||||
|
|
||||||
|
|
||||||
def device_info(latitude, longitude) -> DeviceInfo:
|
def device_info(latitude: float, longitude: float) -> DeviceInfo:
|
||||||
"""Return device registry information."""
|
"""Return device registry information."""
|
||||||
return DeviceInfo(
|
return DeviceInfo(
|
||||||
entry_type=DeviceEntryType.SERVICE,
|
entry_type=DeviceEntryType.SERVICE,
|
||||||
|
@ -1,5 +1,8 @@
|
|||||||
"""Config flow for National Weather Service (NWS) integration."""
|
"""Config flow for National Weather Service (NWS) integration."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
from pynws import SimpleNWS
|
from pynws import SimpleNWS
|
||||||
@ -7,6 +10,7 @@ import voluptuous as vol
|
|||||||
|
|
||||||
from homeassistant import config_entries, core, exceptions
|
from homeassistant import config_entries, core, exceptions
|
||||||
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
|
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
|
||||||
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
from homeassistant.helpers import config_validation as cv
|
from homeassistant.helpers import config_validation as cv
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
|
|
||||||
@ -16,7 +20,9 @@ from .const import CONF_STATION, DOMAIN
|
|||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def validate_input(hass: core.HomeAssistant, data):
|
async def validate_input(
|
||||||
|
hass: core.HomeAssistant, data: dict[str, Any]
|
||||||
|
) -> dict[str, str]:
|
||||||
"""Validate the user input allows us to connect.
|
"""Validate the user input allows us to connect.
|
||||||
|
|
||||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||||
@ -44,9 +50,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
VERSION = 1
|
VERSION = 1
|
||||||
|
|
||||||
async def async_step_user(self, user_input=None):
|
async def async_step_user(
|
||||||
|
self, user_input: dict[str, Any] | None = None
|
||||||
|
) -> FlowResult:
|
||||||
"""Handle the initial step."""
|
"""Handle the initial step."""
|
||||||
errors = {}
|
errors: dict[str, str] = {}
|
||||||
if user_input is not None:
|
if user_input is not None:
|
||||||
await self.async_set_unique_id(
|
await self.async_set_unique_id(
|
||||||
base_unique_id(user_input[CONF_LATITUDE], user_input[CONF_LONGITUDE])
|
base_unique_id(user_input[CONF_LATITUDE], user_input[CONF_LONGITUDE])
|
||||||
|
@ -28,7 +28,7 @@ ATTRIBUTION = "Data from National Weather Service/NOAA"
|
|||||||
ATTR_FORECAST_DETAILED_DESCRIPTION = "detailed_description"
|
ATTR_FORECAST_DETAILED_DESCRIPTION = "detailed_description"
|
||||||
ATTR_FORECAST_DAYTIME = "daytime"
|
ATTR_FORECAST_DAYTIME = "daytime"
|
||||||
|
|
||||||
CONDITION_CLASSES = {
|
CONDITION_CLASSES: dict[str, list[str]] = {
|
||||||
ATTR_CONDITION_EXCEPTIONAL: [
|
ATTR_CONDITION_EXCEPTIONAL: [
|
||||||
"Tornado",
|
"Tornado",
|
||||||
"Hurricane conditions",
|
"Hurricane conditions",
|
||||||
|
@ -2,6 +2,8 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
from types import MappingProxyType
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from pynws import SimpleNWS
|
from pynws import SimpleNWS
|
||||||
|
|
||||||
@ -174,11 +176,11 @@ class NWSSensor(CoordinatorEntity[NwsDataUpdateCoordinator], SensorEntity):
|
|||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
entry_data,
|
entry_data: MappingProxyType[str, Any],
|
||||||
hass_data,
|
hass_data: dict[str, Any],
|
||||||
description: NWSSensorEntityDescription,
|
description: NWSSensorEntityDescription,
|
||||||
station,
|
station: str,
|
||||||
):
|
) -> None:
|
||||||
"""Initialise the platform with a data instance."""
|
"""Initialise the platform with a data instance."""
|
||||||
super().__init__(hass_data[COORDINATOR_OBSERVATION])
|
super().__init__(hass_data[COORDINATOR_OBSERVATION])
|
||||||
self._nws: SimpleNWS = hass_data[NWS_DATA]
|
self._nws: SimpleNWS = hass_data[NWS_DATA]
|
||||||
@ -191,7 +193,7 @@ class NWSSensor(CoordinatorEntity[NwsDataUpdateCoordinator], SensorEntity):
|
|||||||
self._attr_native_unit_of_measurement = description.unit_convert
|
self._attr_native_unit_of_measurement = description.unit_convert
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_value(self):
|
def native_value(self) -> float | None:
|
||||||
"""Return the state."""
|
"""Return the state."""
|
||||||
value = self._nws.observation.get(self.entity_description.key)
|
value = self._nws.observation.get(self.entity_description.key)
|
||||||
if value is None:
|
if value is None:
|
||||||
@ -224,7 +226,7 @@ class NWSSensor(CoordinatorEntity[NwsDataUpdateCoordinator], SensorEntity):
|
|||||||
return value
|
return value
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def unique_id(self):
|
def unique_id(self) -> str:
|
||||||
"""Return a unique_id for this entity."""
|
"""Return a unique_id for this entity."""
|
||||||
return f"{base_unique_id(self._latitude, self._longitude)}_{self.entity_description.key}"
|
return f"{base_unique_id(self._latitude, self._longitude)}_{self.entity_description.key}"
|
||||||
|
|
||||||
|
@ -1,4 +1,9 @@
|
|||||||
"""Support for NWS weather service."""
|
"""Support for NWS weather service."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from types import MappingProxyType
|
||||||
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
from homeassistant.components.weather import (
|
from homeassistant.components.weather import (
|
||||||
ATTR_CONDITION_CLEAR_NIGHT,
|
ATTR_CONDITION_CLEAR_NIGHT,
|
||||||
ATTR_CONDITION_SUNNY,
|
ATTR_CONDITION_SUNNY,
|
||||||
@ -8,6 +13,7 @@ from homeassistant.components.weather import (
|
|||||||
ATTR_FORECAST_PRECIPITATION_PROBABILITY,
|
ATTR_FORECAST_PRECIPITATION_PROBABILITY,
|
||||||
ATTR_FORECAST_TIME,
|
ATTR_FORECAST_TIME,
|
||||||
ATTR_FORECAST_WIND_BEARING,
|
ATTR_FORECAST_WIND_BEARING,
|
||||||
|
Forecast,
|
||||||
WeatherEntity,
|
WeatherEntity,
|
||||||
)
|
)
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
@ -24,6 +30,7 @@ from homeassistant.helpers.entity import DeviceInfo
|
|||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.util.dt import utcnow
|
from homeassistant.util.dt import utcnow
|
||||||
from homeassistant.util.unit_conversion import SpeedConverter, TemperatureConverter
|
from homeassistant.util.unit_conversion import SpeedConverter, TemperatureConverter
|
||||||
|
from homeassistant.util.unit_system import UnitSystem
|
||||||
|
|
||||||
from . import base_unique_id, device_info
|
from . import base_unique_id, device_info
|
||||||
from .const import (
|
from .const import (
|
||||||
@ -45,14 +52,16 @@ from .const import (
|
|||||||
PARALLEL_UPDATES = 0
|
PARALLEL_UPDATES = 0
|
||||||
|
|
||||||
|
|
||||||
def convert_condition(time, weather):
|
def convert_condition(
|
||||||
|
time: str, weather: tuple[tuple[str, int | None], ...]
|
||||||
|
) -> tuple[str, int | None]:
|
||||||
"""
|
"""
|
||||||
Convert NWS codes to HA condition.
|
Convert NWS codes to HA condition.
|
||||||
|
|
||||||
Choose first condition in CONDITION_CLASSES that exists in weather code.
|
Choose first condition in CONDITION_CLASSES that exists in weather code.
|
||||||
If no match is found, return first condition from NWS
|
If no match is found, return first condition from NWS
|
||||||
"""
|
"""
|
||||||
conditions = [w[0] for w in weather]
|
conditions: list[str] = [w[0] for w in weather]
|
||||||
prec_probs = [w[1] or 0 for w in weather]
|
prec_probs = [w[1] or 0 for w in weather]
|
||||||
|
|
||||||
# Choose condition with highest priority.
|
# Choose condition with highest priority.
|
||||||
@ -88,12 +97,27 @@ async def async_setup_entry(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
|
||||||
|
class NWSForecast(Forecast):
|
||||||
|
"""Forecast with extra fields needed for NWS."""
|
||||||
|
|
||||||
|
detailed_description: str | None
|
||||||
|
daytime: bool | None
|
||||||
|
|
||||||
|
|
||||||
class NWSWeather(WeatherEntity):
|
class NWSWeather(WeatherEntity):
|
||||||
"""Representation of a weather condition."""
|
"""Representation of a weather condition."""
|
||||||
|
|
||||||
_attr_should_poll = False
|
_attr_should_poll = False
|
||||||
|
|
||||||
def __init__(self, entry_data, hass_data, mode, units):
|
def __init__(
|
||||||
|
self,
|
||||||
|
entry_data: MappingProxyType[str, Any],
|
||||||
|
hass_data: dict[str, Any],
|
||||||
|
mode: str,
|
||||||
|
units: UnitSystem,
|
||||||
|
) -> None:
|
||||||
"""Initialise the platform with a data instance and station name."""
|
"""Initialise the platform with a data instance and station name."""
|
||||||
self.nws = hass_data[NWS_DATA]
|
self.nws = hass_data[NWS_DATA]
|
||||||
self.latitude = entry_data[CONF_LATITUDE]
|
self.latitude = entry_data[CONF_LATITUDE]
|
||||||
@ -132,67 +156,67 @@ class NWSWeather(WeatherEntity):
|
|||||||
self.async_write_ha_state()
|
self.async_write_ha_state()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def attribution(self):
|
def attribution(self) -> str:
|
||||||
"""Return the attribution."""
|
"""Return the attribution."""
|
||||||
return ATTRIBUTION
|
return ATTRIBUTION
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def name(self):
|
def name(self) -> str:
|
||||||
"""Return the name of the station."""
|
"""Return the name of the station."""
|
||||||
return f"{self.station} {self.mode.title()}"
|
return f"{self.station} {self.mode.title()}"
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_temperature(self):
|
def native_temperature(self) -> float | None:
|
||||||
"""Return the current temperature."""
|
"""Return the current temperature."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("temperature")
|
return self.observation.get("temperature")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_temperature_unit(self):
|
def native_temperature_unit(self) -> str:
|
||||||
"""Return the current temperature unit."""
|
"""Return the current temperature unit."""
|
||||||
return UnitOfTemperature.CELSIUS
|
return UnitOfTemperature.CELSIUS
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_pressure(self):
|
def native_pressure(self) -> int | None:
|
||||||
"""Return the current pressure."""
|
"""Return the current pressure."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("seaLevelPressure")
|
return self.observation.get("seaLevelPressure")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_pressure_unit(self):
|
def native_pressure_unit(self) -> str:
|
||||||
"""Return the current pressure unit."""
|
"""Return the current pressure unit."""
|
||||||
return UnitOfPressure.PA
|
return UnitOfPressure.PA
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def humidity(self):
|
def humidity(self) -> float | None:
|
||||||
"""Return the name of the sensor."""
|
"""Return the name of the sensor."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("relativeHumidity")
|
return self.observation.get("relativeHumidity")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_wind_speed(self):
|
def native_wind_speed(self) -> float | None:
|
||||||
"""Return the current windspeed."""
|
"""Return the current windspeed."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("windSpeed")
|
return self.observation.get("windSpeed")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_wind_speed_unit(self):
|
def native_wind_speed_unit(self) -> str:
|
||||||
"""Return the current windspeed."""
|
"""Return the current windspeed."""
|
||||||
return UnitOfSpeed.KILOMETERS_PER_HOUR
|
return UnitOfSpeed.KILOMETERS_PER_HOUR
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def wind_bearing(self):
|
def wind_bearing(self) -> int | None:
|
||||||
"""Return the current wind bearing (degrees)."""
|
"""Return the current wind bearing (degrees)."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("windDirection")
|
return self.observation.get("windDirection")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def condition(self):
|
def condition(self) -> str | None:
|
||||||
"""Return current condition."""
|
"""Return current condition."""
|
||||||
weather = None
|
weather = None
|
||||||
if self.observation:
|
if self.observation:
|
||||||
@ -205,23 +229,23 @@ class NWSWeather(WeatherEntity):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_visibility(self):
|
def native_visibility(self) -> int | None:
|
||||||
"""Return visibility."""
|
"""Return visibility."""
|
||||||
if self.observation:
|
if self.observation:
|
||||||
return self.observation.get("visibility")
|
return self.observation.get("visibility")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_visibility_unit(self):
|
def native_visibility_unit(self) -> str:
|
||||||
"""Return visibility unit."""
|
"""Return visibility unit."""
|
||||||
return UnitOfLength.METERS
|
return UnitOfLength.METERS
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def forecast(self):
|
def forecast(self) -> list[Forecast] | None:
|
||||||
"""Return forecast."""
|
"""Return forecast."""
|
||||||
if self._forecast is None:
|
if self._forecast is None:
|
||||||
return None
|
return None
|
||||||
forecast = []
|
forecast: list[NWSForecast] = []
|
||||||
for forecast_entry in self._forecast:
|
for forecast_entry in self._forecast:
|
||||||
data = {
|
data = {
|
||||||
ATTR_FORECAST_DETAILED_DESCRIPTION: forecast_entry.get(
|
ATTR_FORECAST_DETAILED_DESCRIPTION: forecast_entry.get(
|
||||||
@ -262,7 +286,7 @@ class NWSWeather(WeatherEntity):
|
|||||||
return forecast
|
return forecast
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def unique_id(self):
|
def unique_id(self) -> str:
|
||||||
"""Return a unique_id for this entity."""
|
"""Return a unique_id for this entity."""
|
||||||
return f"{base_unique_id(self.latitude, self.longitude)}_{self.mode}"
|
return f"{base_unique_id(self.latitude, self.longitude)}_{self.mode}"
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user