458 lines
15 KiB
Python
458 lines
15 KiB
Python
"""Pogoda data updater."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import aiohttp
|
|
import json
|
|
import logging
|
|
import math
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from dateutil import parser
|
|
|
|
from homeassistant.components.weather import (
|
|
ATTR_FORECAST_CONDITION,
|
|
ATTR_FORECAST_NATIVE_APPARENT_TEMP,
|
|
ATTR_FORECAST_NATIVE_TEMP,
|
|
ATTR_FORECAST_NATIVE_TEMP_LOW,
|
|
ATTR_FORECAST_NATIVE_WIND_GUST_SPEED,
|
|
ATTR_FORECAST_NATIVE_WIND_SPEED,
|
|
ATTR_FORECAST_WIND_BEARING,
|
|
Forecast,
|
|
)
|
|
from homeassistant.core import HomeAssistant, HomeAssistantError, callback
|
|
from homeassistant.helpers.device_registry import DeviceEntryType
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
|
|
|
from .const import (
|
|
ATTR_API_CONDITION,
|
|
ATTR_API_DAYTIME,
|
|
ATTR_API_FEELS_LIKE_TEMPERATURE,
|
|
ATTR_API_IMAGE,
|
|
ATTR_API_POLAR,
|
|
ATTR_API_SERVER_TIME,
|
|
ATTR_API_SUNRISE_BEGIN_TIME,
|
|
ATTR_API_SUNRISE_END_TIME,
|
|
ATTR_API_TEMPERATURE,
|
|
ATTR_API_TEMPERATURE_MAX,
|
|
ATTR_API_TEMPERATURE_MIN,
|
|
ATTR_API_TIME,
|
|
ATTR_API_WIND_BEARING,
|
|
ATTR_API_WIND_GUST,
|
|
ATTR_API_WIND_SPEED,
|
|
ATTR_API_YA_CONDITION,
|
|
ATTR_FORECAST_DATA,
|
|
ATTR_FORECAST_HOURLY,
|
|
ATTR_FORECAST_HOURLY_ICONS,
|
|
ATTR_FORECAST_TWICE_DAILY,
|
|
ATTR_FORECAST_TWICE_DAILY_ICONS,
|
|
ATTR_MIN_FORECAST_TEMPERATURE,
|
|
ATTR_WIND_INTERCARDINAL_DIRECTION,
|
|
DEFAULT_UPDATES_PER_DAY,
|
|
UPDATES_PER_DAY_MAP,
|
|
CONDITION_ICONS,
|
|
DOMAIN,
|
|
MANUFACTURER,
|
|
WEATHER_STATES_CONVERSION,
|
|
YA_CONDITION_STATE_MAP,
|
|
get_wind_intercardinal_direction,
|
|
is_daytime,
|
|
map_state,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
API_URL = "https://api.weather.yandex.ru/graphql/query"
|
|
API_HEADER_NAME = "X-Yandex-Weather-Key"
|
|
|
|
API_QUERY_TEMPLATE = f"""{{
|
|
{ATTR_API_SERVER_TIME}
|
|
weatherByPoint(request: {{ lat: %s, lon: %s }}) {{
|
|
now {{
|
|
{ATTR_API_CONDITION}
|
|
{ATTR_API_DAYTIME}
|
|
{ATTR_API_FEELS_LIKE_TEMPERATURE}
|
|
{ATTR_API_IMAGE}(format: PNG_64)
|
|
{ATTR_API_TEMPERATURE}
|
|
{ATTR_API_WIND_BEARING}
|
|
{ATTR_API_WIND_GUST}
|
|
{ATTR_API_WIND_SPEED}
|
|
}}
|
|
forecast {{
|
|
days {{
|
|
{ATTR_API_POLAR}
|
|
{ATTR_API_TIME}
|
|
{ATTR_API_SUNRISE_BEGIN_TIME}
|
|
{ATTR_API_SUNRISE_END_TIME}
|
|
hours{{
|
|
{ATTR_API_TIME}
|
|
{ATTR_API_CONDITION}
|
|
{ATTR_API_FEELS_LIKE_TEMPERATURE}
|
|
{ATTR_API_IMAGE}(format: PNG_64)
|
|
{ATTR_API_TEMPERATURE}
|
|
{ATTR_API_WIND_BEARING}
|
|
{ATTR_API_WIND_GUST}
|
|
{ATTR_API_WIND_SPEED}
|
|
}}
|
|
summary{{
|
|
night{{
|
|
...forecastFields
|
|
}}
|
|
day{{
|
|
...forecastFields
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
fragment forecastFields on Daypart {{
|
|
{ATTR_API_CONDITION}
|
|
{ATTR_API_FEELS_LIKE_TEMPERATURE}
|
|
{ATTR_API_IMAGE}(format: PNG_64)
|
|
{ATTR_API_WIND_BEARING}
|
|
{ATTR_API_WIND_GUST}
|
|
{ATTR_API_WIND_SPEED}
|
|
{ATTR_API_TEMPERATURE_MAX}
|
|
{ATTR_API_TEMPERATURE_MIN}
|
|
}}"""
|
|
|
|
|
|
@dataclass
|
|
class AttributeMapper:
|
|
"""Attribute mapper."""
|
|
|
|
src: str
|
|
_dst: str | None = None
|
|
mapping: dict | None = None
|
|
default: str | float | None = None
|
|
|
|
@property
|
|
def dst(self) -> str:
|
|
"""Destination for mapping."""
|
|
return self.src if self._dst is None else self._dst
|
|
|
|
|
|
CURRENT_WEATHER_ATTRIBUTE_TRANSLATION: list[AttributeMapper] = [
|
|
AttributeMapper(ATTR_API_WIND_BEARING),
|
|
AttributeMapper(
|
|
ATTR_API_WIND_BEARING,
|
|
ATTR_WIND_INTERCARDINAL_DIRECTION,
|
|
mapping=get_wind_intercardinal_direction,
|
|
),
|
|
AttributeMapper(ATTR_API_CONDITION, ATTR_API_YA_CONDITION, YA_CONDITION_STATE_MAP),
|
|
AttributeMapper(
|
|
ATTR_API_CONDITION, f"{ATTR_API_YA_CONDITION}_icon", CONDITION_ICONS
|
|
),
|
|
AttributeMapper(ATTR_API_CONDITION, mapping=WEATHER_STATES_CONVERSION),
|
|
AttributeMapper(ATTR_API_FEELS_LIKE_TEMPERATURE),
|
|
AttributeMapper(ATTR_API_SUNRISE_BEGIN_TIME),
|
|
AttributeMapper(ATTR_API_SUNRISE_END_TIME),
|
|
AttributeMapper(ATTR_API_IMAGE),
|
|
AttributeMapper(ATTR_API_TEMPERATURE),
|
|
AttributeMapper(ATTR_API_WIND_GUST),
|
|
AttributeMapper(ATTR_API_WIND_SPEED, default=0),
|
|
]
|
|
|
|
|
|
FORECAST_HOUR_ATTRIBUTE_TRANSLATION: list[AttributeMapper] = [
|
|
AttributeMapper(ATTR_API_WIND_BEARING, ATTR_FORECAST_WIND_BEARING),
|
|
AttributeMapper(ATTR_API_WIND_SPEED, ATTR_FORECAST_NATIVE_WIND_SPEED, default=0),
|
|
AttributeMapper(
|
|
ATTR_API_FEELS_LIKE_TEMPERATURE, ATTR_FORECAST_NATIVE_APPARENT_TEMP
|
|
),
|
|
AttributeMapper(ATTR_API_TEMPERATURE, ATTR_FORECAST_NATIVE_TEMP),
|
|
AttributeMapper(
|
|
ATTR_API_CONDITION, ATTR_FORECAST_CONDITION, mapping=WEATHER_STATES_CONVERSION
|
|
),
|
|
AttributeMapper(ATTR_API_WIND_GUST, ATTR_FORECAST_NATIVE_WIND_GUST_SPEED),
|
|
]
|
|
|
|
|
|
FORECAST_DAY_ATTRIBUTE_TRANSLATION: list[AttributeMapper] = [
|
|
AttributeMapper(ATTR_API_WIND_BEARING, ATTR_FORECAST_WIND_BEARING),
|
|
AttributeMapper(ATTR_API_WIND_SPEED, ATTR_FORECAST_NATIVE_WIND_SPEED, default=0),
|
|
AttributeMapper(
|
|
ATTR_API_FEELS_LIKE_TEMPERATURE, ATTR_FORECAST_NATIVE_APPARENT_TEMP
|
|
),
|
|
AttributeMapper(ATTR_API_TEMPERATURE_MAX, ATTR_FORECAST_NATIVE_TEMP),
|
|
AttributeMapper(ATTR_API_TEMPERATURE_MIN, ATTR_FORECAST_NATIVE_TEMP_LOW),
|
|
AttributeMapper(
|
|
ATTR_API_CONDITION, ATTR_FORECAST_CONDITION, mapping=WEATHER_STATES_CONVERSION
|
|
),
|
|
AttributeMapper(ATTR_API_WIND_GUST, ATTR_FORECAST_NATIVE_WIND_GUST_SPEED),
|
|
]
|
|
|
|
|
|
class WeatherUpdater(DataUpdateCoordinator):
|
|
"""Weather data updater for interaction with Yandex.Weather API."""
|
|
|
|
def __init__(
|
|
self,
|
|
latitude: float,
|
|
longitude: float,
|
|
api_key: str,
|
|
hass: HomeAssistant,
|
|
device_id: str,
|
|
name="Yandex Weather",
|
|
updates_per_day: int = UPDATES_PER_DAY_MAP[DEFAULT_UPDATES_PER_DAY],
|
|
weather_data: dict | None = None,
|
|
):
|
|
"""Initialize updater.
|
|
|
|
:param latitude: latitude of location for weather data
|
|
:param longitude: longitude of location for weather data
|
|
:param api_key: Yandex weather API. MUST be weather for site tariff plan
|
|
:param hass: Home Assistant object
|
|
:param language: Language for yandex_condition
|
|
:param updates_per_day: int: how many updates per day we should do?
|
|
:param device_id: ID of integration Device in Home Assistant
|
|
:param weather_data: if not None then weather_data will be used for first update
|
|
"""
|
|
|
|
self.__api_key = api_key.strip()
|
|
self._lat = latitude
|
|
self._lon = longitude
|
|
self._device_id = device_id
|
|
self._name = name
|
|
self.weather_data = weather_data
|
|
self.is_first_update_requred = bool(weather_data)
|
|
|
|
self.update_interval = timedelta(
|
|
seconds=math.ceil((24 * 60 * 60) / updates_per_day)
|
|
)
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
name=f"{self._name} updater",
|
|
update_interval=self.update_interval,
|
|
update_method=self.update,
|
|
)
|
|
self.data = {}
|
|
|
|
def process_data(
|
|
self, dst: dict, src: dict, attributes: list[AttributeMapper], is_day: bool
|
|
):
|
|
"""Convert Yandex API weather response to HA friendly.
|
|
|
|
:param dst: weather data for HomeAssistant
|
|
:param src: weather data form Yandex
|
|
:param attributes: how to translate src to dst
|
|
"""
|
|
|
|
for attribute in attributes:
|
|
value = src.get(attribute.src, attribute.default)
|
|
|
|
if value is not None and attribute.mapping is not None:
|
|
if isinstance(attribute.mapping, dict):
|
|
value = map_state(
|
|
src=str(value), is_day=is_day, mapping=attribute.mapping
|
|
)
|
|
else:
|
|
value = attribute.mapping(value)
|
|
|
|
dst[attribute.dst] = value
|
|
|
|
@staticmethod
|
|
def get_min_forecast_temperature(forecasts: list[dict]) -> float | None:
|
|
"""Get minimum temperature from forecast data."""
|
|
low_fc_temperatures: list[float] = []
|
|
|
|
for f in forecasts:
|
|
f_low_temperature: float = f.get(ATTR_FORECAST_NATIVE_TEMP, None)
|
|
if f_low_temperature is not None:
|
|
low_fc_temperatures.append(f_low_temperature)
|
|
|
|
return min(low_fc_temperatures) if len(low_fc_temperatures) > 0 else None
|
|
|
|
async def _get_weather_data(self):
|
|
if self.is_first_update_requred: # No need to make API request
|
|
self.is_first_update_requred = False
|
|
_LOGGER.info("Skip API request")
|
|
return self.weather_data
|
|
|
|
timeout = aiohttp.ClientTimeout(total=20)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
rs = await self.request(session, self.__api_key, self._lat, self._lon)
|
|
|
|
self.weather_data = rs["data"]
|
|
return self.weather_data
|
|
|
|
async def update(self):
|
|
"""Update weather information.
|
|
|
|
:returns: dict with weather data.
|
|
"""
|
|
data = await self._get_weather_data()
|
|
weather_by_point = data["weatherByPoint"]
|
|
now = weather_by_point["now"]
|
|
now_dt = parser.parse(data[ATTR_API_SERVER_TIME])
|
|
result = {
|
|
ATTR_API_SERVER_TIME: now_dt,
|
|
ATTR_FORECAST_DATA: {
|
|
ATTR_FORECAST_HOURLY: [],
|
|
ATTR_FORECAST_TWICE_DAILY: [],
|
|
},
|
|
ATTR_FORECAST_HOURLY_ICONS: [],
|
|
ATTR_FORECAST_TWICE_DAILY_ICONS: [],
|
|
ATTR_API_SUNRISE_BEGIN_TIME: "",
|
|
ATTR_API_SUNRISE_END_TIME: ""
|
|
}
|
|
self.process_data(
|
|
result,
|
|
now,
|
|
CURRENT_WEATHER_ATTRIBUTE_TRANSLATION,
|
|
now[ATTR_API_DAYTIME] == "DAY",
|
|
)
|
|
|
|
for day in weather_by_point["forecast"]["days"]:
|
|
sunrise_begin = parser.parse(day[ATTR_API_SUNRISE_BEGIN_TIME])
|
|
sunset_end = parser.parse(day[ATTR_API_SUNRISE_END_TIME])
|
|
day_dt = parser.parse(day[ATTR_API_TIME])
|
|
polar = day.get(ATTR_API_POLAR)
|
|
|
|
if sunrise_begin > now_dt and not result[ATTR_API_SUNRISE_BEGIN_TIME]:
|
|
result[ATTR_API_SUNRISE_BEGIN_TIME] = sunrise_begin
|
|
|
|
if sunset_end > now_dt and not result[ATTR_API_SUNRISE_END_TIME]:
|
|
result[ATTR_API_SUNRISE_END_TIME] = sunset_end
|
|
|
|
for hour in day["hours"]:
|
|
hour_dt = parser.parse(hour[ATTR_API_TIME])
|
|
if now_dt > hour_dt:
|
|
continue
|
|
|
|
hour_forecast = Forecast(datetime=hour_dt)
|
|
self.process_data(
|
|
dst=hour_forecast,
|
|
src=hour,
|
|
attributes=FORECAST_HOUR_ATTRIBUTE_TRANSLATION,
|
|
is_day=is_daytime(
|
|
hour_dt, day_dt, sunrise_begin, sunset_end, polar
|
|
),
|
|
)
|
|
result[ATTR_FORECAST_DATA][ATTR_FORECAST_HOURLY].append(hour_forecast)
|
|
result[ATTR_FORECAST_HOURLY_ICONS].append(hour.get(ATTR_API_IMAGE))
|
|
|
|
day_part = day["summary"]["day"]
|
|
night_part = day["summary"]["night"]
|
|
for part, is_day in ((night_part, False), (day_part, True)):
|
|
part_forecast = Forecast(datetime=day_dt, is_daytime=is_day)
|
|
self.process_data(
|
|
dst=part_forecast,
|
|
src=part,
|
|
attributes=FORECAST_DAY_ATTRIBUTE_TRANSLATION,
|
|
is_day=is_day,
|
|
)
|
|
result[ATTR_FORECAST_DATA][ATTR_FORECAST_TWICE_DAILY].append(
|
|
part_forecast
|
|
)
|
|
result[ATTR_FORECAST_TWICE_DAILY_ICONS].append(part.get(ATTR_API_IMAGE))
|
|
day_dt = day_dt + timedelta(hours=12)
|
|
|
|
result[ATTR_MIN_FORECAST_TEMPERATURE] = self.get_min_forecast_temperature(
|
|
result[ATTR_FORECAST_DATA][ATTR_FORECAST_HOURLY]
|
|
)
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
async def request(
|
|
session: aiohttp.ClientSession,
|
|
api_key: str,
|
|
lat: float,
|
|
lon: float,
|
|
):
|
|
"""Make request to API endpoint.
|
|
|
|
:param session: aiohttp.ClientSession: HTTP session for request
|
|
:param api_key: str: API key
|
|
:param lat: float: latitude of location where we are getting weather data
|
|
:param lon: float: longitude of location where we ate getting weather data
|
|
|
|
:returns: dict with response data
|
|
:raises AssertionError: when response.status is not 200
|
|
"""
|
|
_LOGGER.info("Sending API request")
|
|
async with session.post(
|
|
API_URL,
|
|
headers={API_HEADER_NAME: api_key},
|
|
json={"query": API_QUERY_TEMPLATE % (lat, lon)},
|
|
) as response:
|
|
try:
|
|
assert response.status == 200
|
|
except AssertionError as e:
|
|
_LOGGER.error("Could not get data from API: %s", response)
|
|
raise aiohttp.ClientError(response.status, await response.text()) from e
|
|
|
|
result = await response.json()
|
|
|
|
if "errors" in result:
|
|
msg = f"Errors in weather API response: {result}"
|
|
_LOGGER.error(msg)
|
|
raise HomeAssistantError(msg)
|
|
|
|
_LOGGER.debug(result)
|
|
return result
|
|
|
|
def __str__(self):
|
|
"""Show as pretty look data json."""
|
|
_d = dict(self.data)
|
|
_d[ATTR_API_SERVER_TIME] = str(_d[ATTR_API_SERVER_TIME])
|
|
return json.dumps(_d, indent=4, sort_keys=True)
|
|
|
|
@property
|
|
def url(self) -> str:
|
|
"""Weather URL."""
|
|
return f"https://yandex.ru/weather/?lat={self._lat}&lon={self._lon}"
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Device info."""
|
|
return DeviceInfo(
|
|
entry_type=DeviceEntryType.SERVICE,
|
|
identifiers={(DOMAIN, self.device_id)},
|
|
manufacturer=MANUFACTURER,
|
|
name=self._name,
|
|
configuration_url=self.url,
|
|
)
|
|
|
|
def schedule_refresh(self, offset: timedelta):
|
|
"""Schedule refresh."""
|
|
if self._unsub_refresh:
|
|
self._unsub_refresh()
|
|
self._unsub_refresh = None
|
|
|
|
_LOGGER.debug(f"scheduling next refresh after {offset=}")
|
|
next_refresh = (
|
|
int(self.hass.loop.time()) + self._microsecond + offset.total_seconds()
|
|
)
|
|
self._unsub_refresh = self.hass.loop.call_at(
|
|
next_refresh, self.__wrap_handle_refresh_interval
|
|
).cancel
|
|
|
|
@callback
|
|
def __wrap_handle_refresh_interval(self) -> None:
|
|
"""Handle a refresh interval occurrence."""
|
|
# We need this private callback from parent class
|
|
if self.config_entry:
|
|
self.config_entry.async_create_background_task(
|
|
self.hass,
|
|
self._handle_refresh_interval(),
|
|
name=f"{self.name} - {self.config_entry.title} - refresh",
|
|
eager_start=True,
|
|
)
|
|
else:
|
|
self.hass.async_create_background_task(
|
|
self._handle_refresh_interval(),
|
|
name=f"{self.name} - refresh",
|
|
eager_start=True,
|
|
)
|
|
|
|
@property
|
|
def device_id(self) -> str:
|
|
"""Device ID."""
|
|
return self._device_id
|