mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 05:07:41 +00:00
Use instance attributes in gdacs (#74400)
This commit is contained in:
parent
36bb34f391
commit
bb974ebf7e
@ -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),
|
||||
|
Loading…
x
Reference in New Issue
Block a user