Add ClimaCell weather integration (#36547)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Raman Gupta 2021-02-24 03:34:27 -05:00 committed by GitHub
parent a632215541
commit eccdae60bf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 1144 additions and 0 deletions

View file

@ -145,6 +145,7 @@ omit =
homeassistant/components/clickatell/notify.py
homeassistant/components/clicksend/notify.py
homeassistant/components/clicksend_tts/notify.py
homeassistant/components/climacell/weather.py
homeassistant/components/cmus/media_player.py
homeassistant/components/co2signal/*
homeassistant/components/coinbase/*

View file

@ -83,6 +83,7 @@ homeassistant/components/circuit/* @braam
homeassistant/components/cisco_ios/* @fbradyirl
homeassistant/components/cisco_mobility_express/* @fbradyirl
homeassistant/components/cisco_webex_teams/* @fbradyirl
homeassistant/components/climacell/* @raman325
homeassistant/components/cloud/* @home-assistant/cloud
homeassistant/components/cloudflare/* @ludeeus @ctalkington
homeassistant/components/color_extractor/* @GenericStudent

View file

@ -0,0 +1,262 @@
"""The ClimaCell integration."""
import asyncio
from datetime import timedelta
import logging
from math import ceil
from typing import Any, Dict, Optional, Union
from pyclimacell import ClimaCell
from pyclimacell.const import (
FORECAST_DAILY,
FORECAST_HOURLY,
FORECAST_NOWCAST,
REALTIME,
)
from pyclimacell.pyclimacell import (
CantConnectException,
InvalidAPIKeyException,
RateLimitedException,
UnknownException,
)
from homeassistant.components.weather import DOMAIN as WEATHER_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from .const import (
ATTRIBUTION,
CONF_TIMESTEP,
CURRENT,
DAILY,
DEFAULT_TIMESTEP,
DOMAIN,
FORECASTS,
HOURLY,
MAX_REQUESTS_PER_DAY,
NOWCAST,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [WEATHER_DOMAIN]
def _set_update_interval(
hass: HomeAssistantType, current_entry: ConfigEntry
) -> timedelta:
"""Recalculate update_interval based on existing ClimaCell instances and update them."""
# We check how many ClimaCell configured instances are using the same API key and
# calculate interval to not exceed allowed numbers of requests. Divide 90% of
# MAX_REQUESTS_PER_DAY by 4 because every update requires four API calls and we want
# a buffer in the number of API calls left at the end of the day.
other_instance_entry_ids = [
entry.entry_id
for entry in hass.config_entries.async_entries(DOMAIN)
if entry.entry_id != current_entry.entry_id
and entry.data[CONF_API_KEY] == current_entry.data[CONF_API_KEY]
]
interval = timedelta(
minutes=(
ceil(
(24 * 60 * (len(other_instance_entry_ids) + 1) * 4)
/ (MAX_REQUESTS_PER_DAY * 0.9)
)
)
)
for entry_id in other_instance_entry_ids:
if entry_id in hass.data[DOMAIN]:
hass.data[DOMAIN][entry_id].update_interval = interval
return interval
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
"""Set up the ClimaCell API component."""
return True
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry) -> bool:
"""Set up ClimaCell API from a config entry."""
hass.data.setdefault(DOMAIN, {})
# If config entry options not set up, set them up
if not config_entry.options:
hass.config_entries.async_update_entry(
config_entry,
options={
CONF_TIMESTEP: DEFAULT_TIMESTEP,
},
)
coordinator = ClimaCellDataUpdateCoordinator(
hass,
config_entry,
ClimaCell(
config_entry.data[CONF_API_KEY],
config_entry.data.get(CONF_LATITUDE, hass.config.latitude),
config_entry.data.get(CONF_LONGITUDE, hass.config.longitude),
session=async_get_clientsession(hass),
),
_set_update_interval(hass, config_entry),
)
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
hass.data[DOMAIN][config_entry.entry_id] = coordinator
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
return True
async def async_unload_entry(
hass: HomeAssistantType, config_entry: ConfigEntry
) -> bool:
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)
hass.data[DOMAIN].pop(config_entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
return unload_ok
class ClimaCellDataUpdateCoordinator(DataUpdateCoordinator):
"""Define an object to hold ClimaCell data."""
def __init__(
self,
hass: HomeAssistantType,
config_entry: ConfigEntry,
api: ClimaCell,
update_interval: timedelta,
) -> None:
"""Initialize."""
self._config_entry = config_entry
self._api = api
self.name = config_entry.data[CONF_NAME]
self.data = {CURRENT: {}, FORECASTS: {}}
super().__init__(
hass,
_LOGGER,
name=config_entry.data[CONF_NAME],
update_interval=update_interval,
)
async def _async_update_data(self) -> Dict[str, Any]:
"""Update data via library."""
data = {FORECASTS: {}}
try:
data[CURRENT] = await self._api.realtime(
self._api.available_fields(REALTIME)
)
data[FORECASTS][HOURLY] = await self._api.forecast_hourly(
self._api.available_fields(FORECAST_HOURLY),
None,
timedelta(hours=24),
)
data[FORECASTS][DAILY] = await self._api.forecast_daily(
self._api.available_fields(FORECAST_DAILY), None, timedelta(days=14)
)
data[FORECASTS][NOWCAST] = await self._api.forecast_nowcast(
self._api.available_fields(FORECAST_NOWCAST),
None,
timedelta(
minutes=min(300, self._config_entry.options[CONF_TIMESTEP] * 30)
),
self._config_entry.options[CONF_TIMESTEP],
)
except (
CantConnectException,
InvalidAPIKeyException,
RateLimitedException,
UnknownException,
) as error:
raise UpdateFailed from error
return data
class ClimaCellEntity(CoordinatorEntity):
"""Base ClimaCell Entity."""
def __init__(
self, config_entry: ConfigEntry, coordinator: ClimaCellDataUpdateCoordinator
) -> None:
"""Initialize ClimaCell Entity."""
super().__init__(coordinator)
self._config_entry = config_entry
@staticmethod
def _get_cc_value(
weather_dict: Dict[str, Any], key: str
) -> Optional[Union[int, float, str]]:
"""Return property from weather_dict."""
items = weather_dict.get(key, {})
# Handle cases where value returned is a list.
# Optimistically find the best value to return.
if isinstance(items, list):
if len(items) == 1:
return items[0].get("value")
return next(
(item.get("value") for item in items if "max" in item),
next(
(item.get("value") for item in items if "min" in item),
items[0].get("value", None),
),
)
return items.get("value")
@property
def name(self) -> str:
"""Return the name of the entity."""
return self._config_entry.data[CONF_NAME]
@property
def unique_id(self) -> str:
"""Return the unique id of the entity."""
return self._config_entry.unique_id
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def device_info(self) -> Dict[str, Any]:
"""Return device registry information."""
return {
"identifiers": {(DOMAIN, self._config_entry.data[CONF_API_KEY])},
"name": self.name,
"manufacturer": "ClimaCell",
"entry_type": "service",
}

View file

@ -0,0 +1,146 @@
"""Config flow for ClimaCell integration."""
import logging
from typing import Any, Dict
from pyclimacell import ClimaCell
from pyclimacell.const import REALTIME
from pyclimacell.exceptions import (
CantConnectException,
InvalidAPIKeyException,
RateLimitedException,
)
import voluptuous as vol
from homeassistant import config_entries, core
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from .const import CONF_TIMESTEP, DEFAULT_NAME, DEFAULT_TIMESTEP
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
def _get_config_schema(
hass: core.HomeAssistant, input_dict: Dict[str, Any] = None
) -> vol.Schema:
"""
Return schema defaults for init step based on user input/config dict.
Retain info already provided for future form views by setting them as
defaults in schema.
"""
if input_dict is None:
input_dict = {}
return vol.Schema(
{
vol.Required(
CONF_NAME, default=input_dict.get(CONF_NAME, DEFAULT_NAME)
): str,
vol.Required(CONF_API_KEY, default=input_dict.get(CONF_API_KEY)): str,
vol.Inclusive(
CONF_LATITUDE,
"location",
default=input_dict.get(CONF_LATITUDE, hass.config.latitude),
): cv.latitude,
vol.Inclusive(
CONF_LONGITUDE,
"location",
default=input_dict.get(CONF_LONGITUDE, hass.config.longitude),
): cv.longitude,
},
extra=vol.REMOVE_EXTRA,
)
def _get_unique_id(hass: HomeAssistantType, input_dict: Dict[str, Any]):
"""Return unique ID from config data."""
return (
f"{input_dict[CONF_API_KEY]}"
f"_{input_dict.get(CONF_LATITUDE, hass.config.latitude)}"
f"_{input_dict.get(CONF_LONGITUDE, hass.config.longitude)}"
)
class ClimaCellOptionsConfigFlow(config_entries.OptionsFlow):
"""Handle ClimaCell options."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize ClimaCell options flow."""
self._config_entry = config_entry
async def async_step_init(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Manage the ClimaCell options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
options_schema = {
vol.Required(
CONF_TIMESTEP,
default=self._config_entry.options.get(CONF_TIMESTEP, DEFAULT_TIMESTEP),
): vol.All(vol.Coerce(int), vol.Range(min=1, max=60)),
}
return self.async_show_form(
step_id="init", data_schema=vol.Schema(options_schema)
)
class ClimaCellConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for ClimaCell Weather API."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> ClimaCellOptionsConfigFlow:
"""Get the options flow for this handler."""
return ClimaCellOptionsConfigFlow(config_entry)
async def async_step_user(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Handle the initial step."""
assert self.hass
errors = {}
if user_input is not None:
await self.async_set_unique_id(
unique_id=_get_unique_id(self.hass, user_input)
)
self._abort_if_unique_id_configured()
try:
await ClimaCell(
user_input[CONF_API_KEY],
str(user_input.get(CONF_LATITUDE, self.hass.config.latitude)),
str(user_input.get(CONF_LONGITUDE, self.hass.config.longitude)),
session=async_get_clientsession(self.hass),
).realtime(ClimaCell.first_field(REALTIME))
return self.async_create_entry(
title=user_input[CONF_NAME], data=user_input
)
except CantConnectException:
errors["base"] = "cannot_connect"
except InvalidAPIKeyException:
errors[CONF_API_KEY] = "invalid_api_key"
except RateLimitedException:
errors[CONF_API_KEY] = "rate_limited"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return self.async_show_form(
step_id="user",
data_schema=_get_config_schema(self.hass, user_input),
errors=errors,
)

View file

@ -0,0 +1,79 @@
"""Constants for the ClimaCell integration."""
from homeassistant.components.weather import (
ATTR_CONDITION_CLEAR_NIGHT,
ATTR_CONDITION_CLOUDY,
ATTR_CONDITION_FOG,
ATTR_CONDITION_HAIL,
ATTR_CONDITION_LIGHTNING,
ATTR_CONDITION_PARTLYCLOUDY,
ATTR_CONDITION_POURING,
ATTR_CONDITION_RAINY,
ATTR_CONDITION_SNOWY,
ATTR_CONDITION_SNOWY_RAINY,
ATTR_CONDITION_SUNNY,
)
CONF_TIMESTEP = "timestep"
DAILY = "daily"
HOURLY = "hourly"
NOWCAST = "nowcast"
FORECAST_TYPES = [DAILY, HOURLY, NOWCAST]
CURRENT = "current"
FORECASTS = "forecasts"
DEFAULT_NAME = "ClimaCell"
DEFAULT_TIMESTEP = 15
DEFAULT_FORECAST_TYPE = DAILY
DOMAIN = "climacell"
ATTRIBUTION = "Powered by ClimaCell"
MAX_REQUESTS_PER_DAY = 1000
CONDITIONS = {
"freezing_rain_heavy": ATTR_CONDITION_SNOWY_RAINY,
"freezing_rain": ATTR_CONDITION_SNOWY_RAINY,
"freezing_rain_light": ATTR_CONDITION_SNOWY_RAINY,
"freezing_drizzle": ATTR_CONDITION_SNOWY_RAINY,
"ice_pellets_heavy": ATTR_CONDITION_HAIL,
"ice_pellets": ATTR_CONDITION_HAIL,
"ice_pellets_light": ATTR_CONDITION_HAIL,
"snow_heavy": ATTR_CONDITION_SNOWY,
"snow": ATTR_CONDITION_SNOWY,
"snow_light": ATTR_CONDITION_SNOWY,
"flurries": ATTR_CONDITION_SNOWY,
"tstorm": ATTR_CONDITION_LIGHTNING,
"rain_heavy": ATTR_CONDITION_POURING,
"rain": ATTR_CONDITION_RAINY,
"rain_light": ATTR_CONDITION_RAINY,
"drizzle": ATTR_CONDITION_RAINY,
"fog_light": ATTR_CONDITION_FOG,
"fog": ATTR_CONDITION_FOG,
"cloudy": ATTR_CONDITION_CLOUDY,
"mostly_cloudy": ATTR_CONDITION_CLOUDY,
"partly_cloudy": ATTR_CONDITION_PARTLYCLOUDY,
}
CLEAR_CONDITIONS = {"night": ATTR_CONDITION_CLEAR_NIGHT, "day": ATTR_CONDITION_SUNNY}
CC_ATTR_TIMESTAMP = "observation_time"
CC_ATTR_TEMPERATURE = "temp"
CC_ATTR_TEMPERATURE_HIGH = "max"
CC_ATTR_TEMPERATURE_LOW = "min"
CC_ATTR_PRESSURE = "baro_pressure"
CC_ATTR_HUMIDITY = "humidity"
CC_ATTR_WIND_SPEED = "wind_speed"
CC_ATTR_WIND_DIRECTION = "wind_direction"
CC_ATTR_OZONE = "o3"
CC_ATTR_CONDITION = "weather_code"
CC_ATTR_VISIBILITY = "visibility"
CC_ATTR_PRECIPITATION = "precipitation"
CC_ATTR_PRECIPITATION_DAILY = "precipitation_accumulation"
CC_ATTR_PRECIPITATION_PROBABILITY = "precipitation_probability"
CC_ATTR_PM_2_5 = "pm25"
CC_ATTR_PM_10 = "pm10"
CC_ATTR_CARBON_MONOXIDE = "co"
CC_ATTR_SULPHUR_DIOXIDE = "so2"
CC_ATTR_NITROGEN_DIOXIDE = "no2"

View file

@ -0,0 +1,8 @@
{
"domain": "climacell",
"name": "ClimaCell",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/climacell",
"requirements": ["pyclimacell==0.14.0"],
"codeowners": ["@raman325"]
}

View file

@ -0,0 +1,34 @@
{
"title": "ClimaCell",
"config": {
"step": {
"user": {
"description": "If [%key:component::climacell::config::step::user::data::latitude%] and [%key:component::climacell::config::step::user::data::longitude%] are not provided, the default values in the Home Assistant configuration will be used. An entity will be created for each forecast type but only the ones you select will be enabled by default.",
"data": {
"name": "Name",
"api_key": "[%key:common::config_flow::data::api_key%]",
"latitude": "Latitude",
"longitude": "Longitude"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"rate_limited": "Currently rate limited, please try again later."
}
},
"options": {
"step": {
"init": {
"title": "Update [%key:component::climacell::title%] Options",
"description": "If you choose to enable the `nowcast` forecast entity, you can configure the number of minutes between each forecast. The number of forecasts provided depends on the number of minutes chosen between forecasts.",
"data": {
"timestep": "Min. Between NowCast Forecasts",
"forecast_types": "Forecast Type(s)"
}
}
}
}
}

View file

@ -0,0 +1,305 @@
"""Weather component that handles meteorological data for your location."""
import logging
from typing import Any, Callable, Dict, List, Optional
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_PRECIPITATION_PROBABILITY,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
WeatherEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
LENGTH_FEET,
LENGTH_KILOMETERS,
LENGTH_METERS,
LENGTH_MILES,
PRESSURE_HPA,
PRESSURE_INHG,
TEMP_CELSIUS,
TEMP_FAHRENHEIT,
)
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.sun import is_up
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util
from homeassistant.util.distance import convert as distance_convert
from homeassistant.util.pressure import convert as pressure_convert
from homeassistant.util.temperature import convert as temp_convert
from . import ClimaCellDataUpdateCoordinator, ClimaCellEntity
from .const import (
CC_ATTR_CONDITION,
CC_ATTR_HUMIDITY,
CC_ATTR_OZONE,
CC_ATTR_PRECIPITATION,
CC_ATTR_PRECIPITATION_DAILY,
CC_ATTR_PRECIPITATION_PROBABILITY,
CC_ATTR_PRESSURE,
CC_ATTR_TEMPERATURE,
CC_ATTR_TEMPERATURE_HIGH,
CC_ATTR_TEMPERATURE_LOW,
CC_ATTR_TIMESTAMP,
CC_ATTR_VISIBILITY,
CC_ATTR_WIND_DIRECTION,
CC_ATTR_WIND_SPEED,
CLEAR_CONDITIONS,
CONDITIONS,
CONF_TIMESTEP,
CURRENT,
DAILY,
DEFAULT_FORECAST_TYPE,
DOMAIN,
FORECASTS,
HOURLY,
NOWCAST,
)
# mypy: allow-untyped-defs, no-check-untyped-defs
_LOGGER = logging.getLogger(__name__)
def _translate_condition(
condition: Optional[str], sun_is_up: bool = True
) -> Optional[str]:
"""Translate ClimaCell condition into an HA condition."""
if not condition:
return None
if "clear" in condition.lower():
if sun_is_up:
return CLEAR_CONDITIONS["day"]
return CLEAR_CONDITIONS["night"]
return CONDITIONS[condition]
def _forecast_dict(
hass: HomeAssistantType,
time: str,
use_datetime: bool,
condition: str,
precipitation: Optional[float],
precipitation_probability: Optional[float],
temp: Optional[float],
temp_low: Optional[float],
wind_direction: Optional[float],
wind_speed: Optional[float],
) -> Dict[str, Any]:
"""Return formatted Forecast dict from ClimaCell forecast data."""
if use_datetime:
translated_condition = _translate_condition(
condition,
is_up(hass, dt_util.as_utc(dt_util.parse_datetime(time))),
)
else:
translated_condition = _translate_condition(condition, True)
if hass.config.units.is_metric:
if precipitation:
precipitation = (
distance_convert(precipitation / 12, LENGTH_FEET, LENGTH_METERS) * 1000
)
if temp:
temp = temp_convert(temp, TEMP_FAHRENHEIT, TEMP_CELSIUS)
if temp_low:
temp_low = temp_convert(temp_low, TEMP_FAHRENHEIT, TEMP_CELSIUS)
if wind_speed:
wind_speed = distance_convert(wind_speed, LENGTH_MILES, LENGTH_KILOMETERS)
data = {
ATTR_FORECAST_TIME: time,
ATTR_FORECAST_CONDITION: translated_condition,
ATTR_FORECAST_PRECIPITATION: precipitation,
ATTR_FORECAST_PRECIPITATION_PROBABILITY: precipitation_probability,
ATTR_FORECAST_TEMP: temp,
ATTR_FORECAST_TEMP_LOW: temp_low,
ATTR_FORECAST_WIND_BEARING: wind_direction,
ATTR_FORECAST_WIND_SPEED: wind_speed,
}
return {k: v for k, v in data.items() if v is not None}
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Entity], bool], None],
) -> None:
"""Set up a config entry."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]
entities = [
ClimaCellWeatherEntity(config_entry, coordinator, forecast_type)
for forecast_type in [DAILY, HOURLY, NOWCAST]
]
async_add_entities(entities)
class ClimaCellWeatherEntity(ClimaCellEntity, WeatherEntity):
"""Entity that talks to ClimaCell API to retrieve weather data."""
def __init__(
self,
config_entry: ConfigEntry,
coordinator: ClimaCellDataUpdateCoordinator,
forecast_type: str,
) -> None:
"""Initialize ClimaCell weather entity."""
super().__init__(config_entry, coordinator)
self.forecast_type = forecast_type
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
if self.forecast_type == DEFAULT_FORECAST_TYPE:
return True
return False
@property
def name(self) -> str:
"""Return the name of the entity."""
return f"{super().name} - {self.forecast_type.title()}"
@property
def unique_id(self) -> str:
"""Return the unique id of the entity."""
return f"{super().unique_id}_{self.forecast_type}"
@property
def temperature(self):
"""Return the platform temperature."""
return self._get_cc_value(self.coordinator.data[CURRENT], CC_ATTR_TEMPERATURE)
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return TEMP_FAHRENHEIT
@property
def pressure(self):
"""Return the pressure."""
pressure = self._get_cc_value(self.coordinator.data[CURRENT], CC_ATTR_PRESSURE)
if self.hass.config.units.is_metric and pressure:
return pressure_convert(pressure, PRESSURE_INHG, PRESSURE_HPA)
return pressure
@property
def humidity(self):
"""Return the humidity."""
return self._get_cc_value(self.coordinator.data[CURRENT], CC_ATTR_HUMIDITY)
@property
def wind_speed(self):
"""Return the wind speed."""
wind_speed = self._get_cc_value(
self.coordinator.data[CURRENT], CC_ATTR_WIND_SPEED
)
if self.hass.config.units.is_metric and wind_speed:
return distance_convert(wind_speed, LENGTH_MILES, LENGTH_KILOMETERS)
return wind_speed
@property
def wind_bearing(self):
"""Return the wind bearing."""
return self._get_cc_value(
self.coordinator.data[CURRENT], CC_ATTR_WIND_DIRECTION
)
@property
def ozone(self):
"""Return the O3 (ozone) level."""
return self._get_cc_value(self.coordinator.data[CURRENT], CC_ATTR_OZONE)
@property
def condition(self):
"""Return the condition."""
return _translate_condition(
self._get_cc_value(self.coordinator.data[CURRENT], CC_ATTR_CONDITION),
is_up(self.hass),
)
@property
def visibility(self):
"""Return the visibility."""
visibility = self._get_cc_value(
self.coordinator.data[CURRENT], CC_ATTR_VISIBILITY
)
if self.hass.config.units.is_metric and visibility:
return distance_convert(visibility, LENGTH_MILES, LENGTH_KILOMETERS)
return visibility
@property
def forecast(self):
"""Return the forecast."""
# Check if forecasts are available
if not self.coordinator.data[FORECASTS].get(self.forecast_type):
return None
forecasts = []
# Set default values (in cases where keys don't exist), None will be
# returned. Override properties per forecast type as needed
for forecast in self.coordinator.data[FORECASTS][self.forecast_type]:
timestamp = self._get_cc_value(forecast, CC_ATTR_TIMESTAMP)
use_datetime = True
condition = self._get_cc_value(forecast, CC_ATTR_CONDITION)
precipitation = self._get_cc_value(forecast, CC_ATTR_PRECIPITATION)
precipitation_probability = self._get_cc_value(
forecast, CC_ATTR_PRECIPITATION_PROBABILITY
)
temp = self._get_cc_value(forecast, CC_ATTR_TEMPERATURE)
temp_low = None
wind_direction = self._get_cc_value(forecast, CC_ATTR_WIND_DIRECTION)
wind_speed = self._get_cc_value(forecast, CC_ATTR_WIND_SPEED)
if self.forecast_type == DAILY:
use_datetime = False
precipitation = self._get_cc_value(
forecast, CC_ATTR_PRECIPITATION_DAILY
)
temp = next(
(
self._get_cc_value(item, CC_ATTR_TEMPERATURE_HIGH)
for item in forecast[CC_ATTR_TEMPERATURE]
if "max" in item
),
temp,
)
temp_low = next(
(
self._get_cc_value(item, CC_ATTR_TEMPERATURE_LOW)
for item in forecast[CC_ATTR_TEMPERATURE]
if "max" in item
),
temp_low,
)
elif self.forecast_type == NOWCAST:
# Precipitation is forecasted in CONF_TIMESTEP increments but in a
# per hour rate, so value needs to be converted to an amount.
if precipitation:
precipitation = (
precipitation / 60 * self._config_entry.options[CONF_TIMESTEP]
)
forecasts.append(
_forecast_dict(
self.hass,
timestamp,
use_datetime,
condition,
precipitation,
precipitation_probability,
temp,
temp_low,
wind_direction,
wind_speed,
)
)
return forecasts

View file

@ -40,6 +40,7 @@ FLOWS = [
"canary",
"cast",
"cert_expiry",
"climacell",
"cloudflare",
"control4",
"coolmaster",

View file

@ -1313,6 +1313,9 @@ pychromecast==8.1.0
# homeassistant.components.pocketcasts
pycketcasts==1.0.0
# homeassistant.components.climacell
pyclimacell==0.14.0
# homeassistant.components.cmus
pycmus==0.1.1

View file

@ -693,6 +693,9 @@ pycfdns==1.2.1
# homeassistant.components.cast
pychromecast==8.1.0
# homeassistant.components.climacell
pyclimacell==0.14.0
# homeassistant.components.comfoconnect
pycomfoconnect==0.4

View file

@ -0,0 +1 @@
"""Tests for the ClimaCell Weather API integration."""

View file

@ -0,0 +1,42 @@
"""Configure py.test."""
from unittest.mock import patch
import pytest
@pytest.fixture(name="skip_notifications", autouse=True)
def skip_notifications_fixture():
"""Skip notification calls."""
with patch("homeassistant.components.persistent_notification.async_create"), patch(
"homeassistant.components.persistent_notification.async_dismiss"
):
yield
@pytest.fixture(name="climacell_config_flow_connect", autouse=True)
def climacell_config_flow_connect():
"""Mock valid climacell config flow setup."""
with patch(
"homeassistant.components.climacell.config_flow.ClimaCell.realtime",
return_value={},
):
yield
@pytest.fixture(name="climacell_config_entry_update")
def climacell_config_entry_update_fixture():
"""Mock valid climacell config entry setup."""
with patch(
"homeassistant.components.climacell.ClimaCell.realtime",
return_value={},
), patch(
"homeassistant.components.climacell.ClimaCell.forecast_hourly",
return_value=[],
), patch(
"homeassistant.components.climacell.ClimaCell.forecast_daily",
return_value=[],
), patch(
"homeassistant.components.climacell.ClimaCell.forecast_nowcast",
return_value=[],
):
yield

View file

@ -0,0 +1,9 @@
"""Constants for climacell tests."""
from homeassistant.const import CONF_API_KEY
API_KEY = "aa"
MIN_CONFIG = {
CONF_API_KEY: API_KEY,
}

View file

@ -0,0 +1,167 @@
"""Test the ClimaCell config flow."""
import logging
from unittest.mock import patch
from pyclimacell.exceptions import (
CantConnectException,
InvalidAPIKeyException,
RateLimitedException,
UnknownException,
)
from homeassistant import data_entry_flow
from homeassistant.components.climacell.config_flow import (
_get_config_schema,
_get_unique_id,
)
from homeassistant.components.climacell.const import (
CONF_TIMESTEP,
DEFAULT_NAME,
DEFAULT_TIMESTEP,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
from homeassistant.helpers.typing import HomeAssistantType
from .const import API_KEY, MIN_CONFIG
from tests.common import MockConfigEntry
_LOGGER = logging.getLogger(__name__)
async def test_user_flow_minimum_fields(hass: HomeAssistantType) -> None:
"""Test user config flow with minimum fields."""
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"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=_get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == DEFAULT_NAME
assert result["data"][CONF_NAME] == DEFAULT_NAME
assert result["data"][CONF_API_KEY] == API_KEY
assert result["data"][CONF_LATITUDE] == hass.config.latitude
assert result["data"][CONF_LONGITUDE] == hass.config.longitude
async def test_user_flow_same_unique_ids(hass: HomeAssistantType) -> None:
"""Test user config flow with the same unique ID as an existing entry."""
user_input = _get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG)
MockConfigEntry(
domain=DOMAIN,
data=user_input,
source=SOURCE_USER,
unique_id=_get_unique_id(hass, user_input),
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=user_input,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_user_flow_cannot_connect(hass: HomeAssistantType) -> None:
"""Test user config flow when ClimaCell can't connect."""
with patch(
"homeassistant.components.climacell.config_flow.ClimaCell.realtime",
side_effect=CantConnectException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=_get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_user_flow_invalid_api(hass: HomeAssistantType) -> None:
"""Test user config flow when API key is invalid."""
with patch(
"homeassistant.components.climacell.config_flow.ClimaCell.realtime",
side_effect=InvalidAPIKeyException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=_get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {CONF_API_KEY: "invalid_api_key"}
async def test_user_flow_rate_limited(hass: HomeAssistantType) -> None:
"""Test user config flow when API key is rate limited."""
with patch(
"homeassistant.components.climacell.config_flow.ClimaCell.realtime",
side_effect=RateLimitedException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=_get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {CONF_API_KEY: "rate_limited"}
async def test_user_flow_unknown_exception(hass: HomeAssistantType) -> None:
"""Test user config flow when unknown error occurs."""
with patch(
"homeassistant.components.climacell.config_flow.ClimaCell.realtime",
side_effect=UnknownException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=_get_config_schema(hass, MIN_CONFIG)(MIN_CONFIG),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown"}
async def test_options_flow(hass: HomeAssistantType) -> None:
"""Test options config flow for climacell."""
user_config = _get_config_schema(hass)(MIN_CONFIG)
entry = MockConfigEntry(
domain=DOMAIN,
data=user_config,
source=SOURCE_USER,
unique_id=_get_unique_id(hass, user_config),
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.options[CONF_TIMESTEP] == DEFAULT_TIMESTEP
assert CONF_TIMESTEP not in entry.data
result = await hass.config_entries.options.async_init(entry.entry_id, data=None)
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_TIMESTEP: 1}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == ""
assert result["data"][CONF_TIMESTEP] == 1
assert entry.options[CONF_TIMESTEP] == 1

View file

@ -0,0 +1,82 @@
"""Tests for Climacell init."""
from datetime import timedelta
import logging
from unittest.mock import patch
import pytest
from homeassistant.components.climacell.config_flow import (
_get_config_schema,
_get_unique_id,
)
from homeassistant.components.climacell.const import DOMAIN
from homeassistant.components.weather import DOMAIN as WEATHER_DOMAIN
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util
from .const import MIN_CONFIG
from tests.common import MockConfigEntry, async_fire_time_changed
_LOGGER = logging.getLogger(__name__)
async def test_load_and_unload(
hass: HomeAssistantType,
climacell_config_entry_update: pytest.fixture,
) -> None:
"""Test loading and unloading entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_get_config_schema(hass)(MIN_CONFIG),
unique_id=_get_unique_id(hass, _get_config_schema(hass)(MIN_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 len(hass.states.async_entity_ids(WEATHER_DOMAIN)) == 1
assert await hass.config_entries.async_remove(config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(WEATHER_DOMAIN)) == 0
async def test_update_interval(
hass: HomeAssistantType,
climacell_config_entry_update: pytest.fixture,
) -> None:
"""Test that update_interval changes based on number of entries."""
now = dt_util.utcnow()
async_fire_time_changed(hass, now)
config = _get_config_schema(hass)(MIN_CONFIG)
for i in range(1, 3):
config_entry = MockConfigEntry(
domain=DOMAIN, data=config, unique_id=_get_unique_id(hass, config) + str(i)
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with patch("homeassistant.components.climacell.ClimaCell.realtime") as mock_api:
# First entry refresh will happen in 7 minutes due to original update interval.
# Next refresh for this entry will happen at 20 minutes due to the update interval
# change.
mock_api.return_value = {}
async_fire_time_changed(hass, now + timedelta(minutes=7))
await hass.async_block_till_done()
assert mock_api.call_count == 1
# Second entry refresh will happen in 13 minutes due to the update interval set
# when it was set up. Next refresh for this entry will happen at 26 minutes due to the
# update interval change.
mock_api.reset_mock()
async_fire_time_changed(hass, now + timedelta(minutes=13))
await hass.async_block_till_done()
assert not mock_api.call_count == 1
# 19 minutes should be after the first update for each config entry and before the
# second update for the first config entry
mock_api.reset_mock()
async_fire_time_changed(hass, now + timedelta(minutes=19))
await hass.async_block_till_done()
assert not mock_api.call_count == 0