Improve meteo_france typing (#107863)
parent
7e28c788cb
commit
b1f1ecb40a
|
@ -4,6 +4,7 @@ import logging
|
||||||
|
|
||||||
from meteofrance_api.client import MeteoFranceClient
|
from meteofrance_api.client import MeteoFranceClient
|
||||||
from meteofrance_api.helpers import is_valid_warning_department
|
from meteofrance_api.helpers import is_valid_warning_department
|
||||||
|
from meteofrance_api.model import CurrentPhenomenons, Forecast, Rain
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||||
|
@ -79,17 +80,17 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
latitude = entry.data[CONF_LATITUDE]
|
latitude = entry.data[CONF_LATITUDE]
|
||||||
longitude = entry.data[CONF_LONGITUDE]
|
longitude = entry.data[CONF_LONGITUDE]
|
||||||
|
|
||||||
async def _async_update_data_forecast_forecast():
|
async def _async_update_data_forecast_forecast() -> Forecast:
|
||||||
"""Fetch data from API endpoint."""
|
"""Fetch data from API endpoint."""
|
||||||
return await hass.async_add_executor_job(
|
return await hass.async_add_executor_job(
|
||||||
client.get_forecast, latitude, longitude
|
client.get_forecast, latitude, longitude
|
||||||
)
|
)
|
||||||
|
|
||||||
async def _async_update_data_rain():
|
async def _async_update_data_rain() -> Rain:
|
||||||
"""Fetch data from API endpoint."""
|
"""Fetch data from API endpoint."""
|
||||||
return await hass.async_add_executor_job(client.get_rain, latitude, longitude)
|
return await hass.async_add_executor_job(client.get_rain, latitude, longitude)
|
||||||
|
|
||||||
async def _async_update_data_alert():
|
async def _async_update_data_alert() -> CurrentPhenomenons:
|
||||||
"""Fetch data from API endpoint."""
|
"""Fetch data from API endpoint."""
|
||||||
return await hass.async_add_executor_job(
|
return await hass.async_add_executor_job(
|
||||||
client.get_warning_current_phenomenoms, department, 0, True
|
client.get_warning_current_phenomenoms, department, 0, True
|
||||||
|
@ -136,7 +137,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
entry.title,
|
entry.title,
|
||||||
department,
|
department,
|
||||||
)
|
)
|
||||||
if is_valid_warning_department(department):
|
if department is not None and is_valid_warning_department(department):
|
||||||
if not hass.data[DOMAIN].get(department):
|
if not hass.data[DOMAIN].get(department):
|
||||||
coordinator_alert = DataUpdateCoordinator(
|
coordinator_alert = DataUpdateCoordinator(
|
||||||
hass,
|
hass,
|
||||||
|
|
|
@ -2,14 +2,17 @@
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from meteofrance_api.client import MeteoFranceClient
|
from meteofrance_api.client import MeteoFranceClient
|
||||||
|
from meteofrance_api.model import Place
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.config_entries import SOURCE_IMPORT
|
from homeassistant.config_entries import SOURCE_IMPORT
|
||||||
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
|
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
|
|
||||||
from .const import CONF_CITY, DOMAIN
|
from .const import CONF_CITY, DOMAIN
|
||||||
|
|
||||||
|
@ -21,12 +24,16 @@ class MeteoFranceFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
|
|
||||||
VERSION = 1
|
VERSION = 1
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self) -> None:
|
||||||
"""Init MeteoFranceFlowHandler."""
|
"""Init MeteoFranceFlowHandler."""
|
||||||
self.places = []
|
self.places: list[Place] = []
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _show_setup_form(self, user_input=None, errors=None):
|
def _show_setup_form(
|
||||||
|
self,
|
||||||
|
user_input: dict[str, Any] | None = None,
|
||||||
|
errors: dict[str, str] | None = None,
|
||||||
|
) -> FlowResult:
|
||||||
"""Show the setup form to the user."""
|
"""Show the setup form to the user."""
|
||||||
|
|
||||||
if user_input is None:
|
if user_input is None:
|
||||||
|
@ -40,9 +47,11 @@ class MeteoFranceFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
errors=errors or {},
|
errors=errors or {},
|
||||||
)
|
)
|
||||||
|
|
||||||
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 initiated by the user."""
|
"""Handle a flow initiated by the user."""
|
||||||
errors = {}
|
errors: dict[str, str] = {}
|
||||||
|
|
||||||
if user_input is None:
|
if user_input is None:
|
||||||
return self._show_setup_form(user_input, errors)
|
return self._show_setup_form(user_input, errors)
|
||||||
|
@ -72,15 +81,17 @@ class MeteoFranceFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
data={CONF_LATITUDE: latitude, CONF_LONGITUDE: longitude},
|
data={CONF_LATITUDE: latitude, CONF_LONGITUDE: longitude},
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_import(self, user_input):
|
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
|
||||||
"""Import a config entry."""
|
"""Import a config entry."""
|
||||||
return await self.async_step_user(user_input)
|
return await self.async_step_user(user_input)
|
||||||
|
|
||||||
async def async_step_cities(self, user_input=None):
|
async def async_step_cities(
|
||||||
|
self, user_input: dict[str, Any] | None = None
|
||||||
|
) -> FlowResult:
|
||||||
"""Step where the user choose the city from the API search results."""
|
"""Step where the user choose the city from the API search results."""
|
||||||
if not user_input:
|
if not user_input:
|
||||||
if len(self.places) > 1 and self.source != SOURCE_IMPORT:
|
if len(self.places) > 1 and self.source != SOURCE_IMPORT:
|
||||||
places_for_form = {}
|
places_for_form: dict[str, str] = {}
|
||||||
for place in self.places:
|
for place in self.places:
|
||||||
places_for_form[_build_place_key(place)] = f"{place}"
|
places_for_form[_build_place_key(place)] = f"{place}"
|
||||||
|
|
||||||
|
@ -106,5 +117,5 @@ class MeteoFranceFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _build_place_key(place) -> str:
|
def _build_place_key(place: Place) -> str:
|
||||||
return f"{place};{place.latitude};{place.longitude}"
|
return f"{place};{place.latitude};{place.longitude}"
|
||||||
|
|
|
@ -303,7 +303,7 @@ class MeteoFranceRainSensor(MeteoFranceSensor[Rain]):
|
||||||
return dt_util.utc_from_timestamp(next_rain["dt"]) if next_rain else None
|
return dt_util.utc_from_timestamp(next_rain["dt"]) if next_rain else None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def extra_state_attributes(self):
|
def extra_state_attributes(self) -> dict[str, Any]:
|
||||||
"""Return the state attributes."""
|
"""Return the state attributes."""
|
||||||
reference_dt = self.coordinator.data.forecast[0]["dt"]
|
reference_dt = self.coordinator.data.forecast[0]["dt"]
|
||||||
return {
|
return {
|
||||||
|
@ -330,7 +330,7 @@ class MeteoFranceAlertSensor(MeteoFranceSensor[CurrentPhenomenons]):
|
||||||
self._attr_unique_id = self._attr_name
|
self._attr_unique_id = self._attr_name
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_value(self):
|
def native_value(self) -> str | None:
|
||||||
"""Return the state."""
|
"""Return the state."""
|
||||||
return get_warning_text_status_from_indice_color(
|
return get_warning_text_status_from_indice_color(
|
||||||
self.coordinator.data.get_domain_max_color()
|
self.coordinator.data.get_domain_max_color()
|
||||||
|
|
|
@ -110,7 +110,7 @@ class MeteoFranceWeather(
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def unique_id(self):
|
def unique_id(self) -> str:
|
||||||
"""Return the unique id of the sensor."""
|
"""Return the unique id of the sensor."""
|
||||||
return self._unique_id
|
return self._unique_id
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue