Use instance attributes in geonetnz_quakes (#74401)

This commit is contained in:
epenet 2022-07-09 23:14:07 +02:00 committed by GitHub
parent 3922141f5c
commit d49c58cf87
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,12 +1,15 @@
"""Geolocation support for GeoNet NZ Quakes Feeds.""" """Geolocation support for GeoNet NZ Quakes Feeds."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable
import logging import logging
from typing import Any
from aio_geojson_geonetnz_quakes.feed_entry import GeonetnzQuakesFeedEntry
from homeassistant.components.geo_location import GeolocationEvent from homeassistant.components.geo_location import GeolocationEvent
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_TIME, ATTR_TIME,
CONF_UNIT_SYSTEM_IMPERIAL, CONF_UNIT_SYSTEM_IMPERIAL,
LENGTH_KILOMETERS, LENGTH_KILOMETERS,
@ -18,6 +21,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.unit_system import IMPERIAL_SYSTEM from homeassistant.util.unit_system import IMPERIAL_SYSTEM
from . import GeonetnzQuakesFeedEntityManager
from .const import DOMAIN, FEED from .const import DOMAIN, FEED
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -40,10 +44,14 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up the GeoNet NZ Quakes Feed platform.""" """Set up the GeoNet NZ Quakes Feed platform."""
manager = hass.data[DOMAIN][FEED][entry.entry_id] manager: GeonetnzQuakesFeedEntityManager = hass.data[DOMAIN][FEED][entry.entry_id]
@callback @callback
def async_add_geolocation(feed_manager, integration_id, external_id): def async_add_geolocation(
feed_manager: GeonetnzQuakesFeedEntityManager,
integration_id: str,
external_id: str,
) -> None:
"""Add geolocation entity from feed.""" """Add geolocation entity from feed."""
new_entity = GeonetnzQuakesEvent(feed_manager, integration_id, external_id) new_entity = GeonetnzQuakesEvent(feed_manager, integration_id, external_id)
_LOGGER.debug("Adding geolocation %s", new_entity) _LOGGER.debug("Adding geolocation %s", new_entity)
@ -63,27 +71,34 @@ async def async_setup_entry(
class GeonetnzQuakesEvent(GeolocationEvent): class GeonetnzQuakesEvent(GeolocationEvent):
"""This represents an external event with GeoNet NZ Quakes feed data.""" """This represents an external event with GeoNet NZ Quakes feed data."""
def __init__(self, feed_manager, integration_id, external_id): _attr_icon = "mdi:pulse"
_attr_should_poll = False
_attr_source = SOURCE
def __init__(
self,
feed_manager: GeonetnzQuakesFeedEntityManager,
integration_id: str,
external_id: str,
) -> None:
"""Initialize entity with data from feed entry.""" """Initialize entity with data from feed entry."""
self._feed_manager = feed_manager self._feed_manager = feed_manager
self._integration_id = integration_id
self._external_id = external_id self._external_id = external_id
self._title = None self._attr_unique_id = f"{integration_id}_{external_id}"
self._distance = None self._attr_unit_of_measurement = LENGTH_KILOMETERS
self._latitude = None
self._longitude = None
self._attribution = None
self._depth = None self._depth = None
self._locality = None self._locality = None
self._magnitude = None self._magnitude = None
self._mmi = None self._mmi = None
self._quality = None self._quality = None
self._time = None self._time = None
self._remove_signal_delete = None self._remove_signal_delete: Callable[[], None]
self._remove_signal_update = 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.""" """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._remove_signal_delete = async_dispatcher_connect(
self.hass, self.hass,
f"geonetnz_quakes_delete_{self._external_id}", f"geonetnz_quakes_delete_{self._external_id}",
@ -105,40 +120,35 @@ class GeonetnzQuakesEvent(GeolocationEvent):
entity_registry.async_remove(self.entity_id) entity_registry.async_remove(self.entity_id)
@callback @callback
def _delete_callback(self): def _delete_callback(self) -> None:
"""Remove this entity.""" """Remove this entity."""
self.hass.async_create_task(self.async_remove(force_remove=True)) self.hass.async_create_task(self.async_remove(force_remove=True))
@callback @callback
def _update_callback(self): def _update_callback(self) -> None:
"""Call update method.""" """Call update method."""
self.async_schedule_update_ha_state(True) self.async_schedule_update_ha_state(True)
@property async def async_update(self) -> None:
def should_poll(self):
"""No polling needed for GeoNet NZ Quakes feed location events."""
return False
async def async_update(self):
"""Update this entity from the data held in the feed manager.""" """Update this entity from the data held in the feed manager."""
_LOGGER.debug("Updating %s", self._external_id) _LOGGER.debug("Updating %s", self._external_id)
feed_entry = self._feed_manager.get_entry(self._external_id) feed_entry = self._feed_manager.get_entry(self._external_id)
if feed_entry: if feed_entry:
self._update_from_feed(feed_entry) self._update_from_feed(feed_entry)
def _update_from_feed(self, feed_entry): def _update_from_feed(self, feed_entry: GeonetnzQuakesFeedEntry) -> None:
"""Update the internal state from the provided feed entry.""" """Update the internal state from the provided feed entry."""
self._title = feed_entry.title self._attr_name = feed_entry.title
# Convert distance if not metric system. # Convert distance if not metric system.
if self.hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL: 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 feed_entry.distance_to_home, LENGTH_KILOMETERS
) )
else: else:
self._distance = feed_entry.distance_to_home self._attr_distance = feed_entry.distance_to_home
self._latitude = feed_entry.coordinates[0] self._attr_latitude = feed_entry.coordinates[0]
self._longitude = feed_entry.coordinates[1] self._attr_longitude = feed_entry.coordinates[1]
self._attribution = feed_entry.attribution self._attr_attribution = feed_entry.attribution
self._depth = feed_entry.depth self._depth = feed_entry.depth
self._locality = feed_entry.locality self._locality = feed_entry.locality
self._magnitude = feed_entry.magnitude self._magnitude = feed_entry.magnitude
@ -147,54 +157,11 @@ class GeonetnzQuakesEvent(GeolocationEvent):
self._time = feed_entry.time self._time = feed_entry.time
@property @property
def unique_id(self) -> str | None: def extra_state_attributes(self) -> dict[str, Any]:
"""Return a unique ID containing latitude/longitude and external id."""
return f"{self._integration_id}_{self._external_id}"
@property
def icon(self):
"""Return the icon to use in the frontend, if any."""
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._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):
"""Return the device state attributes.""" """Return the device state attributes."""
attributes = {} attributes = {}
for key, value in ( for key, value in (
(ATTR_EXTERNAL_ID, self._external_id), (ATTR_EXTERNAL_ID, self._external_id),
(ATTR_ATTRIBUTION, self._attribution),
(ATTR_DEPTH, self._depth), (ATTR_DEPTH, self._depth),
(ATTR_LOCALITY, self._locality), (ATTR_LOCALITY, self._locality),
(ATTR_MAGNITUDE, self._magnitude), (ATTR_MAGNITUDE, self._magnitude),