From db111645c2270d7a595ad14c6e280538ac72afa3 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Sat, 9 Jul 2022 23:38:53 +0200 Subject: [PATCH] Use instance attributes in nsw_rural_fire_service_feed (#74398) --- .../geo_location.py | 135 +++++++----------- 1 file changed, 54 insertions(+), 81 deletions(-) diff --git a/homeassistant/components/nsw_rural_fire_service_feed/geo_location.py b/homeassistant/components/nsw_rural_fire_service_feed/geo_location.py index fe5d27feee2..553f73a128e 100644 --- a/homeassistant/components/nsw_rural_fire_service_feed/geo_location.py +++ b/homeassistant/components/nsw_rural_fire_service_feed/geo_location.py @@ -1,15 +1,19 @@ """Support for NSW Rural Fire Service 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_nsw_rfs_incidents import NswRuralFireServiceIncidentsFeedManager +from aio_geojson_nsw_rfs_incidents.feed_entry import ( + NswRuralFireServiceIncidentsFeedEntry, +) import voluptuous as vol from homeassistant.components.geo_location import PLATFORM_SCHEMA, GeolocationEvent from homeassistant.const import ( - ATTR_ATTRIBUTION, ATTR_LOCATION, CONF_LATITUDE, CONF_LONGITUDE, @@ -19,7 +23,7 @@ from homeassistant.const import ( EVENT_HOMEASSISTANT_STOP, 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.dispatcher import ( async_dispatcher_connect, @@ -73,23 +77,23 @@ async def async_setup_platform( discovery_info: DiscoveryInfoType | None = None, ) -> None: """Set up the NSW Rural Fire Service Feed platform.""" - scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) - coordinates = ( + scan_interval: timedelta = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) + coordinates: tuple[float, float] = ( config.get(CONF_LATITUDE, hass.config.latitude), config.get(CONF_LONGITUDE, hass.config.longitude), ) - radius_in_km = config[CONF_RADIUS] - categories = config.get(CONF_CATEGORIES) + radius_in_km: float = config[CONF_RADIUS] + categories: list[str] = config[CONF_CATEGORIES] # Initialize the entity manager. manager = NswRuralFireServiceFeedEntityManager( 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.""" await manager.async_init() - async def stop_feed_manager(event): + async def stop_feed_manager(event: Event) -> None: """Stop feed manager.""" await manager.async_stop() @@ -103,13 +107,13 @@ class NswRuralFireServiceFeedEntityManager: def __init__( self, - hass, - async_add_entities, - scan_interval, - coordinates, - radius_in_km, - categories, - ): + hass: HomeAssistant, + async_add_entities: AddEntitiesCallback, + scan_interval: timedelta, + coordinates: tuple[float, float], + radius_in_km: float, + categories: list[str], + ) -> None: """Initialize the Feed Entity Manager.""" self._hass = hass websession = aiohttp_client.async_get_clientsession(hass) @@ -124,12 +128,12 @@ class NswRuralFireServiceFeedEntityManager: ) self._async_add_entities = async_add_entities 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.""" - async def update(event_time): + async def update(event_time: datetime) -> None: """Update.""" await self.async_update() @@ -140,32 +144,34 @@ class NswRuralFireServiceFeedEntityManager: _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") - async def async_stop(self): + async def async_stop(self) -> None: """Stop this feed entity manager from refreshing.""" if self._track_time_remove_callback: self._track_time_remove_callback() _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.""" 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 = NswRuralFireServiceLocationEvent(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)) @@ -173,15 +179,16 @@ class NswRuralFireServiceFeedEntityManager: class NswRuralFireServiceLocationEvent(GeolocationEvent): """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.""" 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._category = None self._publication_date = None self._location = None @@ -191,10 +198,10 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent): self._fire = None self._size = None self._responsible_agency = 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, @@ -213,34 +220,31 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent): self._remove_signal_update() @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 NSW Rural Fire Service 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: NswRuralFireServiceIncidentsFeedEntry + ) -> 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._category = feed_entry.category self._publication_date = feed_entry.publication_date self._location = feed_entry.location @@ -252,51 +256,20 @@ class NswRuralFireServiceLocationEvent(GeolocationEvent): self._responsible_agency = feed_entry.responsible_agency @property - def icon(self): + def icon(self) -> str: """Return the icon to use in the frontend.""" if self._fire: return "mdi:fire" return "mdi:alarm-light" @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 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_CATEGORY, self._category), (ATTR_LOCATION, self._location), - (ATTR_ATTRIBUTION, self._attribution), (ATTR_PUBLICATION_DATE, self._publication_date), (ATTR_COUNCIL_AREA, self._council_area), (ATTR_STATUS, self._status),