Use DataUpdateCoordinator in ISS (#65178)
* Move update method to coordinator * Add missing type annotations * Simplify update function * Add missing type annotation for coordinates * Forgot to extend with CoordinatorEntity * ... * Tweaks * ... * Fix linting and conflicts * import coordinatorentity * ... * Hopefully fixed linting * ... * Fix suggestions
This commit is contained in:
Simon Hansen
GitHub
parent
fa0dfd812c
commit
cf5a11a1e7
2 changed files with 88 additions and 63 deletions
|
|
@ -1,18 +1,71 @@
|
|||
"""The iss component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
|
||||
import pyiss
|
||||
import requests
|
||||
from requests.exceptions import HTTPError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS = [Platform.BINARY_SENSOR]
|
||||
|
||||
|
||||
@dataclass
|
||||
class IssData:
|
||||
"""Dataclass representation of data returned from pyiss."""
|
||||
|
||||
number_of_people_in_space: int
|
||||
current_location: dict[str, str]
|
||||
is_above: bool
|
||||
next_rise: datetime
|
||||
|
||||
|
||||
def update(iss: pyiss.ISS, latitude: float, longitude: float) -> IssData:
|
||||
"""Retrieve data from the pyiss API."""
|
||||
return IssData(
|
||||
number_of_people_in_space=iss.number_of_people_in_space(),
|
||||
current_location=iss.current_location(),
|
||||
is_above=iss.is_ISS_above(latitude, longitude),
|
||||
next_rise=iss.next_rise(latitude, longitude),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up this integration using UI."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
latitude = hass.config.latitude
|
||||
longitude = hass.config.longitude
|
||||
|
||||
iss = pyiss.ISS()
|
||||
|
||||
async def async_update() -> IssData:
|
||||
try:
|
||||
return await hass.async_add_executor_job(update, iss, latitude, longitude)
|
||||
except (HTTPError, requests.exceptions.ConnectionError) as ex:
|
||||
raise UpdateFailed("Unable to retrieve data") from ex
|
||||
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_method=async_update,
|
||||
update_interval=timedelta(seconds=60),
|
||||
)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data[DOMAIN] = coordinator
|
||||
|
||||
entry.async_on_unload(entry.add_update_listener(update_listener))
|
||||
|
||||
|
|
|
|||
|
|
@ -1,19 +1,21 @@
|
|||
"""Support for iss binary sensor."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
import pyiss
|
||||
import requests
|
||||
from requests.exceptions import HTTPError
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_SHOW_ON_MAP
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.util import Throttle
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
)
|
||||
|
||||
from . import IssData
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -23,8 +25,6 @@ ATTR_ISS_NUMBER_PEOPLE_SPACE = "number_of_people_in_space"
|
|||
DEFAULT_NAME = "ISS"
|
||||
DEFAULT_DEVICE_CLASS = "visible"
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
|
|
@ -32,27 +32,26 @@ async def async_setup_entry(
|
|||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the sensor platform."""
|
||||
coordinator: DataUpdateCoordinator[IssData] = hass.data[DOMAIN]
|
||||
|
||||
name = entry.title
|
||||
show_on_map = entry.options.get(CONF_SHOW_ON_MAP, False)
|
||||
|
||||
try:
|
||||
iss_data = IssData(hass.config.latitude, hass.config.longitude)
|
||||
await hass.async_add_executor_job(iss_data.update)
|
||||
except HTTPError as error:
|
||||
_LOGGER.error(error)
|
||||
return
|
||||
|
||||
async_add_entities([IssBinarySensor(iss_data, name, show_on_map)], True)
|
||||
async_add_entities([IssBinarySensor(coordinator, name, show_on_map)])
|
||||
|
||||
|
||||
class IssBinarySensor(BinarySensorEntity):
|
||||
class IssBinarySensor(
|
||||
CoordinatorEntity[DataUpdateCoordinator[IssData]], BinarySensorEntity
|
||||
):
|
||||
"""Implementation of the ISS binary sensor."""
|
||||
|
||||
_attr_device_class = DEFAULT_DEVICE_CLASS
|
||||
|
||||
def __init__(self, iss_data, name, show):
|
||||
def __init__(
|
||||
self, coordinator: DataUpdateCoordinator[IssData], name: str, show: bool
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
self.iss_data = iss_data
|
||||
super().__init__(coordinator)
|
||||
self._state = None
|
||||
self._attr_name = name
|
||||
self._show_on_map = show
|
||||
|
|
@ -60,51 +59,24 @@ class IssBinarySensor(BinarySensorEntity):
|
|||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if the binary sensor is on."""
|
||||
return self.iss_data.is_above if self.iss_data else False
|
||||
return self.coordinator.data.is_above is True
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the state attributes."""
|
||||
if self.iss_data:
|
||||
attrs = {
|
||||
ATTR_ISS_NUMBER_PEOPLE_SPACE: self.iss_data.number_of_people_in_space,
|
||||
ATTR_ISS_NEXT_RISE: self.iss_data.next_rise,
|
||||
}
|
||||
if self._show_on_map:
|
||||
attrs[ATTR_LONGITUDE] = self.iss_data.position.get("longitude")
|
||||
attrs[ATTR_LATITUDE] = self.iss_data.position.get("latitude")
|
||||
else:
|
||||
attrs["long"] = self.iss_data.position.get("longitude")
|
||||
attrs["lat"] = self.iss_data.position.get("latitude")
|
||||
attrs = {
|
||||
ATTR_ISS_NUMBER_PEOPLE_SPACE: self.coordinator.data.number_of_people_in_space,
|
||||
ATTR_ISS_NEXT_RISE: self.coordinator.data.next_rise,
|
||||
}
|
||||
if self._show_on_map:
|
||||
attrs[ATTR_LONGITUDE] = self.coordinator.data.current_location.get(
|
||||
"longitude"
|
||||
)
|
||||
attrs[ATTR_LATITUDE] = self.coordinator.data.current_location.get(
|
||||
"latitude"
|
||||
)
|
||||
else:
|
||||
attrs["long"] = self.coordinator.data.current_location.get("longitude")
|
||||
attrs["lat"] = self.coordinator.data.current_location.get("latitude")
|
||||
|
||||
return attrs
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data from ISS API and updates the states."""
|
||||
self.iss_data.update()
|
||||
|
||||
|
||||
class IssData:
|
||||
"""Get data from the ISS API."""
|
||||
|
||||
def __init__(self, latitude, longitude):
|
||||
"""Initialize the data object."""
|
||||
self.is_above = None
|
||||
self.next_rise = None
|
||||
self.number_of_people_in_space = None
|
||||
self.position = None
|
||||
self.latitude = latitude
|
||||
self.longitude = longitude
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
def update(self):
|
||||
"""Get the latest data from the ISS API."""
|
||||
try:
|
||||
iss = pyiss.ISS()
|
||||
self.is_above = iss.is_ISS_above(self.latitude, self.longitude)
|
||||
self.next_rise = iss.next_rise(self.latitude, self.longitude)
|
||||
self.number_of_people_in_space = iss.number_of_people_in_space()
|
||||
self.position = iss.current_location()
|
||||
except (HTTPError, requests.exceptions.ConnectionError):
|
||||
_LOGGER.error("Unable to retrieve data")
|
||||
return False
|
||||
return attrs
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue