Use instance attributes in geo_json_events (#74397)

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

View File

@ -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 {}