Use instance attributes in usgs_earthquakes_feed (#74403)

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