Use separate data coordinators for AccuWeather observation and forecast (#115628)
* Remove forecast option * Update strings * Use separate DataUpdateCoordinator for observation and forecast * Fix tests * Remove unneeded variable * Separate data coordinator classes * Use list comprehension * Separate coordinator clasess to add type annotations * Test the availability of the forecast sensor entity * Add DataUpdateCoordinator types * Use snapshot for test_sensor() --------- Co-authored-by: Maciej Bieniek <478555+bieniu@users.noreply.github.com>
This commit is contained in:
parent
81036967f0
commit
e7076ac83f
16 changed files with 6913 additions and 852 deletions
124
homeassistant/components/accuweather/coordinator.py
Normal file
124
homeassistant/components/accuweather/coordinator.py
Normal file
|
@ -0,0 +1,124 @@
|
|||
"""The AccuWeather coordinator."""
|
||||
|
||||
from asyncio import timeout
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from accuweather import AccuWeather, ApiError, InvalidApiKeyError, RequestsExceededError
|
||||
from aiohttp.client_exceptions import ClientConnectorError
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
DataUpdateCoordinator,
|
||||
TimestampDataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
|
||||
EXCEPTIONS = (ApiError, ClientConnectorError, InvalidApiKeyError, RequestsExceededError)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AccuWeatherObservationDataUpdateCoordinator(
|
||||
DataUpdateCoordinator[dict[str, Any]]
|
||||
):
|
||||
"""Class to manage fetching AccuWeather data API."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
accuweather: AccuWeather,
|
||||
name: str,
|
||||
coordinator_type: str,
|
||||
update_interval: timedelta,
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
self.accuweather = accuweather
|
||||
self.location_key = accuweather.location_key
|
||||
|
||||
if TYPE_CHECKING:
|
||||
assert self.location_key is not None
|
||||
|
||||
self.device_info = _get_device_info(self.location_key, name)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=f"{name} ({coordinator_type})",
|
||||
update_interval=update_interval,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> dict[str, Any]:
|
||||
"""Update data via library."""
|
||||
try:
|
||||
async with timeout(10):
|
||||
result = await self.accuweather.async_get_current_conditions()
|
||||
except EXCEPTIONS as error:
|
||||
raise UpdateFailed(error) from error
|
||||
|
||||
_LOGGER.debug("Requests remaining: %d", self.accuweather.requests_remaining)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class AccuWeatherDailyForecastDataUpdateCoordinator(
|
||||
TimestampDataUpdateCoordinator[list[dict[str, Any]]]
|
||||
):
|
||||
"""Class to manage fetching AccuWeather data API."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
accuweather: AccuWeather,
|
||||
name: str,
|
||||
coordinator_type: str,
|
||||
update_interval: timedelta,
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
self.accuweather = accuweather
|
||||
self.location_key = accuweather.location_key
|
||||
|
||||
if TYPE_CHECKING:
|
||||
assert self.location_key is not None
|
||||
|
||||
self.device_info = _get_device_info(self.location_key, name)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=f"{name} ({coordinator_type})",
|
||||
update_interval=update_interval,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> list[dict[str, Any]]:
|
||||
"""Update data via library."""
|
||||
try:
|
||||
async with timeout(10):
|
||||
result = await self.accuweather.async_get_daily_forecast()
|
||||
except EXCEPTIONS as error:
|
||||
raise UpdateFailed(error) from error
|
||||
|
||||
_LOGGER.debug("Requests remaining: %d", self.accuweather.requests_remaining)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _get_device_info(location_key: str, name: str) -> DeviceInfo:
|
||||
"""Get device info."""
|
||||
return DeviceInfo(
|
||||
entry_type=DeviceEntryType.SERVICE,
|
||||
identifiers={(DOMAIN, location_key)},
|
||||
manufacturer=MANUFACTURER,
|
||||
name=name,
|
||||
# You don't need to provide specific details for the URL,
|
||||
# so passing in _ characters is fine if the location key
|
||||
# is correct
|
||||
configuration_url=(
|
||||
"http://accuweather.com/en/"
|
||||
f"_/_/{location_key}/weather-forecast/{location_key}/"
|
||||
),
|
||||
)
|
Loading…
Add table
Add a link
Reference in a new issue