diff --git a/homeassistant/components/geo_json_events/geo_location.py b/homeassistant/components/geo_json_events/geo_location.py index 896fb7d36da..c0d6abe694f 100644 --- a/homeassistant/components/geo_json_events/geo_location.py +++ b/homeassistant/components/geo_json_events/geo_location.py @@ -1,10 +1,13 @@ """Support for generic GeoJSON events.""" 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_generic_client import GenericFeedManager +from aio_geojson_generic_client.feed_entry import GenericFeedEntry import voluptuous as vol from homeassistant.components.geo_location import PLATFORM_SCHEMA, GeolocationEvent @@ -17,7 +20,7 @@ from homeassistant.const import ( EVENT_HOMEASSISTANT_START, LENGTH_KILOMETERS, ) -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import Event, HomeAssistant, callback from homeassistant.helpers import aiohttp_client import homeassistant.helpers.config_validation as cv from homeassistant.helpers.dispatcher import ( @@ -55,20 +58,20 @@ async def async_setup_platform( discovery_info: DiscoveryInfoType | None = None, ) -> None: """Set up the GeoJSON Events platform.""" - url = config[CONF_URL] - scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL) - coordinates = ( + url: str = config[CONF_URL] + 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] + radius_in_km: float = config[CONF_RADIUS] # Initialize the entity manager. manager = GeoJsonFeedEntityManager( hass, async_add_entities, scan_interval, coordinates, url, radius_in_km ) await manager.async_init() - async def start_feed_manager(event=None): + async def start_feed_manager(event: Event) -> None: """Start feed manager.""" await manager.async_update() @@ -79,8 +82,14 @@ class GeoJsonFeedEntityManager: """Feed Entity Manager for GeoJSON feeds.""" def __init__( - self, hass, async_add_entities, scan_interval, coordinates, url, radius_in_km - ): + self, + hass: HomeAssistant, + async_add_entities: AddEntitiesCallback, + scan_interval: timedelta, + coordinates: tuple[float, float], + url: str, + radius_in_km: float, + ) -> None: """Initialize the GeoJSON Feed Manager.""" self._hass = hass @@ -97,10 +106,10 @@ class GeoJsonFeedEntityManager: 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() @@ -108,26 +117,26 @@ class GeoJsonFeedEntityManager: 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) -> GenericFeedEntry | 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 = GeoJsonLocationEvent(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, f"geo_json_events_update_{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, f"geo_json_events_delete_{external_id}") @@ -135,18 +144,18 @@ class GeoJsonFeedEntityManager: class GeoJsonLocationEvent(GeolocationEvent): """This represents an external event with GeoJSON 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: GenericFeedManager, 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._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, @@ -160,68 +169,33 @@ class GeoJsonLocationEvent(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 GeoJSON 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: GenericFeedEntry) -> 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._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] @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.""" if not self._external_id: return {}