diff --git a/homeassistant/components/iss/__init__.py b/homeassistant/components/iss/__init__.py index 476a944be81..d6065fd4f78 100644 --- a/homeassistant/components/iss/__init__.py +++ b/homeassistant/components/iss/__init__.py @@ -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)) diff --git a/homeassistant/components/iss/binary_sensor.py b/homeassistant/components/iss/binary_sensor.py index e2034fb48f9..77cb86fc45a 100644 --- a/homeassistant/components/iss/binary_sensor.py +++ b/homeassistant/components/iss/binary_sensor.py @@ -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