diff --git a/homeassistant/components/usgs_earthquakes_feed/geo_location.py b/homeassistant/components/usgs_earthquakes_feed/geo_location.py index 6e3eb9b2337..c82705174fe 100644 --- a/homeassistant/components/usgs_earthquakes_feed/geo_location.py +++ b/homeassistant/components/usgs_earthquakes_feed/geo_location.py @@ -1,15 +1,19 @@ """Support for U.S. Geological Survey Earthquake Hazards Program Feeds.""" from __future__ import annotations -from datetime import timedelta +from collections.abc import Callable +from datetime import datetime, timedelta import logging +from typing import Any from aio_geojson_usgs_earthquakes import UsgsEarthquakeHazardsProgramFeedManager +from aio_geojson_usgs_earthquakes.feed_entry import ( + UsgsEarthquakeHazardsProgramFeedEntry, +) import voluptuous as vol from homeassistant.components.geo_location import PLATFORM_SCHEMA, GeolocationEvent from homeassistant.const import ( - ATTR_ATTRIBUTION, ATTR_TIME, CONF_LATITUDE, CONF_LONGITUDE, @@ -96,14 +100,14 @@ async def async_setup_platform( discovery_info: DiscoveryInfoType | None = None, ) -> None: """Set up the USGS Earthquake Hazards Program Feed platform.""" - scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) - feed_type = config[CONF_FEED_TYPE] - coordinates = ( + scan_interval: timedelta = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) + feed_type: str = config[CONF_FEED_TYPE] + coordinates: tuple[float, float] = ( config.get(CONF_LATITUDE, hass.config.latitude), config.get(CONF_LONGITUDE, hass.config.longitude), ) - radius_in_km = config[CONF_RADIUS] - minimum_magnitude = config[CONF_MINIMUM_MAGNITUDE] + radius_in_km: float = config[CONF_RADIUS] + minimum_magnitude: float = config[CONF_MINIMUM_MAGNITUDE] # Initialize the entity manager. manager = UsgsEarthquakesFeedEntityManager( hass, @@ -128,14 +132,14 @@ class UsgsEarthquakesFeedEntityManager: def __init__( self, - hass, - async_add_entities, - scan_interval, - coordinates, - feed_type, - radius_in_km, - minimum_magnitude, - ): + hass: HomeAssistant, + async_add_entities: AddEntitiesCallback, + scan_interval: timedelta, + coordinates: tuple[float, float], + feed_type: str, + radius_in_km: float, + minimum_magnitude: float, + ) -> None: """Initialize the Feed Entity Manager.""" self._hass = hass @@ -153,10 +157,10 @@ class UsgsEarthquakesFeedEntityManager: self._async_add_entities = async_add_entities self._scan_interval = scan_interval - async def async_init(self): + async def async_init(self) -> None: """Schedule initial and regular updates based on configured time interval.""" - async def update(event_time): + async def update(event_time: datetime) -> None: """Update.""" await self.async_update() @@ -164,26 +168,28 @@ class UsgsEarthquakesFeedEntityManager: async_track_time_interval(self._hass, update, self._scan_interval) _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") - def get_entry(self, external_id): + def get_entry( + self, external_id: str + ) -> UsgsEarthquakeHazardsProgramFeedEntry | None: """Get feed entry by external id.""" return self._feed_manager.feed_entries.get(external_id) - async def _generate_entity(self, external_id): + async def _generate_entity(self, external_id: str) -> None: """Generate new entity.""" new_entity = UsgsEarthquakesEvent(self, external_id) # Add new entities to HA. self._async_add_entities([new_entity], True) - async def _update_entity(self, external_id): + async def _update_entity(self, external_id: str) -> None: """Update entity.""" async_dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY.format(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, SIGNAL_DELETE_ENTITY.format(external_id)) @@ -191,15 +197,17 @@ class UsgsEarthquakesFeedEntityManager: class UsgsEarthquakesEvent(GeolocationEvent): """This represents an external event with USGS Earthquake data.""" - def __init__(self, feed_manager, external_id): + _attr_icon = "mdi:pulse" + _attr_should_poll = False + _attr_source = SOURCE + _attr_unit_of_measurement = DEFAULT_UNIT_OF_MEASUREMENT + + def __init__( + self, feed_manager: UsgsEarthquakesFeedEntityManager, external_id: str + ) -> None: """Initialize entity with data from feed entry.""" self._feed_manager = feed_manager self._external_id = external_id - self._name = None - self._distance = None - self._latitude = None - self._longitude = None - self._attribution = None self._place = None self._magnitude = None self._time = None @@ -207,10 +215,10 @@ class UsgsEarthquakesEvent(GeolocationEvent): self._status = None self._type = None self._alert = 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.""" self._remove_signal_delete = async_dispatcher_connect( self.hass, @@ -224,36 +232,33 @@ class UsgsEarthquakesEvent(GeolocationEvent): ) @callback - def _delete_callback(self): + def _delete_callback(self) -> None: """Remove this entity.""" self._remove_signal_delete() self._remove_signal_update() 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 USGS Earthquake 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: UsgsEarthquakeHazardsProgramFeedEntry + ) -> None: """Update the internal state from the provided feed entry.""" - self._name = feed_entry.title - 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_name = feed_entry.title + 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._place = feed_entry.place self._magnitude = feed_entry.magnitude self._time = feed_entry.time @@ -263,42 +268,7 @@ class UsgsEarthquakesEvent(GeolocationEvent): self._alert = feed_entry.alert @property - def icon(self): - """Return the icon to use in the frontend.""" - return "mdi:pulse" - - @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._name - - @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.""" - return DEFAULT_UNIT_OF_MEASUREMENT - - @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the device state attributes.""" attributes = {} for key, value in ( @@ -310,7 +280,6 @@ class UsgsEarthquakesEvent(GeolocationEvent): (ATTR_STATUS, self._status), (ATTR_TYPE, self._type), (ATTR_ALERT, self._alert), - (ATTR_ATTRIBUTION, self._attribution), ): if value or isinstance(value, bool): attributes[key] = value