diff --git a/homeassistant/components/gdacs/geo_location.py b/homeassistant/components/gdacs/geo_location.py index 5bfbd915b6c..715ac779668 100644 --- a/homeassistant/components/gdacs/geo_location.py +++ b/homeassistant/components/gdacs/geo_location.py @@ -1,12 +1,16 @@ """Geolocation support for GDACS Feed.""" from __future__ import annotations +from collections.abc import Callable 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 from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( - ATTR_ATTRIBUTION, CONF_UNIT_SYSTEM_IMPERIAL, LENGTH_KILOMETERS, LENGTH_MILES, @@ -17,6 +21,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util.unit_system import IMPERIAL_SYSTEM +from . import GdacsFeedEntityManager from .const import DEFAULT_ICON, DOMAIN, FEED _LOGGER = logging.getLogger(__name__) @@ -52,10 +57,12 @@ 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] @callback - def async_add_geolocation(feed_manager, integration_id, external_id): + def async_add_geolocation( + feed_manager: GdacsFeedManager, integration_id: str, external_id: str + ) -> None: """Add geolocation entity from feed.""" new_entity = GdacsEvent(feed_manager, integration_id, external_id) _LOGGER.debug("Adding geolocation %s", new_entity) @@ -75,16 +82,17 @@ async def async_setup_entry( class GdacsEvent(GeolocationEvent): """This represents an external event with GDACS feed data.""" - def __init__(self, feed_manager, integration_id, external_id): + _attr_should_poll = False + _attr_source = SOURCE + + def __init__( + self, feed_manager: GdacsFeedManager, integration_id: str, external_id: str + ) -> None: """Initialize entity with data from feed entry.""" self._feed_manager = feed_manager - self._integration_id = integration_id self._external_id = external_id - self._title = None - self._distance = None - self._latitude = None - self._longitude = None - self._attribution = None + self._attr_unique_id = f"{integration_id}_{external_id}" + self._attr_unit_of_measurement = LENGTH_KILOMETERS self._alert_level = None self._country = None self._description = None @@ -97,11 +105,13 @@ class GdacsEvent(GeolocationEvent): self._severity = None self._vulnerability = None self._version = None - self._remove_signal_delete = None - self._remove_signal_update = None + self._remove_signal_delete: Callable[[], None] + self._remove_signal_update: Callable[[], None] - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Call when entity is added to hass.""" + if self.hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL: + self._attr_unit_of_measurement = LENGTH_MILES self._remove_signal_delete = async_dispatcher_connect( self.hass, f"gdacs_delete_{self._external_id}", self._delete_callback ) @@ -119,43 +129,38 @@ class GdacsEvent(GeolocationEvent): entity_registry.async_remove(self.entity_id) @callback - def _delete_callback(self): + def _delete_callback(self) -> None: """Remove this entity.""" self.hass.async_create_task(self.async_remove(force_remove=True)) @callback - def _update_callback(self): + def _update_callback(self) -> None: """Call update method.""" self.async_schedule_update_ha_state(True) - @property - def should_poll(self): - """No polling needed for GDACS feed location events.""" - return False - - async def async_update(self): + async def async_update(self) -> None: """Update this entity from the data held in the feed manager.""" _LOGGER.debug("Updating %s", self._external_id) feed_entry = self._feed_manager.get_entry(self._external_id) if feed_entry: self._update_from_feed(feed_entry) - def _update_from_feed(self, feed_entry): + def _update_from_feed(self, feed_entry: GdacsFeedEntry) -> None: """Update the internal state from the provided feed entry.""" if not (event_name := feed_entry.event_name): # Earthquakes usually don't have an event name. event_name = f"{feed_entry.country} ({feed_entry.event_id})" - self._title = f"{feed_entry.event_type}: {event_name}" + self._attr_name = f"{feed_entry.event_type}: {event_name}" # Convert distance if not metric system. if self.hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL: - self._distance = IMPERIAL_SYSTEM.length( + self._attr_distance = IMPERIAL_SYSTEM.length( feed_entry.distance_to_home, LENGTH_KILOMETERS ) else: - self._distance = feed_entry.distance_to_home - self._latitude = feed_entry.coordinates[0] - self._longitude = feed_entry.coordinates[1] - self._attribution = feed_entry.attribution + self._attr_distance = feed_entry.distance_to_home + self._attr_latitude = feed_entry.coordinates[0] + self._attr_longitude = feed_entry.coordinates[1] + self._attr_attribution = feed_entry.attribution self._alert_level = feed_entry.alert_level self._country = feed_entry.country self._description = feed_entry.title @@ -173,57 +178,19 @@ class GdacsEvent(GeolocationEvent): self._version = feed_entry.version @property - def unique_id(self) -> str | None: - """Return a unique ID containing latitude/longitude and external id.""" - return f"{self._integration_id}_{self._external_id}" - - @property - def icon(self): + def icon(self) -> str: """Return the icon to use in the frontend, if any.""" if self._event_type_short and self._event_type_short in ICONS: return ICONS[self._event_type_short] return DEFAULT_ICON @property - def source(self) -> str: - """Return source value of this external event.""" - return SOURCE - - @property - def name(self) -> str | None: - """Return the name of the entity.""" - return self._title - - @property - def distance(self) -> float | None: - """Return distance value of this external event.""" - return self._distance - - @property - def latitude(self) -> float | None: - """Return latitude value of this external event.""" - return self._latitude - - @property - def longitude(self) -> float | None: - """Return longitude value of this external event.""" - return self._longitude - - @property - def unit_of_measurement(self): - """Return the unit of measurement.""" - if self.hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL: - return LENGTH_MILES - return LENGTH_KILOMETERS - - @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the device state attributes.""" attributes = {} for key, value in ( (ATTR_EXTERNAL_ID, self._external_id), (ATTR_DESCRIPTION, self._description), - (ATTR_ATTRIBUTION, self._attribution), (ATTR_EVENT_TYPE, self._event_type), (ATTR_ALERT_LEVEL, self._alert_level), (ATTR_COUNTRY, self._country),