From f968b43f6a197afee02e3fee9bc1fba2dc9007cc Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Mon, 15 Jan 2024 09:32:30 +0100 Subject: [PATCH] Improve gdacs typing (#108040) --- homeassistant/components/gdacs/__init__.py | 41 +++++++++++-------- homeassistant/components/gdacs/config_flow.py | 8 +++- .../components/gdacs/geo_location.py | 33 ++++++++------- homeassistant/components/gdacs/sensor.py | 37 ++++++++++------- 4 files changed, 70 insertions(+), 49 deletions(-) diff --git a/homeassistant/components/gdacs/__init__.py b/homeassistant/components/gdacs/__init__.py index 1530e4712d8..0ec582f8d06 100644 --- a/homeassistant/components/gdacs/__init__.py +++ b/homeassistant/components/gdacs/__init__.py @@ -1,8 +1,13 @@ """The Global Disaster Alert and Coordination System (GDACS) integration.""" -from datetime import timedelta +from __future__ import annotations + +from collections.abc import Callable +from datetime import datetime, timedelta import logging +from aio_georss_client.status_update import StatusUpdate from aio_georss_gdacs import GdacsFeedManager +from aio_georss_gdacs.feed_entry import FeedEntry from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( @@ -50,7 +55,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload an GDACS component config entry.""" - manager = hass.data[DOMAIN][FEED].pop(entry.entry_id) + manager: GdacsFeedEntityManager = hass.data[DOMAIN][FEED].pop(entry.entry_id) await manager.async_stop() return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) @@ -58,7 +63,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: class GdacsFeedEntityManager: """Feed Entity Manager for GDACS feed.""" - def __init__(self, hass, config_entry, radius_in_km): + def __init__( + self, hass: HomeAssistant, config_entry: ConfigEntry, radius_in_km: float + ) -> None: """Initialize the Feed Entity Manager.""" self._hass = hass self._config_entry = config_entry @@ -80,18 +87,18 @@ class GdacsFeedEntityManager: ) self._config_entry_id = config_entry.entry_id self._scan_interval = timedelta(seconds=config_entry.data[CONF_SCAN_INTERVAL]) - self._track_time_remove_callback = None - self._status_info = None - self.listeners = [] + self._track_time_remove_callback: Callable[[], None] | None = None + self._status_info: StatusUpdate | None = None + self.listeners: list[Callable[[], None]] = [] - async def async_init(self): + async def async_init(self) -> None: """Schedule initial and regular updates based on configured time interval.""" await self._hass.config_entries.async_forward_entry_setups( self._config_entry, PLATFORMS ) - async def update(event_time): + async def update(event_time: datetime) -> None: """Update.""" await self.async_update() @@ -102,12 +109,12 @@ class GdacsFeedEntityManager: _LOGGER.debug("Feed entity manager initialized") - async def async_update(self): + async def async_update(self) -> None: """Refresh data.""" await self._feed_manager.update() _LOGGER.debug("Feed entity manager updated") - async def async_stop(self): + async def async_stop(self) -> None: """Stop this feed entity manager from refreshing.""" for unsub_dispatcher in self.listeners: unsub_dispatcher() @@ -117,19 +124,19 @@ class GdacsFeedEntityManager: _LOGGER.debug("Feed entity manager stopped") @callback - def async_event_new_entity(self): + def async_event_new_entity(self) -> str: """Return manager specific event to signal new entity.""" return f"gdacs_new_geolocation_{self._config_entry_id}" - def get_entry(self, external_id): + def get_entry(self, external_id: str) -> FeedEntry | None: """Get feed entry by external id.""" return self._feed_manager.feed_entries.get(external_id) - def status_info(self): + def status_info(self) -> StatusUpdate | None: """Return latest status update info received.""" return self._status_info - async def _generate_entity(self, external_id): + async def _generate_entity(self, external_id: str) -> None: """Generate new entity.""" async_dispatcher_send( self._hass, @@ -139,15 +146,15 @@ class GdacsFeedEntityManager: external_id, ) - async def _update_entity(self, external_id): + async def _update_entity(self, external_id: str) -> None: """Update entity.""" async_dispatcher_send(self._hass, f"gdacs_update_{external_id}") - async def _remove_entity(self, external_id): + async def _remove_entity(self, external_id: str) -> None: """Remove entity.""" async_dispatcher_send(self._hass, f"gdacs_delete_{external_id}") - async def _status_update(self, status_info): + async def _status_update(self, status_info: StatusUpdate) -> None: """Propagate status update.""" _LOGGER.debug("Status update received: %s", status_info) self._status_info = status_info diff --git a/homeassistant/components/gdacs/config_flow.py b/homeassistant/components/gdacs/config_flow.py index b59b3bcc775..acc3bbc1991 100644 --- a/homeassistant/components/gdacs/config_flow.py +++ b/homeassistant/components/gdacs/config_flow.py @@ -1,5 +1,6 @@ """Config flow to configure the GDACS integration.""" import logging +from typing import Any import voluptuous as vol @@ -10,6 +11,7 @@ from homeassistant.const import ( CONF_RADIUS, CONF_SCAN_INTERVAL, ) +from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers import config_validation as cv from .const import CONF_CATEGORIES, DEFAULT_RADIUS, DEFAULT_SCAN_INTERVAL, DOMAIN @@ -24,13 +26,15 @@ _LOGGER = logging.getLogger(__name__) class GdacsFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): """Handle a GDACS config flow.""" - async def _show_form(self, errors=None): + async def _show_form(self, errors: dict[str, str] | None = None) -> FlowResult: """Show the form to the user.""" return self.async_show_form( step_id="user", data_schema=DATA_SCHEMA, 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 the start of the config flow.""" _LOGGER.debug("User input: %s", user_input) if not user_input: diff --git a/homeassistant/components/gdacs/geo_location.py b/homeassistant/components/gdacs/geo_location.py index 1d3dabc464c..5c4fb9d33bc 100644 --- a/homeassistant/components/gdacs/geo_location.py +++ b/homeassistant/components/gdacs/geo_location.py @@ -2,10 +2,10 @@ from __future__ import annotations from collections.abc import Callable +from datetime import datetime import logging from typing import Any -from aio_georss_gdacs import GdacsFeedManager from aio_georss_gdacs.feed_entry import GdacsFeedEntry from homeassistant.components.geo_location import GeolocationEvent @@ -58,7 +58,7 @@ async def async_setup_entry( @callback def async_add_geolocation( - feed_manager: GdacsFeedManager, integration_id: str, external_id: str + feed_manager: GdacsFeedEntityManager, integration_id: str, external_id: str ) -> None: """Add geolocation entity from feed.""" new_entity = GdacsEvent(feed_manager, integration_id, external_id) @@ -83,25 +83,28 @@ class GdacsEvent(GeolocationEvent): _attr_source = SOURCE def __init__( - self, feed_manager: GdacsFeedManager, integration_id: str, external_id: str + self, + feed_manager: GdacsFeedEntityManager, + integration_id: str, + external_id: str, ) -> None: """Initialize entity with data from feed entry.""" self._feed_manager = feed_manager self._external_id = external_id self._attr_unique_id = f"{integration_id}_{external_id}" self._attr_unit_of_measurement = UnitOfLength.KILOMETERS - self._alert_level = None - self._country = None - self._description = None - self._duration_in_week = None - self._event_type_short = None - self._event_type = None - self._from_date = None - self._to_date = None - self._population = None - self._severity = None - self._vulnerability = None - self._version = None + self._alert_level: str | None = None + self._country: str | None = None + self._description: str | None = None + self._duration_in_week: int | None = None + self._event_type_short: str | None = None + self._event_type: str | None = None + self._from_date: datetime | None = None + self._to_date: datetime | None = None + self._population: str | None = None + self._severity: str | None = None + self._vulnerability: str | float | None = None + self._version: int | None = None self._remove_signal_delete: Callable[[], None] self._remove_signal_update: Callable[[], None] diff --git a/homeassistant/components/gdacs/sensor.py b/homeassistant/components/gdacs/sensor.py index 8a0a0113ced..8039d5274ed 100644 --- a/homeassistant/components/gdacs/sensor.py +++ b/homeassistant/components/gdacs/sensor.py @@ -2,7 +2,11 @@ from __future__ import annotations from collections.abc import Callable +from datetime import datetime import logging +from typing import Any + +from aio_georss_client.status_update import StatusUpdate from homeassistant.components.sensor import SensorEntity from homeassistant.config_entries import ConfigEntry @@ -12,6 +16,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util import dt as dt_util +from . import GdacsFeedEntityManager from .const import DEFAULT_ICON, DOMAIN, FEED _LOGGER = logging.getLogger(__name__) @@ -34,7 +39,7 @@ async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback ) -> None: """Set up the GDACS Feed platform.""" - manager = hass.data[DOMAIN][FEED][entry.entry_id] + manager: GdacsFeedEntityManager = hass.data[DOMAIN][FEED][entry.entry_id] sensor = GdacsSensor(entry, manager) async_add_entities([sensor]) @@ -48,20 +53,22 @@ class GdacsSensor(SensorEntity): _attr_has_entity_name = True _attr_name = None - def __init__(self, config_entry: ConfigEntry, manager) -> None: + def __init__( + self, config_entry: ConfigEntry, manager: GdacsFeedEntityManager + ) -> None: """Initialize entity.""" assert config_entry.unique_id self._config_entry_id = config_entry.entry_id self._attr_unique_id = config_entry.unique_id self._manager = manager - self._status = None - self._last_update = None - self._last_update_successful = None - self._last_timestamp = None - self._total = None - self._created = None - self._updated = None - self._removed = None + self._status: str | None = None + self._last_update: datetime | None = None + self._last_update_successful: datetime | None = None + self._last_timestamp: datetime | None = None + self._total: int | None = None + self._created: int | None = None + self._updated: int | None = None + self._removed: int | None = None self._remove_signal_status: Callable[[], None] | None = None self._attr_device_info = DeviceInfo( identifiers={(DOMAIN, config_entry.unique_id)}, @@ -86,7 +93,7 @@ class GdacsSensor(SensorEntity): self._remove_signal_status() @callback - def _update_status_callback(self): + def _update_status_callback(self) -> None: """Call status update method.""" _LOGGER.debug("Received status update for %s", self._config_entry_id) self.async_schedule_update_ha_state(True) @@ -99,7 +106,7 @@ class GdacsSensor(SensorEntity): if status_info: self._update_from_status_info(status_info) - def _update_from_status_info(self, status_info): + def _update_from_status_info(self, status_info: StatusUpdate) -> None: """Update the internal state from the provided information.""" self._status = status_info.status self._last_update = ( @@ -118,14 +125,14 @@ class GdacsSensor(SensorEntity): self._removed = status_info.removed @property - def native_value(self): + def native_value(self) -> int | None: """Return the state of the sensor.""" return self._total @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the device state attributes.""" - attributes = {} + attributes: dict[str, Any] = {} for key, value in ( (ATTR_STATUS, self._status), (ATTR_LAST_UPDATE, self._last_update),