Use instance attributes in nsw_rural_fire_service_feed (#74398)

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

View File

@ -1,15 +1,19 @@
"""Support for NSW Rural Fire Service Feeds.""" """Support for NSW Rural Fire Service Feeds."""
from __future__ import annotations from __future__ import annotations
from datetime import timedelta from collections.abc import Callable
from datetime import datetime, timedelta
import logging import logging
from typing import Any
from aio_geojson_nsw_rfs_incidents import NswRuralFireServiceIncidentsFeedManager from aio_geojson_nsw_rfs_incidents import NswRuralFireServiceIncidentsFeedManager
from aio_geojson_nsw_rfs_incidents.feed_entry import (
NswRuralFireServiceIncidentsFeedEntry,
)
import voluptuous as vol import voluptuous as vol
from homeassistant.components.geo_location import PLATFORM_SCHEMA, GeolocationEvent from homeassistant.components.geo_location import PLATFORM_SCHEMA, GeolocationEvent
from homeassistant.const import ( from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_LOCATION, ATTR_LOCATION,
CONF_LATITUDE, CONF_LATITUDE,
CONF_LONGITUDE, CONF_LONGITUDE,
@ -19,7 +23,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP, EVENT_HOMEASSISTANT_STOP,
LENGTH_KILOMETERS, LENGTH_KILOMETERS,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers import aiohttp_client, config_validation as cv from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.dispatcher import ( from homeassistant.helpers.dispatcher import (
async_dispatcher_connect, async_dispatcher_connect,
@ -73,23 +77,23 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Set up the NSW Rural Fire Service Feed platform.""" """Set up the NSW Rural Fire Service Feed platform."""
scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) scan_interval: timedelta = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
coordinates = ( coordinates: tuple[float, float] = (
config.get(CONF_LATITUDE, hass.config.latitude), config.get(CONF_LATITUDE, hass.config.latitude),
config.get(CONF_LONGITUDE, hass.config.longitude), config.get(CONF_LONGITUDE, hass.config.longitude),
) )
radius_in_km = config[CONF_RADIUS] radius_in_km: float = config[CONF_RADIUS]
categories = config.get(CONF_CATEGORIES) categories: list[str] = config[CONF_CATEGORIES]
# Initialize the entity manager. # Initialize the entity manager.
manager = NswRuralFireServiceFeedEntityManager( manager = NswRuralFireServiceFeedEntityManager(
hass, async_add_entities, scan_interval, coordinates, radius_in_km, categories hass, async_add_entities, scan_interval, coordinates, radius_in_km, categories
) )
async def start_feed_manager(event): async def start_feed_manager(event: Event) -> None:
"""Start feed manager.""" """Start feed manager."""
await manager.async_init() await manager.async_init()
async def stop_feed_manager(event): async def stop_feed_manager(event: Event) -> None:
"""Stop feed manager.""" """Stop feed manager."""
await manager.async_stop() await manager.async_stop()
@ -103,13 +107,13 @@ class NswRuralFireServiceFeedEntityManager:
def __init__( def __init__(
self, self,
hass, hass: HomeAssistant,
async_add_entities, async_add_entities: AddEntitiesCallback,
scan_interval, scan_interval: timedelta,
coordinates, coordinates: tuple[float, float],
radius_in_km, radius_in_km: float,
categories, categories: list[str],
): ) -> None:
"""Initialize the Feed Entity Manager.""" """Initialize the Feed Entity Manager."""
self._hass = hass self._hass = hass
websession = aiohttp_client.async_get_clientsession(hass) websession = aiohttp_client.async_get_clientsession(hass)
@ -124,12 +128,12 @@ class NswRuralFireServiceFeedEntityManager:
) )
self._async_add_entities = async_add_entities self._async_add_entities = async_add_entities
self._scan_interval = scan_interval self._scan_interval = scan_interval
self._track_time_remove_callback = None self._track_time_remove_callback: Callable[[], None] | None = None
async def async_init(self): async def async_init(self) -> None:
"""Schedule initial and regular updates based on configured time interval.""" """Schedule initial and regular updates based on configured time interval."""
async def update(event_time): async def update(event_time: datetime) -> None:
"""Update.""" """Update."""
await self.async_update() await self.async_update()
@ -140,32 +144,34 @@ class NswRuralFireServiceFeedEntityManager:
_LOGGER.debug("Feed entity manager initialized") _LOGGER.debug("Feed entity manager initialized")
async def async_update(self): async def async_update(self) -> None:
"""Refresh data.""" """Refresh data."""
await self._feed_manager.update() await self._feed_manager.update()
_LOGGER.debug("Feed entity manager updated") _LOGGER.debug("Feed entity manager updated")
async def async_stop(self): async def async_stop(self) -> None:
"""Stop this feed entity manager from refreshing.""" """Stop this feed entity manager from refreshing."""
if self._track_time_remove_callback: if self._track_time_remove_callback:
self._track_time_remove_callback() self._track_time_remove_callback()
_LOGGER.debug("Feed entity manager stopped") _LOGGER.debug("Feed entity manager stopped")
def get_entry(self, external_id): def get_entry(
self, external_id: str
) -> NswRuralFireServiceIncidentsFeedEntry | None:
"""Get feed entry by external id.""" """Get feed entry by external id."""
return self._feed_manager.feed_entries.get(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.""" """Generate new entity."""
new_entity = NswRuralFireServiceLocationEvent(self, external_id) new_entity = NswRuralFireServiceLocationEvent(self, external_id)
# Add new entities to HA. # Add new entities to HA.
self._async_add_entities([new_entity], True) 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.""" """Update entity."""
async_dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY.format(external_id)) 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.""" """Remove entity."""
async_dispatcher_send(self._hass, SIGNAL_DELETE_ENTITY.format(external_id)) async_dispatcher_send(self._hass, SIGNAL_DELETE_ENTITY.format(external_id))
@ -173,15 +179,16 @@ class NswRuralFireServiceFeedEntityManager:
class NswRuralFireServiceLocationEvent(GeolocationEvent): class NswRuralFireServiceLocationEvent(GeolocationEvent):
"""This represents an external event with NSW Rural Fire Service data.""" """This represents an external event with NSW Rural Fire Service data."""
def __init__(self, feed_manager, external_id): _attr_should_poll = False
_attr_source = SOURCE
_attr_unit_of_measurement = LENGTH_KILOMETERS
def __init__(
self, feed_manager: NswRuralFireServiceFeedEntityManager, 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._external_id = external_id self._external_id = external_id
self._name = None
self._distance = None
self._latitude = None
self._longitude = None
self._attribution = None
self._category = None self._category = None
self._publication_date = None self._publication_date = None
self._location = None self._location = None
@ -191,10 +198,10 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent):
self._fire = None self._fire = None
self._size = None self._size = None
self._responsible_agency = None self._responsible_agency = 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."""
self._remove_signal_delete = async_dispatcher_connect( self._remove_signal_delete = async_dispatcher_connect(
self.hass, self.hass,
@ -213,34 +220,31 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent):
self._remove_signal_update() self._remove_signal_update()
@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 NSW Rural Fire Service 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: NswRuralFireServiceIncidentsFeedEntry
) -> None:
"""Update the internal state from the provided feed entry.""" """Update the internal state from the provided feed entry."""
self._name = feed_entry.title self._attr_name = feed_entry.title
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._category = feed_entry.category self._category = feed_entry.category
self._publication_date = feed_entry.publication_date self._publication_date = feed_entry.publication_date
self._location = feed_entry.location self._location = feed_entry.location
@ -252,51 +256,20 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent):
self._responsible_agency = feed_entry.responsible_agency self._responsible_agency = feed_entry.responsible_agency
@property @property
def icon(self): def icon(self) -> str:
"""Return the icon to use in the frontend.""" """Return the icon to use in the frontend."""
if self._fire: if self._fire:
return "mdi:fire" return "mdi:fire"
return "mdi:alarm-light" return "mdi:alarm-light"
@property @property
def source(self) -> str: def extra_state_attributes(self) -> dict[str, Any]:
"""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 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_CATEGORY, self._category), (ATTR_CATEGORY, self._category),
(ATTR_LOCATION, self._location), (ATTR_LOCATION, self._location),
(ATTR_ATTRIBUTION, self._attribution),
(ATTR_PUBLICATION_DATE, self._publication_date), (ATTR_PUBLICATION_DATE, self._publication_date),
(ATTR_COUNCIL_AREA, self._council_area), (ATTR_COUNCIL_AREA, self._council_area),
(ATTR_STATUS, self._status), (ATTR_STATUS, self._status),