Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add OpenWeatherMap config_flow (#34659) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/39676/head
Evgeny 2020-09-05 01:22:50 +02:00 committed by GitHub
parent 1e770f089d
commit b0192cf9c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1193 additions and 424 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -623,6 +623,9 @@ omit =
homeassistant/components/openuv/sensor.py
homeassistant/components/openweathermap/sensor.py
homeassistant/components/openweathermap/weather.py
homeassistant/components/openweathermap/forecast_update_coordinator.py
homeassistant/components/openweathermap/weather_update_coordinator.py
homeassistant/components/openweathermap/abstract_owm_sensor.py
homeassistant/components/opnsense/*
homeassistant/components/opple/light.py
homeassistant/components/orangepi_gpio/*

2
CODEOWNERS
View File

@ -305,7 +305,7 @@ homeassistant/components/openerz/* @misialq
homeassistant/components/opengarage/* @danielhiversen
homeassistant/components/opentherm_gw/* @mvn23
homeassistant/components/openuv/* @bachya
homeassistant/components/openweathermap/* @fabaff
homeassistant/components/openweathermap/* @fabaff @freekode
homeassistant/components/opnsense/* @mtreinish
homeassistant/components/orangepi_gpio/* @pascallj
homeassistant/components/oru/* @bvlaicu

127
homeassistant/components/openweathermap/__init__.py
View File

@ -1 +1,128 @@
"""The openweathermap component."""
import asyncio
import logging
from pyowm import OWM
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_MODE,
CONF_NAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import (
COMPONENTS,
CONF_LANGUAGE,
DOMAIN,
ENTRY_FORECAST_COORDINATOR,
ENTRY_NAME,
ENTRY_WEATHER_COORDINATOR,
UPDATE_LISTENER,
)
from .forecast_update_coordinator import ForecastUpdateCoordinator
from .weather_update_coordinator import WeatherUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the OpenWeatherMap component."""
hass.data.setdefault(DOMAIN, {})
weather_configs = _filter_domain_configs(config.get("weather", []), DOMAIN)
sensor_configs = _filter_domain_configs(config.get("sensor", []), DOMAIN)
_import_configs(hass, weather_configs + sensor_configs)
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
"""Set up OpenWeatherMap as config entry."""
name = config_entry.data[CONF_NAME]
api_key = config_entry.data[CONF_API_KEY]
latitude = config_entry.data.get(CONF_LATITUDE, hass.config.latitude)
longitude = config_entry.data.get(CONF_LONGITUDE, hass.config.longitude)
forecast_mode = _get_config_value(config_entry, CONF_MODE)
language = _get_config_value(config_entry, CONF_LANGUAGE)
owm = OWM(API_key=api_key, language=language)
weather_coordinator = WeatherUpdateCoordinator(owm, latitude, longitude, hass)
forecast_coordinator = ForecastUpdateCoordinator(
owm, latitude, longitude, forecast_mode, hass
)
await weather_coordinator.async_refresh()
await forecast_coordinator.async_refresh()
if (
not weather_coordinator.last_update_success
and not forecast_coordinator.last_update_success
):
raise ConfigEntryNotReady
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = {
ENTRY_NAME: name,
ENTRY_WEATHER_COORDINATOR: weather_coordinator,
ENTRY_FORECAST_COORDINATOR: forecast_coordinator,
}
for component in COMPONENTS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
update_listener = config_entry.add_update_listener(async_update_options)
hass.data[DOMAIN][config_entry.entry_id][UPDATE_LISTENER] = update_listener
return True
async def async_update_options(hass: HomeAssistant, config_entry: ConfigEntry):
"""Update options."""
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in COMPONENTS
]
)
)
if unload_ok:
update_listener = hass.data[DOMAIN][config_entry.entry_id][UPDATE_LISTENER]
update_listener()
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok
def _import_configs(hass, configs):
for config in configs:
_LOGGER.debug("Importing OpenWeatherMap %s", config)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
def _filter_domain_configs(elements, domain):
return list(filter(lambda elem: elem["platform"] == domain, elements))
def _get_config_value(config_entry, key):
if config_entry.options:
return config_entry.options[key]
return config_entry.data[key]

81
homeassistant/components/openweathermap/abstract_owm_sensor.py Normal file
View File

@ -0,0 +1,81 @@
"""Abstraction form OWM sensors."""
import logging
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import ATTRIBUTION, SENSOR_DEVICE_CLASS, SENSOR_NAME, SENSOR_UNIT
_LOGGER = logging.getLogger(__name__)
class AbstractOpenWeatherMapSensor(Entity):
"""Abstract class for an OpenWeatherMap sensor."""
def __init__(
self,
name,
unique_id,
sensor_type,
sensor_configuration,
coordinator: DataUpdateCoordinator,
):
"""Initialize the sensor."""
self._name = name
self._unique_id = unique_id
self._sensor_type = sensor_type
self._sensor_name = sensor_configuration[SENSOR_NAME]
self._unit_of_measurement = sensor_configuration.get(SENSOR_UNIT)
self._device_class = sensor_configuration.get(SENSOR_DEVICE_CLASS)
self._coordinator = coordinator
@property
def name(self):
"""Return the name of the sensor."""
return f"{self._name} {self._sensor_name}"
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return self._unique_id
@property
def should_poll(self):
"""Return the polling requirement of the entity."""
return False
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def device_class(self):
"""Return the device_class."""
return self._device_class
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
@property
def device_state_attributes(self):
"""Return the state attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}
@property
def available(self):
"""Return True if entity is available."""
return self._coordinator.last_update_success
async def async_added_to_hass(self):
"""Connect to dispatcher listening for entity data notifications."""
self.async_on_remove(
self._coordinator.async_add_listener(self.async_write_ha_state)
)
async def async_update(self):
"""Get the latest data from OWM and updates the states."""
await self._coordinator.async_request_refresh()

138
homeassistant/components/openweathermap/config_flow.py Normal file
View File

@ -0,0 +1,138 @@
"""Config flow for OpenWeatherMap."""
import logging
from pyowm import OWM
from pyowm.exceptions.api_call_error import APICallError
from pyowm.exceptions.api_response_error import UnauthorizedError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_MODE,
CONF_NAME,
)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from .const import (
CONF_LANGUAGE,
DEFAULT_FORECAST_MODE,
DEFAULT_LANGUAGE,
DEFAULT_NAME,
FORECAST_MODES,
LANGUAGES,
)
from .const import DOMAIN # pylint:disable=unused-import
SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): str,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_MODE, default=DEFAULT_FORECAST_MODE): vol.In(FORECAST_MODES),
vol.Optional(CONF_LANGUAGE, default=DEFAULT_LANGUAGE): vol.In(LANGUAGES),
}
)
_LOGGER = logging.getLogger(__name__)
class OpenWeatherMapConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Config flow for OpenWeatherMap."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Get the options flow for this handler."""
return OpenWeatherMapOptionsFlow(config_entry)
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
latitude = user_input[CONF_LATITUDE]
longitude = user_input[CONF_LONGITUDE]
await self.async_set_unique_id(f"{latitude}-{longitude}")
self._abort_if_unique_id_configured()
try:
api_online = await _is_owm_api_online(
self.hass, user_input[CONF_API_KEY]
)
if not api_online:
errors["base"] = "auth"
except UnauthorizedError:
errors["base"] = "auth"
except APICallError:
errors["base"] = "connection"
if not errors:
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
return self.async_show_form(step_id="user", data_schema=SCHEMA, errors=errors)
async def async_step_import(self, import_input=None):
"""Set the config entry up from yaml."""
config = import_input.copy()
if CONF_NAME not in config:
config[CONF_NAME] = DEFAULT_NAME
if CONF_LATITUDE not in config:
config[CONF_LATITUDE] = self.hass.config.latitude
if CONF_LONGITUDE not in config:
config[CONF_LONGITUDE] = self.hass.config.longitude
if CONF_MODE not in config:
config[CONF_MODE] = DEFAULT_LANGUAGE
if CONF_LANGUAGE not in config:
config[CONF_LANGUAGE] = DEFAULT_LANGUAGE
return await self.async_step_user(config)
class OpenWeatherMapOptionsFlow(config_entries.OptionsFlow):
"""Handle options."""
def __init__(self, config_entry):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=self._get_options_schema(),
)
def _get_options_schema(self):
return vol.Schema(
{
vol.Optional(
CONF_MODE,
default=self.config_entry.options.get(
CONF_MODE, DEFAULT_FORECAST_MODE
),
): vol.In(FORECAST_MODES),
vol.Optional(
CONF_LANGUAGE,
default=self.config_entry.options.get(
CONF_LANGUAGE, DEFAULT_LANGUAGE
),
): vol.In(LANGUAGES),
}
)
async def _is_owm_api_online(hass, api_key):
owm = OWM(api_key)
return await hass.async_add_executor_job(owm.is_API_online)

142
homeassistant/components/openweathermap/const.py Normal file
View File

@ -0,0 +1,142 @@
"""Consts for the OpenWeatherMap."""
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
)
from homeassistant.const import (
DEGREE,
DEVICE_CLASS_HUMIDITY,
DEVICE_CLASS_PRESSURE,
DEVICE_CLASS_TEMPERATURE,
DEVICE_CLASS_TIMESTAMP,
PRESSURE_PA,
SPEED_METERS_PER_SECOND,
TEMP_CELSIUS,
UNIT_PERCENTAGE,
)
DOMAIN = "openweathermap"
DEFAULT_NAME = "OpenWeatherMap"
DEFAULT_LANGUAGE = "en"
DEFAULT_FORECAST_MODE = "freedaily"
ATTRIBUTION = "Data provided by OpenWeatherMap"
CONF_LANGUAGE = "language"
ENTRY_NAME = "name"
ENTRY_FORECAST_COORDINATOR = "forecast_coordinator"
ENTRY_WEATHER_COORDINATOR = "weather_coordinator"
ATTR_API_PRECIPITATION = "precipitation"
ATTR_API_DATETIME = "datetime"
ATTR_API_WEATHER = "weather"
ATTR_API_TEMPERATURE = "temperature"
ATTR_API_WIND_SPEED = "wind_speed"
ATTR_API_WIND_BEARING = "wind_bearing"
ATTR_API_HUMIDITY = "humidity"
ATTR_API_PRESSURE = "pressure"
ATTR_API_CONDITION = "condition"
ATTR_API_CLOUDS = "clouds"
ATTR_API_RAIN = "rain"
ATTR_API_SNOW = "snow"
ATTR_API_WEATHER_CODE = "weather_code"
ATTR_API_FORECAST = "forecast"
ATTR_API_THIS_DAY_FORECAST = "this_day_forecast"
SENSOR_NAME = "sensor_name"
SENSOR_UNIT = "sensor_unit"
SENSOR_DEVICE_CLASS = "sensor_device_class"
UPDATE_LISTENER = "update_listener"
COMPONENTS = ["sensor", "weather"]
FORECAST_MODES = ["hourly", "daily", "freedaily"]
MONITORED_CONDITIONS = [
ATTR_API_WEATHER,
ATTR_API_TEMPERATURE,
ATTR_API_WIND_SPEED,
ATTR_API_WIND_BEARING,
ATTR_API_HUMIDITY,
ATTR_API_PRESSURE,
ATTR_API_CLOUDS,
ATTR_API_RAIN,
ATTR_API_SNOW,
ATTR_API_CONDITION,
ATTR_API_WEATHER_CODE,
]
FORECAST_MONITORED_CONDITIONS = [
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
]
LANGUAGES = ["en", "es", "ru", "it"]
CONDITION_CLASSES = {
"cloudy": [803, 804],
"fog": [701, 741],
"hail": [906],
"lightning": [210, 211, 212, 221],
"lightning-rainy": [200, 201, 202, 230, 231, 232],
"partlycloudy": [801, 802],
"pouring": [504, 314, 502, 503, 522],
"rainy": [300, 301, 302, 310, 311, 312, 313, 500, 501, 520, 521],
"snowy": [600, 601, 602, 611, 612, 620, 621, 622],
"snowy-rainy": [511, 615, 616],
"sunny": [800],
"windy": [905, 951, 952, 953, 954, 955, 956, 957],
"windy-variant": [958, 959, 960, 961],
"exceptional": [711, 721, 731, 751, 761, 762, 771, 900, 901, 962, 903, 904],
}
WEATHER_SENSOR_TYPES = {
ATTR_API_WEATHER: {SENSOR_NAME: "Weather"},
ATTR_API_TEMPERATURE: {
SENSOR_NAME: "Temperature",
SENSOR_UNIT: TEMP_CELSIUS,
SENSOR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
},
ATTR_API_WIND_SPEED: {
SENSOR_NAME: "Wind speed",
SENSOR_UNIT: SPEED_METERS_PER_SECOND,
},
ATTR_API_WIND_BEARING: {SENSOR_NAME: "Wind bearing", SENSOR_UNIT: DEGREE},
ATTR_API_HUMIDITY: {
SENSOR_NAME: "Humidity",
SENSOR_UNIT: UNIT_PERCENTAGE,
SENSOR_DEVICE_CLASS: DEVICE_CLASS_HUMIDITY,
},
ATTR_API_PRESSURE: {
SENSOR_NAME: "Pressure",
SENSOR_UNIT: PRESSURE_PA,
SENSOR_DEVICE_CLASS: DEVICE_CLASS_PRESSURE,
},
ATTR_API_CLOUDS: {SENSOR_NAME: "Cloud coverage", SENSOR_UNIT: UNIT_PERCENTAGE},
ATTR_API_RAIN: {SENSOR_NAME: "Rain", SENSOR_UNIT: "mm"},
ATTR_API_SNOW: {SENSOR_NAME: "Snow", SENSOR_UNIT: "mm"},
ATTR_API_CONDITION: {SENSOR_NAME: "Condition"},
ATTR_API_WEATHER_CODE: {SENSOR_NAME: "Weather Code"},
}
FORECAST_SENSOR_TYPES = {
ATTR_FORECAST_CONDITION: {SENSOR_NAME: "Condition"},
ATTR_FORECAST_PRECIPITATION: {SENSOR_NAME: "Precipitation"},
ATTR_FORECAST_TEMP: {
SENSOR_NAME: "Temperature",
SENSOR_UNIT: TEMP_CELSIUS,
SENSOR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
},
ATTR_FORECAST_TEMP_LOW: {
SENSOR_NAME: "Temperature Low",
SENSOR_UNIT: TEMP_CELSIUS,
SENSOR_DEVICE_CLASS: DEVICE_CLASS_TEMPERATURE,
},
ATTR_FORECAST_TIME: {
SENSOR_NAME: "Time",
SENSOR_DEVICE_CLASS: DEVICE_CLASS_TIMESTAMP,
},
ATTR_API_WIND_BEARING: {SENSOR_NAME: "Wind bearing", SENSOR_UNIT: DEGREE},
ATTR_API_WIND_SPEED: {
SENSOR_NAME: "Wind speed",
SENSOR_UNIT: SPEED_METERS_PER_SECOND,
},
}

137
homeassistant/components/openweathermap/forecast_update_coordinator.py Normal file
View File

@ -0,0 +1,137 @@
"""Forecast data coordinator for the OpenWeatherMap (OWM) service."""
from datetime import timedelta
import logging
import async_timeout
from pyowm.exceptions.api_call_error import APICallError
from pyowm.exceptions.api_response_error import UnauthorizedError
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
ATTR_API_FORECAST,
ATTR_API_THIS_DAY_FORECAST,
CONDITION_CLASSES,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
FORECAST_UPDATE_INTERVAL = timedelta(minutes=30)
class ForecastUpdateCoordinator(DataUpdateCoordinator):
"""Forecast data update coordinator."""
def __init__(self, owm, latitude, longitude, forecast_mode, hass):
"""Initialize coordinator."""
self._owm_client = owm
self._forecast_mode = forecast_mode
self._latitude = latitude
self._longitude = longitude
self._forecast_limit = 15
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=FORECAST_UPDATE_INTERVAL
)
async def _async_update_data(self):
data = {}
with async_timeout.timeout(20):
try:
forecast_response = await self._get_owm_forecast()
data = self._convert_forecast_response(forecast_response)
except (APICallError, UnauthorizedError) as error:
raise UpdateFailed(error) from error
return data
async def _get_owm_forecast(self):
if self._forecast_mode == "daily":
forecast_response = await self.hass.async_add_executor_job(
self._owm_client.daily_forecast_at_coords,
self._latitude,
self._longitude,
self._forecast_limit,
)
else:
forecast_response = await self.hass.async_add_executor_job(
self._owm_client.three_hours_forecast_at_coords,
self._latitude,
self._longitude,
)
return forecast_response.get_forecast()
def _convert_forecast_response(self, forecast_response):
weathers = self._get_weathers(forecast_response)
forecast_entries = self._convert_forecast_entries(weathers)
return {
ATTR_API_FORECAST: forecast_entries,
ATTR_API_THIS_DAY_FORECAST: forecast_entries[0],
}
def _get_weathers(self, forecast_response):
if self._forecast_mode == "freedaily":
return forecast_response.get_weathers()[::8]
return forecast_response.get_weathers()
def _convert_forecast_entries(self, entries):
if self._forecast_mode == "daily":
return list(map(self._convert_daily_forecast, entries))
return list(map(self._convert_forecast, entries))
def _convert_daily_forecast(self, entry):
return {
ATTR_FORECAST_TIME: entry.get_reference_time("unix") * 1000,
ATTR_FORECAST_TEMP: entry.get_temperature("celsius").get("day"),
ATTR_FORECAST_TEMP_LOW: entry.get_temperature("celsius").get("night"),
ATTR_FORECAST_PRECIPITATION: self._calc_daily_precipitation(
entry.get_rain().get("all"), entry.get_snow().get("all")
),
ATTR_FORECAST_WIND_SPEED: entry.get_wind().get("speed"),
ATTR_FORECAST_WIND_BEARING: entry.get_wind().get("deg"),
ATTR_FORECAST_CONDITION: self._get_condition(entry.get_weather_code()),
}
def _convert_forecast(self, entry):
return {
ATTR_FORECAST_TIME: entry.get_reference_time("unix") * 1000,
ATTR_FORECAST_TEMP: entry.get_temperature("celsius").get("temp"),
ATTR_FORECAST_PRECIPITATION: self._calc_precipitation(entry),
ATTR_FORECAST_WIND_SPEED: entry.get_wind().get("speed"),
ATTR_FORECAST_WIND_BEARING: entry.get_wind().get("deg"),
ATTR_FORECAST_CONDITION: self._get_condition(entry.get_weather_code()),
}
@staticmethod
def _calc_daily_precipitation(rain, snow):
"""Calculate the precipitation."""
rain_value = 0 if rain is None else rain
snow_value = 0 if snow is None else snow
if round(rain_value + snow_value, 1) == 0:
return None
return round(rain_value + snow_value, 1)
@staticmethod
def _calc_precipitation(entry):
return (
round(entry.get_rain().get("1h"), 1)
if entry.get_rain().get("1h") is not None
and (round(entry.get_rain().get("1h"), 1) > 0)
else None
)
@staticmethod
def _get_condition(weather_code):
return [k for k, v in CONDITION_CLASSES.items() if weather_code in v][0]

5
homeassistant/components/openweathermap/manifest.json
View File

@ -1,7 +1,8 @@
{
"domain": "openweathermap",
"name": "Openweathermap",
"name": "OpenWeatherMap",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/openweathermap",
"requirements": ["pyowm==2.10.0"],
"codeowners": ["@fabaff"]
"codeowners": ["@fabaff", "@freekode"]
}

282
homeassistant/components/openweathermap/sensor.py
View File

@ -1,227 +1,105 @@
"""Support for the OpenWeatherMap (OWM) service."""
from datetime import timedelta
import logging
from pyowm import OWM
from pyowm.exceptions.api_call_error import APICallError
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_API_KEY,
CONF_MONITORED_CONDITIONS,
CONF_NAME,
DEGREE,
SPEED_METERS_PER_SECOND,
TEMP_CELSIUS,
TEMP_FAHRENHEIT,
UNIT_PERCENTAGE,
from .abstract_owm_sensor import AbstractOpenWeatherMapSensor
from .const import (
ATTR_API_THIS_DAY_FORECAST,
DOMAIN,
ENTRY_FORECAST_COORDINATOR,
ENTRY_NAME,
ENTRY_WEATHER_COORDINATOR,
FORECAST_MONITORED_CONDITIONS,
FORECAST_SENSOR_TYPES,
MONITORED_CONDITIONS,
WEATHER_SENSOR_TYPES,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
from .forecast_update_coordinator import ForecastUpdateCoordinator
from .weather_update_coordinator import WeatherUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by OpenWeatherMap"
CONF_FORECAST = "forecast"
CONF_LANGUAGE = "language"
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up OpenWeatherMap sensor entities based on a config entry."""
domain_data = hass.data[DOMAIN][config_entry.entry_id]
name = domain_data[ENTRY_NAME]
weather_coordinator = domain_data[ENTRY_WEATHER_COORDINATOR]
forecast_coordinator = domain_data[ENTRY_FORECAST_COORDINATOR]
DEFAULT_NAME = "OWM"
weather_sensor_types = WEATHER_SENSOR_TYPES
forecast_sensor_types = FORECAST_SENSOR_TYPES
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=120)
SENSOR_TYPES = {
"weather": ["Condition", None],
"temperature": ["Temperature", None],
"wind_speed": ["Wind speed", SPEED_METERS_PER_SECOND],
"wind_bearing": ["Wind bearing", DEGREE],
"humidity": ["Humidity", UNIT_PERCENTAGE],
"pressure": ["Pressure", "mbar"],
"clouds": ["Cloud coverage", UNIT_PERCENTAGE],
"rain": ["Rain", "mm"],
"snow": ["Snow", "mm"],
"weather_code": ["Weather code", None],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_API_KEY): cv.string,
vol.Optional(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]
),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_FORECAST, default=False): cv.boolean,
vol.Optional(CONF_LANGUAGE): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the OpenWeatherMap sensor."""
if None in (hass.config.latitude, hass.config.longitude):
_LOGGER.error("Latitude or longitude not set in Home Assistant config")
return
SENSOR_TYPES["temperature"][1] = hass.config.units.temperature_unit
name = config.get(CONF_NAME)
forecast = config.get(CONF_FORECAST)
language = config.get(CONF_LANGUAGE)
if isinstance(language, str):
language = language.lower()[:2]
owm = OWM(API_key=config.get(CONF_API_KEY), language=language)
if not owm:
_LOGGER.error("Unable to connect to OpenWeatherMap")
return
data = WeatherData(owm, forecast, hass.config.latitude, hass.config.longitude)
dev = []
for variable in config[CONF_MONITORED_CONDITIONS]:
dev.append(
OpenWeatherMapSensor(name, data, variable, SENSOR_TYPES[variable][1])
entities = []
for sensor_type in MONITORED_CONDITIONS:
unique_id = f"{config_entry.unique_id}-{sensor_type}"
entities.append(
OpenWeatherMapSensor(
name,
unique_id,
sensor_type,
weather_sensor_types[sensor_type],
weather_coordinator,
)
)
if forecast:
SENSOR_TYPES["forecast"] = ["Forecast", None]
dev.append(
OpenWeatherMapSensor(name, data, "forecast", SENSOR_TYPES["temperature"][1])
for sensor_type in FORECAST_MONITORED_CONDITIONS:
unique_id = f"{config_entry.unique_id}-forecast-{sensor_type}"
entities.append(
OpenWeatherMapForecastSensor(
f"{name} Forecast",
unique_id,
sensor_type,
forecast_sensor_types[sensor_type],
forecast_coordinator,
)
)
add_entities(dev, True)
async_add_entities(entities)
class OpenWeatherMapSensor(Entity):
class OpenWeatherMapSensor(AbstractOpenWeatherMapSensor):
"""Implementation of an OpenWeatherMap sensor."""
def __init__(self, name, weather_data, sensor_type, temp_unit):
def __init__(
self,
name,
unique_id,
sensor_type,
sensor_configuration,
weather_coordinator: WeatherUpdateCoordinator,
):
"""Initialize the sensor."""
self.client_name = name
self._name = SENSOR_TYPES[sensor_type][0]
self.owa_client = weather_data
self.temp_unit = temp_unit
self.type = sensor_type
self._state = None
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
@property
def name(self):
"""Return the name of the sensor."""
return f"{self.client_name} {self._name}"
super().__init__(
name, unique_id, sensor_type, sensor_configuration, weather_coordinator
)
self._weather_coordinator = weather_coordinator
@property
def state(self):
"""Return the state of the device."""
return self._state
return self._weather_coordinator.data.get(self._sensor_type, None)
class OpenWeatherMapForecastSensor(AbstractOpenWeatherMapSensor):
"""Implementation of an OpenWeatherMap this day forecast sensor."""
def __init__(
self,
name,
unique_id,
sensor_type,
sensor_configuration,
forecast_coordinator: ForecastUpdateCoordinator,
):
"""Initialize the sensor."""
super().__init__(
name, unique_id, sensor_type, sensor_configuration, forecast_coordinator
)
self._forecast_coordinator = forecast_coordinator
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
@property
def device_state_attributes(self):
"""Return the state attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}
def update(self):
"""Get the latest data from OWM and updates the states."""
try:
self.owa_client.update()
except APICallError:
_LOGGER.error("Error when calling API to update data")
return
data = self.owa_client.data
fc_data = self.owa_client.fc_data
if data is None:
return
try:
if self.type == "weather":
self._state = data.get_detailed_status()
elif self.type == "temperature":
if self.temp_unit == TEMP_CELSIUS:
self._state = round(data.get_temperature("celsius")["temp"], 1)
elif self.temp_unit == TEMP_FAHRENHEIT:
self._state = round(data.get_temperature("fahrenheit")["temp"], 1)
else:
self._state = round(data.get_temperature()["temp"], 1)
elif self.type == "wind_speed":
self._state = round(data.get_wind()["speed"], 1)
elif self.type == "wind_bearing":
self._state = round(data.get_wind()["deg"], 1)
elif self.type == "humidity":
self._state = round(data.get_humidity(), 1)
elif self.type == "pressure":
self._state = round(data.get_pressure()["press"], 0)
elif self.type == "clouds":
self._state = data.get_clouds()
elif self.type == "rain":
rain = data.get_rain()
if "1h" in rain:
self._state = round(rain["1h"], 0)
self._unit_of_measurement = "mm"
else:
self._state = "not raining"
self._unit_of_measurement = ""
elif self.type == "snow":
snow = data.get_snow()
if "1h" in snow:
self._state = round(snow["1h"], 0)
self._unit_of_measurement = "mm"
else:
self._state = "not snowing"
self._unit_of_measurement = ""
elif self.type == "forecast":
if fc_data is None:
return
self._state = fc_data.get_weathers()[0].get_detailed_status()
elif self.type == "weather_code":
self._state = data.get_weather_code()
except KeyError:
self._state = None
_LOGGER.warning("Condition is currently not available: %s", self.type)
class WeatherData:
"""Get the latest data from OpenWeatherMap."""
def __init__(self, owm, forecast, latitude, longitude):
"""Initialize the data object."""
self.owm = owm
self.forecast = forecast
self.latitude = latitude
self.longitude = longitude
self.data = None
self.fc_data = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Get the latest data from OpenWeatherMap."""
try:
obs = self.owm.weather_at_coords(self.latitude, self.longitude)
except (APICallError, TypeError):
_LOGGER.error("Error when calling API to get weather at coordinates")
obs = None
if obs is None:
_LOGGER.warning("Failed to fetch data")
return
self.data = obs.get_weather()
if self.forecast == 1:
try:
obs = self.owm.three_hours_forecast_at_coords(
self.latitude, self.longitude
)
self.fc_data = obs.get_forecast()
except (ConnectionResetError, TypeError):
_LOGGER.warning("Failed to fetch forecast")
def state(self):
"""Return the state of the device."""
return self._forecast_coordinator.data[ATTR_API_THIS_DAY_FORECAST].get(
self._sensor_type, None
)

35
homeassistant/components/openweathermap/strings.json Normal file
View File

@ -0,0 +1,35 @@
{
"config": {
"abort": {
"already_configured": "OpenWeatherMap integration for these coordinates is already configured."
},
"error": {
"auth": "API key is not correct.",
"connection": "Can't connect to OWM API"
},
"step": {
"user": {
"data": {
"api_key": "OpenWeatherMap API key",
"language": "Language",
"latitude": "Latitude",
"longitude": "Longitude",
"mode": "Mode",
"name": "Name of the integration"
},
"description": "Set up OpenWeatherMap integration. To generate API key go to https://openweathermap.org/appid",
"title": "OpenWeatherMap"
}
}
},
"options": {
"step": {
"init": {
"data": {
"language": "Language",
"mode": "Mode"
}
}
}
}
}

35
homeassistant/components/openweathermap/translations/en.json Normal file
View File

@ -0,0 +1,35 @@
{
"config": {
"abort": {
"already_configured": "OpenWeatherMap integration for these coordinates is already configured."
},
"error": {
"auth": "API key is not correct.",
"connection": "Can't connect to OWM API"
},
"step": {
"user": {
"data": {
"api_key": "OpenWeatherMap API key",
"language": "Language",
"latitude": "Latitude",
"longitude": "Longitude",
"mode": "Mode",
"name": "Name of the integration"
},
"description": "Set up OpenWeatherMap integration. To generate API key go to https://openweathermap.org/appid",
"title": "OpenWeatherMap"
}
}
},
"options": {
"step": {
"init": {
"data": {
"language": "Language",
"mode": "Mode"
}
}
}
}
}

299
homeassistant/components/openweathermap/weather.py
View File

@ -1,131 +1,89 @@
"""Support for the OpenWeatherMap (OWM) service."""
from datetime import timedelta
import logging
from pyowm import OWM
from pyowm.exceptions.api_call_error import APICallError
import voluptuous as vol
from homeassistant.components.weather import WeatherEntity
from homeassistant.const import TEMP_CELSIUS
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
PLATFORM_SCHEMA,
WeatherEntity,
from .const import (
ATTR_API_CONDITION,
ATTR_API_FORECAST,
ATTR_API_HUMIDITY,
ATTR_API_PRESSURE,
ATTR_API_TEMPERATURE,
ATTR_API_WIND_BEARING,
ATTR_API_WIND_SPEED,
ATTRIBUTION,
DOMAIN,
ENTRY_FORECAST_COORDINATOR,
ENTRY_NAME,
ENTRY_WEATHER_COORDINATOR,
)
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_MODE,
CONF_NAME,
PRESSURE_HPA,
PRESSURE_INHG,
STATE_UNKNOWN,
TEMP_CELSIUS,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle
from homeassistant.util.pressure import convert as convert_pressure
from .forecast_update_coordinator import ForecastUpdateCoordinator
from .weather_update_coordinator import WeatherUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by OpenWeatherMap"
FORECAST_MODE = ["hourly", "daily", "freedaily"]
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up OpenWeatherMap weather entity based on a config entry."""
domain_data = hass.data[DOMAIN][config_entry.entry_id]
name = domain_data[ENTRY_NAME]
weather_coordinator = domain_data[ENTRY_WEATHER_COORDINATOR]
forecast_coordinator = domain_data[ENTRY_FORECAST_COORDINATOR]
DEFAULT_NAME = "OpenWeatherMap"
MIN_TIME_BETWEEN_FORECAST_UPDATES = timedelta(minutes=30)
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=10)
CONDITION_CLASSES = {
"cloudy": [803, 804],
"fog": [701, 741],
"hail": [906],
"lightning": [210, 211, 212, 221],
"lightning-rainy": [200, 201, 202, 230, 231, 232],
"partlycloudy": [801, 802],
"pouring": [504, 314, 502, 503, 522],
"rainy": [300, 301, 302, 310, 311, 312, 313, 500, 501, 520, 521],
"snowy": [600, 601, 602, 611, 612, 620, 621, 622],
"snowy-rainy": [511, 615, 616],
"sunny": [800],
"windy": [905, 951, 952, 953, 954, 955, 956, 957],
"windy-variant": [958, 959, 960, 961],
"exceptional": [711, 721, 731, 751, 761, 762, 771, 900, 901, 962, 903, 904],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_API_KEY): cv.string,
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_MODE, default="hourly"): vol.In(FORECAST_MODE),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the OpenWeatherMap weather platform."""
longitude = config.get(CONF_LONGITUDE, round(hass.config.longitude, 5))
latitude = config.get(CONF_LATITUDE, round(hass.config.latitude, 5))
name = config.get(CONF_NAME)
mode = config.get(CONF_MODE)
try:
owm = OWM(config.get(CONF_API_KEY))
except APICallError:
_LOGGER.error("Error while connecting to OpenWeatherMap")
return False
data = WeatherData(owm, latitude, longitude, mode)
add_entities(
[OpenWeatherMapWeather(name, data, hass.config.units.temperature_unit, mode)],
True,
unique_id = f"{config_entry.unique_id}"
owm_weather = OpenWeatherMapWeather(
name, unique_id, weather_coordinator, forecast_coordinator
)
async_add_entities([owm_weather], False)
class OpenWeatherMapWeather(WeatherEntity):
"""Implementation of an OpenWeatherMap sensor."""
def __init__(self, name, owm, temperature_unit, mode):
def __init__(
self,
name,
unique_id,
weather_coordinator: WeatherUpdateCoordinator,
forecast_coordinator: ForecastUpdateCoordinator,
):
"""Initialize the sensor."""
self._name = name
self._owm = owm
self._temperature_unit = temperature_unit
self._mode = mode
self.data = None
self.forecast_data = None
self._unique_id = unique_id
self._weather_coordinator = weather_coordinator
self._forecast_coordinator = forecast_coordinator
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return self._unique_id
@property
def should_poll(self):
"""Return the polling requirement of the entity."""
return False
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def condition(self):
"""Return the current condition."""
try:
return [
k
for k, v in CONDITION_CLASSES.items()
if self.data.get_weather_code() in v
][0]
except IndexError:
return STATE_UNKNOWN
return self._weather_coordinator.data[ATTR_API_CONDITION]
@property
def temperature(self):
"""Return the temperature."""
return self.data.get_temperature("celsius").get("temp")
return self._weather_coordinator.data[ATTR_API_TEMPERATURE]
@property
def temperature_unit(self):
@ -135,146 +93,49 @@ class OpenWeatherMapWeather(WeatherEntity):
@property
def pressure(self):
"""Return the pressure."""
pressure = self.data.get_pressure().get("press")
if self.hass.config.units.name == "imperial":
return round(convert_pressure(pressure, PRESSURE_HPA, PRESSURE_INHG), 2)
return pressure
return self._weather_coordinator.data[ATTR_API_PRESSURE]
@property
def humidity(self):
"""Return the humidity."""
return self.data.get_humidity()
return self._weather_coordinator.data[ATTR_API_HUMIDITY]
@property
def wind_speed(self):
"""Return the wind speed."""
wind_speed = self._weather_coordinator.data[ATTR_API_WIND_SPEED]
if self.hass.config.units.name == "imperial":
return round(self.data.get_wind().get("speed") * 2.24, 2)
return round(self.data.get_wind().get("speed") * 3.6, 2)
return round(wind_speed * 2.24, 2)
return round(wind_speed * 3.6, 2)
@property
def wind_bearing(self):
"""Return the wind bearing."""
return self.data.get_wind().get("deg")
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
return self._weather_coordinator.data[ATTR_API_WIND_BEARING]
@property
def forecast(self):
"""Return the forecast array."""
data = []
return self._forecast_coordinator.data[ATTR_API_FORECAST]
def calc_precipitation(rain, snow):
"""Calculate the precipitation."""
rain_value = 0 if rain is None else rain
snow_value = 0 if snow is None else snow
if round(rain_value + snow_value, 1) == 0:
return None
return round(rain_value + snow_value, 1)
@property
def available(self):
"""Return True if entity is available."""
return (
self._weather_coordinator.last_update_success
and self._forecast_coordinator.last_update_success
)
if self._mode == "freedaily":
weather = self.forecast_data.get_weathers()[::8]
else:
weather = self.forecast_data.get_weathers()
async def async_added_to_hass(self):
"""Connect to dispatcher listening for entity data notifications."""
self.async_on_remove(
self._weather_coordinator.async_add_listener(self.async_write_ha_state)
)
self.async_on_remove(
self._forecast_coordinator.async_add_listener(self.async_write_ha_state)
)
for entry in weather:
if self._mode == "daily":
data.append(
{
ATTR_FORECAST_TIME: entry.get_reference_time("unix") * 1000,
ATTR_FORECAST_TEMP: entry.get_temperature("celsius").get("day"),
ATTR_FORECAST_TEMP_LOW: entry.get_temperature("celsius").get(
"night"
),
ATTR_FORECAST_PRECIPITATION: calc_precipitation(
entry.get_rain().get("all"), entry.get_snow().get("all")
),
ATTR_FORECAST_WIND_SPEED: entry.get_wind().get("speed"),
ATTR_FORECAST_WIND_BEARING: entry.get_wind().get("deg"),
ATTR_FORECAST_CONDITION: [
k
for k, v in CONDITION_CLASSES.items()
if entry.get_weather_code() in v
][0],
}
)
else:
rain = entry.get_rain().get("1h")
if rain is not None:
rain = round(rain, 1)
data.append(
{
ATTR_FORECAST_TIME: entry.get_reference_time("unix") * 1000,
ATTR_FORECAST_TEMP: entry.get_temperature("celsius").get(
"temp"
),
ATTR_FORECAST_PRECIPITATION: rain,
ATTR_FORECAST_CONDITION: [
k
for k, v in CONDITION_CLASSES.items()
if entry.get_weather_code() in v
][0],
}
)
return data
def update(self):
async def async_update(self):
"""Get the latest data from OWM and updates the states."""
try:
self._owm.update()
self._owm.update_forecast()
except APICallError:
_LOGGER.error("Exception when calling OWM web API to update data")
return
self.data = self._owm.data
self.forecast_data = self._owm.forecast_data
class WeatherData:
"""Get the latest data from OpenWeatherMap."""
def __init__(self, owm, latitude, longitude, mode):
"""Initialize the data object."""
self._mode = mode
self.owm = owm
self.latitude = latitude
self.longitude = longitude
self.data = None
self.forecast_data = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Get the latest data from OpenWeatherMap."""
obs = self.owm.weather_at_coords(self.latitude, self.longitude)
if obs is None:
_LOGGER.warning("Failed to fetch data from OWM")
return
self.data = obs.get_weather()
@Throttle(MIN_TIME_BETWEEN_FORECAST_UPDATES)
def update_forecast(self):
"""Get the latest forecast from OpenWeatherMap."""
try:
if self._mode == "daily":
fcd = self.owm.daily_forecast_at_coords(
self.latitude, self.longitude, 15
)
else:
fcd = self.owm.three_hours_forecast_at_coords(
self.latitude, self.longitude
)
except APICallError:
_LOGGER.error("Exception when calling OWM web API to update forecast")
return
if fcd is None:
_LOGGER.warning("Failed to fetch forecast data from OWM")
return
self.forecast_data = fcd.get_forecast()
await self._weather_coordinator.async_request_refresh()
await self._forecast_coordinator.async_request_refresh()

94
homeassistant/components/openweathermap/weather_update_coordinator.py Normal file
View File

@ -0,0 +1,94 @@
"""Weather data coordinator for the OpenWeatherMap (OWM) service."""
from datetime import timedelta
import logging
import async_timeout
from pyowm.exceptions.api_call_error import APICallError
from pyowm.exceptions.api_response_error import UnauthorizedError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
ATTR_API_CLOUDS,
ATTR_API_CONDITION,
ATTR_API_HUMIDITY,
ATTR_API_PRESSURE,
ATTR_API_RAIN,
ATTR_API_SNOW,
ATTR_API_TEMPERATURE,
ATTR_API_WEATHER,
ATTR_API_WEATHER_CODE,
ATTR_API_WIND_BEARING,
ATTR_API_WIND_SPEED,
CONDITION_CLASSES,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
WEATHER_UPDATE_INTERVAL = timedelta(minutes=10)
class WeatherUpdateCoordinator(DataUpdateCoordinator):
"""Weather data update coordinator."""
def __init__(self, owm, latitude, longitude, hass):
"""Initialize coordinator."""
self._owm_client = owm
self._latitude = latitude
self._longitude = longitude
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=WEATHER_UPDATE_INTERVAL
)
async def _async_update_data(self):
data = {}
with async_timeout.timeout(20):
try:
weather_response = await self._get_owm_weather()
data = self._convert_weather_response(weather_response)
except (APICallError, UnauthorizedError) as error:
raise UpdateFailed(error) from error
return data
async def _get_owm_weather(self):
weather = await self.hass.async_add_executor_job(
self._owm_client.weather_at_coords, self._latitude, self._longitude
)
return weather.get_weather()
def _convert_weather_response(self, weather_response):
return {
ATTR_API_TEMPERATURE: weather_response.get_temperature("celsius").get(
"temp"
),
ATTR_API_PRESSURE: weather_response.get_pressure().get("press"),
ATTR_API_HUMIDITY: weather_response.get_humidity(),
ATTR_API_WIND_BEARING: weather_response.get_wind().get("deg"),
ATTR_API_WIND_SPEED: weather_response.get_wind().get("speed"),
ATTR_API_CLOUDS: weather_response.get_clouds(),
ATTR_API_RAIN: self._get_rain(weather_response.get_rain()),
ATTR_API_SNOW: self._get_snow(weather_response.get_snow()),
ATTR_API_WEATHER: weather_response.get_detailed_status(),
ATTR_API_CONDITION: self._get_condition(
weather_response.get_weather_code()
),
ATTR_API_WEATHER_CODE: weather_response.get_weather_code(),
}
@staticmethod
def _get_rain(rain):
if "1h" in rain:
return round(rain["1h"], 0)
return "not raining"
@staticmethod
def _get_snow(snow):
if snow:
return round(snow, 0)
return "not snowing"
@staticmethod
def _get_condition(weather_code):
return [k for k, v in CONDITION_CLASSES.items() if weather_code in v][0]

1
homeassistant/generated/config_flows.py
View File

@ -128,6 +128,7 @@ FLOWS = [
"onvif",
"opentherm_gw",
"openuv",
"openweathermap",
"ovo_energy",
"owntracks",
"ozw",

3
requirements_test_all.txt
View File

@ -756,6 +756,9 @@ pyotgw==0.6b1
# homeassistant.components.otp
pyotp==2.3.0
# homeassistant.components.openweathermap
pyowm==2.10.0
# homeassistant.components.point
pypoint==1.1.2

1
tests/components/openweathermap/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the OpenWeatherMap integration."""

232
tests/components/openweathermap/test_config_flow.py Normal file
View File

@ -0,0 +1,232 @@
"""Define tests for the OpenWeatherMap config flow."""
from asynctest import MagicMock, patch
from pyowm.exceptions.api_call_error import APICallError
from pyowm.exceptions.api_response_error import UnauthorizedError
from homeassistant import data_entry_flow
from homeassistant.components.openweathermap.const import (
CONF_LANGUAGE,
DEFAULT_FORECAST_MODE,
DEFAULT_LANGUAGE,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_MODE,
CONF_NAME,
)
from tests.common import MockConfigEntry
CONFIG = {
CONF_NAME: "openweathermap",
CONF_API_KEY: "foo",
CONF_LATITUDE: 50,
CONF_LONGITUDE: 40,
CONF_MODE: DEFAULT_FORECAST_MODE,
CONF_LANGUAGE: DEFAULT_LANGUAGE,
}
VALID_YAML_CONFIG = {CONF_API_KEY: "foo"}
async def test_form(hass):
"""Test that the form is served with valid input."""
mocked_owm = _create_mocked_owm(True)
with patch(
"pyowm.weatherapi25.owm25.OWM25",
return_value=mocked_owm,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == SOURCE_USER
assert result["errors"] == {}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
)
await hass.async_block_till_done()
conf_entries = hass.config_entries.async_entries(DOMAIN)
entry = conf_entries[0]
assert entry.state == "loaded"
await hass.config_entries.async_unload(conf_entries[0].entry_id)
await hass.async_block_till_done()
assert entry.state == "not_loaded"
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == CONFIG[CONF_NAME]
assert result["data"][CONF_LATITUDE] == CONFIG[CONF_LATITUDE]
assert result["data"][CONF_LONGITUDE] == CONFIG[CONF_LONGITUDE]
assert result["data"][CONF_API_KEY] == CONFIG[CONF_API_KEY]
async def test_form_import(hass):
"""Test we can import yaml config."""
mocked_owm = _create_mocked_owm(True)
with patch("pyowm.weatherapi25.owm25.OWM25", return_value=mocked_owm), patch(
"homeassistant.components.openweathermap.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.openweathermap.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=VALID_YAML_CONFIG.copy(),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_LATITUDE] == hass.config.latitude
assert result["data"][CONF_LONGITUDE] == hass.config.longitude
assert result["data"][CONF_API_KEY] == VALID_YAML_CONFIG[CONF_API_KEY]
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_options(hass):
"""Test that the options form."""
mocked_owm = _create_mocked_owm(True)
with patch(
"pyowm.weatherapi25.owm25.OWM25",
return_value=mocked_owm,
):
config_entry = MockConfigEntry(
domain=DOMAIN, unique_id="openweathermap_unique_id", data=CONFIG
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == "loaded"
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_MODE: "daily"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {
CONF_MODE: "daily",
CONF_LANGUAGE: DEFAULT_LANGUAGE,
}
await hass.async_block_till_done()
assert config_entry.state == "loaded"
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_MODE: "freedaily"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {
CONF_MODE: "freedaily",
CONF_LANGUAGE: DEFAULT_LANGUAGE,
}
await hass.async_block_till_done()
assert config_entry.state == "loaded"
async def test_form_invalid_api_key(hass):
"""Test that the form is served with no input."""
mocked_owm = _create_mocked_owm(True)
with patch(
"pyowm.weatherapi25.owm25.OWM25",
return_value=mocked_owm,
side_effect=UnauthorizedError(""),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
)
assert result["errors"] == {"base": "auth"}
async def test_form_api_call_error(hass):
"""Test setting up with api call error."""
mocked_owm = _create_mocked_owm(True)
with patch(
"pyowm.weatherapi25.owm25.OWM25",
return_value=mocked_owm,
side_effect=APICallError(""),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
)
assert result["errors"] == {"base": "connection"}
async def test_form_api_offline(hass):
"""Test setting up with api call error."""
mocked_owm = _create_mocked_owm(False)
with patch(
"homeassistant.components.openweathermap.config_flow.OWM",
return_value=mocked_owm,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
)
assert result["errors"] == {"base": "auth"}
def _create_mocked_owm(is_api_online: bool):
mocked_owm = MagicMock()
mocked_owm.is_API_online.return_value = is_api_online
weather = MagicMock()
weather.get_temperature.return_value.get.return_value = 10
weather.get_pressure.return_value.get.return_value = 10
weather.get_humidity.return_value = 10
weather.get_wind.return_value.get.return_value = 0
weather.get_clouds.return_value = "clouds"
weather.get_rain.return_value = []
weather.get_snow.return_value = 3
weather.get_detailed_status.return_value = "status"
weather.get_weather_code.return_value = 803
mocked_owm.weather_at_coords.return_value.get_weather.return_value = weather
one_day_forecast = MagicMock()
one_day_forecast.get_reference_time.return_value = 10
one_day_forecast.get_temperature.return_value.get.return_value = 10
one_day_forecast.get_rain.return_value.get.return_value = 0
one_day_forecast.get_snow.return_value.get.return_value = 0
one_day_forecast.get_wind.return_value.get.return_value = 0
one_day_forecast.get_weather_code.return_value = 803
mocked_owm.three_hours_forecast_at_coords.return_value.get_forecast.return_value.get_weathers.return_value = [
one_day_forecast
]
return mocked_owm