evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate OpenWeaterMap to new library (support API 3.0) (#116870)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Evgeny 2024-05-24 09:51:10 +02:00 • committed by GitHub
parent 5bca9d142c
commit 24d31924a0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 580 additions and 523 deletions
Download patch file Download diff file Expand all files Collapse all files

297
homeassistant/components/openweathermap/coordinator.py
View file

@ -1,39 +1,35 @@
"""Weather data coordinator for the OpenWeatherMap (OWM) service."""
import asyncio
from datetime import timedelta
import logging
from pyowm.commons.exceptions import APIRequestError, UnauthorizedError
from pyopenweathermap import (
CurrentWeather,
DailyWeatherForecast,
HourlyWeatherForecast,
OWMClient,
RequestError,
WeatherReport,
)
from homeassistant.components.weather import (
ATTR_CONDITION_CLEAR_NIGHT,
ATTR_CONDITION_SUNNY,
Forecast,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers import sun
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_conversion import TemperatureConverter
from .const import (
ATTR_API_CLOUDS,
ATTR_API_CONDITION,
ATTR_API_CURRENT,
ATTR_API_DAILY_FORECAST,
ATTR_API_DEW_POINT,
ATTR_API_FEELS_LIKE_TEMPERATURE,
ATTR_API_FORECAST,
ATTR_API_FORECAST_CLOUDS,
ATTR_API_FORECAST_CONDITION,
ATTR_API_FORECAST_FEELS_LIKE_TEMPERATURE,
ATTR_API_FORECAST_HUMIDITY,
ATTR_API_FORECAST_PRECIPITATION,
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_API_FORECAST_PRESSURE,
ATTR_API_FORECAST_TEMP,
ATTR_API_FORECAST_TEMP_LOW,
ATTR_API_FORECAST_TIME,
ATTR_API_FORECAST_WIND_BEARING,
ATTR_API_FORECAST_WIND_SPEED,
ATTR_API_HOURLY_FORECAST,
ATTR_API_HUMIDITY,
ATTR_API_PRECIPITATION_KIND,
ATTR_API_PRESSURE,
@ -49,10 +45,6 @@ from .const import (
ATTR_API_WIND_SPEED,
CONDITION_MAP,
DOMAIN,
FORECAST_MODE_DAILY,
FORECAST_MODE_HOURLY,
FORECAST_MODE_ONECALL_DAILY,
FORECAST_MODE_ONECALL_HOURLY,
WEATHER_CODE_SUNNY_OR_CLEAR_NIGHT,
)
@ -64,15 +56,17 @@ WEATHER_UPDATE_INTERVAL = timedelta(minutes=10)
class WeatherUpdateCoordinator(DataUpdateCoordinator):
"""Weather data update coordinator."""
def __init__(self, owm, latitude, longitude, forecast_mode, hass):
def __init__(
self,
owm_client: OWMClient,
latitude,
longitude,
hass: HomeAssistant,
) -> None:
"""Initialize coordinator."""
self._owm_client = owm
self._owm_client = owm_client
self._latitude = latitude
self._longitude = longitude
self.forecast_mode = forecast_mode
self._forecast_limit = None
if forecast_mode == FORECAST_MODE_DAILY:
self._forecast_limit = 15
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=WEATHER_UPDATE_INTERVAL
@ -80,184 +74,122 @@ class WeatherUpdateCoordinator(DataUpdateCoordinator):
async def _async_update_data(self):
"""Update the data."""
data = {}
async with asyncio.timeout(20):
try:
weather_response = await self._get_owm_weather()
data = self._convert_weather_response(weather_response)
except (APIRequestError, UnauthorizedError) as error:
raise UpdateFailed(error) from error
return data
async def _get_owm_weather(self):
"""Poll weather data from OWM."""
if self.forecast_mode in (
FORECAST_MODE_ONECALL_HOURLY,
FORECAST_MODE_ONECALL_DAILY,
):
weather = await self.hass.async_add_executor_job(
self._owm_client.one_call, self._latitude, self._longitude
)
else:
weather = await self.hass.async_add_executor_job(
self._get_legacy_weather_and_forecast
try:
weather_report = await self._owm_client.get_weather(
self._latitude, self._longitude
)
except RequestError as error:
raise UpdateFailed(error) from error
return self._convert_weather_response(weather_report)
return weather
def _get_legacy_weather_and_forecast(self):
"""Get weather and forecast data from OWM."""
interval = self._get_legacy_forecast_interval()
weather = self._owm_client.weather_at_coords(self._latitude, self._longitude)
forecast = self._owm_client.forecast_at_coords(
self._latitude, self._longitude, interval, self._forecast_limit
)
return LegacyWeather(weather.weather, forecast.forecast.weathers)
def _get_legacy_forecast_interval(self):
"""Get the correct forecast interval depending on the forecast mode."""
interval = "daily"
if self.forecast_mode == FORECAST_MODE_HOURLY:
interval = "3h"
return interval
def _convert_weather_response(self, weather_response):
def _convert_weather_response(self, weather_report: WeatherReport):
"""Format the weather response correctly."""
current_weather = weather_response.current
forecast_weather = self._get_forecast_from_weather_response(weather_response)
_LOGGER.debug("OWM weather response: %s", weather_report)
return {
ATTR_API_TEMPERATURE: current_weather.temperature("celsius").get("temp"),
ATTR_API_FEELS_LIKE_TEMPERATURE: current_weather.temperature("celsius").get(
"feels_like"
),
ATTR_API_DEW_POINT: self._fmt_dewpoint(current_weather.dewpoint),
ATTR_API_PRESSURE: current_weather.pressure.get("press"),
ATTR_API_CURRENT: self._get_current_weather_data(weather_report.current),
ATTR_API_HOURLY_FORECAST: [
self._get_hourly_forecast_weather_data(item)
for item in weather_report.hourly_forecast
],
ATTR_API_DAILY_FORECAST: [
self._get_daily_forecast_weather_data(item)
for item in weather_report.daily_forecast
],
}
def _get_current_weather_data(self, current_weather: CurrentWeather):
return {
ATTR_API_CONDITION: self._get_condition(current_weather.condition.id),
ATTR_API_TEMPERATURE: current_weather.temperature,
ATTR_API_FEELS_LIKE_TEMPERATURE: current_weather.feels_like,
ATTR_API_PRESSURE: current_weather.pressure,
ATTR_API_HUMIDITY: current_weather.humidity,
ATTR_API_WIND_BEARING: current_weather.wind().get("deg"),
ATTR_API_WIND_GUST: current_weather.wind().get("gust"),
ATTR_API_WIND_SPEED: current_weather.wind().get("speed"),
ATTR_API_CLOUDS: current_weather.clouds,
ATTR_API_RAIN: self._get_rain(current_weather.rain),
ATTR_API_SNOW: self._get_snow(current_weather.snow),
ATTR_API_DEW_POINT: current_weather.dew_point,
ATTR_API_CLOUDS: current_weather.cloud_coverage,
ATTR_API_WIND_SPEED: current_weather.wind_speed,
ATTR_API_WIND_GUST: current_weather.wind_gust,
ATTR_API_WIND_BEARING: current_weather.wind_bearing,
ATTR_API_WEATHER: current_weather.condition.description,
ATTR_API_WEATHER_CODE: current_weather.condition.id,
ATTR_API_UV_INDEX: current_weather.uv_index,
ATTR_API_VISIBILITY_DISTANCE: current_weather.visibility,
ATTR_API_RAIN: self._get_precipitation_value(current_weather.rain),
ATTR_API_SNOW: self._get_precipitation_value(current_weather.snow),
ATTR_API_PRECIPITATION_KIND: self._calc_precipitation_kind(
current_weather.rain, current_weather.snow
),
ATTR_API_WEATHER: current_weather.detailed_status,
ATTR_API_CONDITION: self._get_condition(current_weather.weather_code),
ATTR_API_UV_INDEX: current_weather.uvi,
ATTR_API_VISIBILITY_DISTANCE: current_weather.visibility_distance,
ATTR_API_WEATHER_CODE: current_weather.weather_code,
ATTR_API_FORECAST: forecast_weather,
}
def _get_forecast_from_weather_response(self, weather_response):
"""Extract the forecast data from the weather response."""
forecast_arg = "forecast"
if self.forecast_mode == FORECAST_MODE_ONECALL_HOURLY:
forecast_arg = "forecast_hourly"
elif self.forecast_mode == FORECAST_MODE_ONECALL_DAILY:
forecast_arg = "forecast_daily"
return [
self._convert_forecast(x) for x in getattr(weather_response, forecast_arg)
]
def _get_hourly_forecast_weather_data(self, forecast: HourlyWeatherForecast):
return Forecast(
datetime=forecast.date_time.isoformat(),
condition=self._get_condition(forecast.condition.id),
temperature=forecast.temperature,
native_apparent_temperature=forecast.feels_like,
pressure=forecast.pressure,
humidity=forecast.humidity,
native_dew_point=forecast.dew_point,
cloud_coverage=forecast.cloud_coverage,
wind_speed=forecast.wind_speed,
native_wind_gust_speed=forecast.wind_gust,
wind_bearing=forecast.wind_bearing,
uv_index=float(forecast.uv_index),
precipitation_probability=round(forecast.precipitation_probability * 100),
precipitation=self._calc_precipitation(forecast.rain, forecast.snow),
)
def _convert_forecast(self, entry):
"""Convert the forecast data."""
forecast = {
ATTR_API_FORECAST_TIME: dt_util.utc_from_timestamp(
entry.reference_time("unix")
).isoformat(),
ATTR_API_FORECAST_PRECIPITATION: self._calc_precipitation(
entry.rain, entry.snow
),
ATTR_API_FORECAST_PRECIPITATION_PROBABILITY: (
round(entry.precipitation_probability * 100)
),
ATTR_API_FORECAST_PRESSURE: entry.pressure.get("press"),
ATTR_API_FORECAST_WIND_SPEED: entry.wind().get("speed"),
ATTR_API_FORECAST_WIND_BEARING: entry.wind().get("deg"),
ATTR_API_FORECAST_CONDITION: self._get_condition(
entry.weather_code, entry.reference_time("unix")
),
ATTR_API_FORECAST_CLOUDS: entry.clouds,
ATTR_API_FORECAST_FEELS_LIKE_TEMPERATURE: entry.temperature("celsius").get(
"feels_like_day"
),
ATTR_API_FORECAST_HUMIDITY: entry.humidity,
}
temperature_dict = entry.temperature("celsius")
if "max" in temperature_dict and "min" in temperature_dict:
forecast[ATTR_API_FORECAST_TEMP] = entry.temperature("celsius").get("max")
forecast[ATTR_API_FORECAST_TEMP_LOW] = entry.temperature("celsius").get(
"min"
)
else:
forecast[ATTR_API_FORECAST_TEMP] = entry.temperature("celsius").get("temp")
return forecast
@staticmethod
def _fmt_dewpoint(dewpoint):
"""Format the dewpoint data."""
if dewpoint is not None:
return round(
TemperatureConverter.convert(
dewpoint, UnitOfTemperature.KELVIN, UnitOfTemperature.CELSIUS
),
1,
)
return None
@staticmethod
def _get_rain(rain):
"""Get rain data from weather data."""
if "all" in rain:
return round(rain["all"], 2)
if "3h" in rain:
return round(rain["3h"], 2)
if "1h" in rain:
return round(rain["1h"], 2)
return 0
@staticmethod
def _get_snow(snow):
"""Get snow data from weather data."""
if snow:
if "all" in snow:
return round(snow["all"], 2)
if "3h" in snow:
return round(snow["3h"], 2)
if "1h" in snow:
return round(snow["1h"], 2)
return 0
def _get_daily_forecast_weather_data(self, forecast: DailyWeatherForecast):
return Forecast(
datetime=forecast.date_time.isoformat(),
condition=self._get_condition(forecast.condition.id),
temperature=forecast.temperature.max,
templow=forecast.temperature.min,
native_apparent_temperature=forecast.feels_like,
pressure=forecast.pressure,
humidity=forecast.humidity,
native_dew_point=forecast.dew_point,
cloud_coverage=forecast.cloud_coverage,
wind_speed=forecast.wind_speed,
native_wind_gust_speed=forecast.wind_gust,
wind_bearing=forecast.wind_bearing,
uv_index=float(forecast.uv_index),
precipitation_probability=round(forecast.precipitation_probability * 100),
precipitation=round(forecast.rain + forecast.snow, 2),
)
@staticmethod
def _calc_precipitation(rain, snow):
"""Calculate the precipitation."""
rain_value = 0
if WeatherUpdateCoordinator._get_rain(rain) != 0:
rain_value = WeatherUpdateCoordinator._get_rain(rain)
snow_value = 0
if WeatherUpdateCoordinator._get_snow(snow) != 0:
snow_value = WeatherUpdateCoordinator._get_snow(snow)
rain_value = WeatherUpdateCoordinator._get_precipitation_value(rain)
snow_value = WeatherUpdateCoordinator._get_precipitation_value(snow)
return round(rain_value + snow_value, 2)
@staticmethod
def _calc_precipitation_kind(rain, snow):
"""Determine the precipitation kind."""
if WeatherUpdateCoordinator._get_rain(rain) != 0:
if WeatherUpdateCoordinator._get_snow(snow) != 0:
rain_value = WeatherUpdateCoordinator._get_precipitation_value(rain)
snow_value = WeatherUpdateCoordinator._get_precipitation_value(snow)
if rain_value != 0:
if snow_value != 0:
return "Snow and Rain"
return "Rain"
if WeatherUpdateCoordinator._get_snow(snow) != 0:
if snow_value != 0:
return "Snow"
return "None"
@staticmethod
def _get_precipitation_value(precipitation):
"""Get precipitation value from weather data."""
if "all" in precipitation:
return round(precipitation["all"], 2)
if "3h" in precipitation:
return round(precipitation["3h"], 2)
if "1h" in precipitation:
return round(precipitation["1h"], 2)
return 0
def _get_condition(self, weather_code, timestamp=None):
"""Get weather condition from weather data."""
if weather_code == WEATHER_CODE_SUNNY_OR_CLEAR_NIGHT:
@ -269,12 +201,3 @@ class WeatherUpdateCoordinator(DataUpdateCoordinator):
return ATTR_CONDITION_CLEAR_NIGHT
return CONDITION_MAP.get(weather_code)
class LegacyWeather:
"""Class to harmonize weather data model for hourly, daily and One Call APIs."""
def __init__(self, current_weather, forecast):
"""Initialize weather object."""
self.current = current_weather
self.forecast = forecast