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

Airtouch4 integration (#43513) 

* airtouch 4 climate control integration

* enhance tests for airtouch. Fix linting issues

* Fix tests

* rework tests

* fix latest qa issues

* Clean up

* add already_configured message

* Use common string

* further qa fixes

* simplify airtouch4 domain storage

Co-authored-by: Franck Nijhof <git@frenck.dev>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/54769/head
LonePurpleWolf 2021-08-18 01:29:20 +10:00 committed by GitHub
parent ea8061469c
commit 35f563e23e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 653 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -36,6 +36,9 @@ omit =
homeassistant/components/agent_dvr/helpers.py
homeassistant/components/airnow/__init__.py
homeassistant/components/airnow/sensor.py
homeassistant/components/airtouch4/__init__.py
homeassistant/components/airtouch4/climate.py
homeassistant/components/airtouch4/const.py
homeassistant/components/airvisual/__init__.py
homeassistant/components/airvisual/sensor.py
homeassistant/components/aladdin_connect/*

1
CODEOWNERS
View File

@ -29,6 +29,7 @@ homeassistant/components/aemet/* @noltari
homeassistant/components/agent_dvr/* @ispysoftware
homeassistant/components/airly/* @bieniu
homeassistant/components/airnow/* @asymworks
homeassistant/components/airtouch4/* @LonePurpleWolf
homeassistant/components/airvisual/* @bachya
homeassistant/components/alarmdecoder/* @ajschmidt8
homeassistant/components/alexa/* @home-assistant/cloud @ochlocracy

81
homeassistant/components/airtouch4/__init__.py Normal file
View File

@ -0,0 +1,81 @@
"""The AirTouch4 integration."""
import logging
from airtouch4pyapi import AirTouch
from airtouch4pyapi.airtouch import AirTouchStatus
from homeassistant.components.climate import SCAN_INTERVAL
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["climate"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up AirTouch4 from a config entry."""
hass.data.setdefault(DOMAIN, {})
host = entry.data[CONF_HOST]
airtouch = AirTouch(host)
await airtouch.UpdateInfo()
info = airtouch.GetAcs()
if not info:
raise ConfigEntryNotReady
coordinator = AirtouchDataUpdateCoordinator(hass, airtouch)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
class AirtouchDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching Airtouch data."""
def __init__(self, hass, airtouch):
"""Initialize global Airtouch data updater."""
self.airtouch = airtouch
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
)
async def _async_update_data(self):
"""Fetch data from Airtouch."""
await self.airtouch.UpdateInfo()
if self.airtouch.Status != AirTouchStatus.OK:
raise UpdateFailed("Airtouch connection issue")
return {
"acs": [
{"ac_number": ac.AcNumber, "is_on": ac.IsOn}
for ac in self.airtouch.GetAcs()
],
"groups": [
{
"group_number": group.GroupNumber,
"group_name": group.GroupName,
"is_on": group.IsOn,
}
for group in self.airtouch.GetGroups()
],
}

335
homeassistant/components/airtouch4/climate.py Normal file
View File

@ -0,0 +1,335 @@
"""AirTouch 4 component to control of AirTouch 4 Climate Devices."""
import logging
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
FAN_AUTO,
FAN_DIFFUSE,
FAN_FOCUS,
FAN_HIGH,
FAN_LOW,
FAN_MEDIUM,
HVAC_MODE_AUTO,
HVAC_MODE_COOL,
HVAC_MODE_DRY,
HVAC_MODE_FAN_ONLY,
HVAC_MODE_HEAT,
HVAC_MODE_OFF,
SUPPORT_FAN_MODE,
SUPPORT_TARGET_TEMPERATURE,
)
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS
from homeassistant.core import callback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
SUPPORT_FLAGS = SUPPORT_TARGET_TEMPERATURE | SUPPORT_FAN_MODE
AT_TO_HA_STATE = {
"Heat": HVAC_MODE_HEAT,
"Cool": HVAC_MODE_COOL,
"AutoHeat": HVAC_MODE_AUTO, # airtouch reports either autoheat or autocool
"AutoCool": HVAC_MODE_AUTO,
"Auto": HVAC_MODE_AUTO,
"Dry": HVAC_MODE_DRY,
"Fan": HVAC_MODE_FAN_ONLY,
}
HA_STATE_TO_AT = {
HVAC_MODE_HEAT: "Heat",
HVAC_MODE_COOL: "Cool",
HVAC_MODE_AUTO: "Auto",
HVAC_MODE_DRY: "Dry",
HVAC_MODE_FAN_ONLY: "Fan",
HVAC_MODE_OFF: "Off",
}
AT_TO_HA_FAN_SPEED = {
"Quiet": FAN_DIFFUSE,
"Low": FAN_LOW,
"Medium": FAN_MEDIUM,
"High": FAN_HIGH,
"Powerful": FAN_FOCUS,
"Auto": FAN_AUTO,
"Turbo": "turbo",
}
AT_GROUP_MODES = [HVAC_MODE_OFF, HVAC_MODE_FAN_ONLY]
HA_FAN_SPEED_TO_AT = {value: key for key, value in AT_TO_HA_FAN_SPEED.items()}
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Airtouch 4."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]
info = coordinator.data
entities = [
AirtouchGroup(coordinator, group["group_number"], info)
for group in info["groups"]
] + [AirtouchAC(coordinator, ac["ac_number"], info) for ac in info["acs"]]
_LOGGER.debug(" Found entities %s", entities)
async_add_entities(entities)
class AirtouchAC(CoordinatorEntity, ClimateEntity):
"""Representation of an AirTouch 4 ac."""
_attr_supported_features = SUPPORT_TARGET_TEMPERATURE | SUPPORT_FAN_MODE
_attr_temperature_unit = TEMP_CELSIUS
def __init__(self, coordinator, ac_number, info):
"""Initialize the climate device."""
super().__init__(coordinator)
self._ac_number = ac_number
self._airtouch = coordinator.airtouch
self._info = info
self._unit = self._airtouch.GetAcs()[self._ac_number]
@callback
def _handle_coordinator_update(self):
self._unit = self._airtouch.GetAcs()[self._ac_number]
return super()._handle_coordinator_update()
@property
def device_info(self):
"""Return device info for this device."""
return {
"identifiers": {(DOMAIN, self.unique_id)},
"name": self.name,
"manufacturer": "Airtouch",
"model": "Airtouch 4",
}
@property
def unique_id(self):
"""Return unique ID for this device."""
return f"ac_{self._ac_number}"
@property
def current_temperature(self):
"""Return the current temperature."""
return self._unit.Temperature
@property
def name(self):
"""Return the name of the climate device."""
return f"AC {self._ac_number}"
@property
def fan_mode(self):
"""Return fan mode of the AC this group belongs to."""
return AT_TO_HA_FAN_SPEED[self._airtouch.acs[self._ac_number].AcFanSpeed]
@property
def fan_modes(self):
"""Return the list of available fan modes."""
airtouch_fan_speeds = self._airtouch.GetSupportedFanSpeedsForAc(self._ac_number)
return [AT_TO_HA_FAN_SPEED[speed] for speed in airtouch_fan_speeds]
@property
def hvac_mode(self):
"""Return hvac target hvac state."""
is_off = self._unit.PowerState == "Off"
if is_off:
return HVAC_MODE_OFF
return AT_TO_HA_STATE[self._airtouch.acs[self._ac_number].AcMode]
@property
def hvac_modes(self):
"""Return the list of available operation modes."""
airtouch_modes = self._airtouch.GetSupportedCoolingModesForAc(self._ac_number)
modes = [AT_TO_HA_STATE[mode] for mode in airtouch_modes]
modes.append(HVAC_MODE_OFF)
return modes
async def async_set_hvac_mode(self, hvac_mode):
"""Set new operation mode."""
if hvac_mode not in HA_STATE_TO_AT:
raise ValueError(f"Unsupported HVAC mode: {hvac_mode}")
if hvac_mode == HVAC_MODE_OFF:
return await self.async_turn_off()
await self._airtouch.SetCoolingModeForAc(
self._ac_number, HA_STATE_TO_AT[hvac_mode]
)
# in case it isn't already, unless the HVAC mode was off, then the ac should be on
await self.async_turn_on()
self._unit = self._airtouch.GetAcs()[self._ac_number]
_LOGGER.debug("Setting operation mode of %s to %s", self._ac_number, hvac_mode)
self.async_write_ha_state()
async def async_set_fan_mode(self, fan_mode):
"""Set new fan mode."""
if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan mode: {fan_mode}")
_LOGGER.debug("Setting fan mode of %s to %s", self._ac_number, fan_mode)
await self._airtouch.SetFanSpeedForAc(
self._ac_number, HA_FAN_SPEED_TO_AT[fan_mode]
)
self._unit = self._airtouch.GetAcs()[self._ac_number]
self.async_write_ha_state()
async def async_turn_on(self):
"""Turn on."""
_LOGGER.debug("Turning %s on", self.unique_id)
# in case ac is not on. Airtouch turns itself off if no groups are turned on
# (even if groups turned back on)
await self._airtouch.TurnAcOn(self._ac_number)
async def async_turn_off(self):
"""Turn off."""
_LOGGER.debug("Turning %s off", self.unique_id)
await self._airtouch.TurnAcOff(self._ac_number)
self.async_write_ha_state()
class AirtouchGroup(CoordinatorEntity, ClimateEntity):
"""Representation of an AirTouch 4 group."""
_attr_supported_features = SUPPORT_TARGET_TEMPERATURE
_attr_temperature_unit = TEMP_CELSIUS
_attr_hvac_modes = AT_GROUP_MODES
def __init__(self, coordinator, group_number, info):
"""Initialize the climate device."""
super().__init__(coordinator)
self._group_number = group_number
self._airtouch = coordinator.airtouch
self._info = info
self._unit = self._airtouch.GetGroupByGroupNumber(self._group_number)
@callback
def _handle_coordinator_update(self):
self._unit = self._airtouch.GetGroupByGroupNumber(self._group_number)
return super()._handle_coordinator_update()
@property
def device_info(self):
"""Return device info for this device."""
return {
"identifiers": {(DOMAIN, self.unique_id)},
"name": self.name,
"manufacturer": "Airtouch",
"model": "Airtouch 4",
}
@property
def unique_id(self):
"""Return unique ID for this device."""
return self._group_number
@property
def min_temp(self):
"""Return Minimum Temperature for AC of this group."""
return self._airtouch.acs[self._unit.BelongsToAc].MinSetpoint
@property
def max_temp(self):
"""Return Max Temperature for AC of this group."""
return self._airtouch.acs[self._unit.BelongsToAc].MaxSetpoint
@property
def name(self):
"""Return the name of the climate device."""
return self._unit.GroupName
@property
def current_temperature(self):
"""Return the current temperature."""
return self._unit.Temperature
@property
def target_temperature(self):
"""Return the temperature we are trying to reach."""
return self._unit.TargetSetpoint
@property
def hvac_mode(self):
"""Return hvac target hvac state."""
# there are other power states that aren't 'on' but still count as on (eg. 'Turbo')
is_off = self._unit.PowerState == "Off"
if is_off:
return HVAC_MODE_OFF
return HVAC_MODE_FAN_ONLY
async def async_set_hvac_mode(self, hvac_mode):
"""Set new operation mode."""
if hvac_mode not in HA_STATE_TO_AT:
raise ValueError(f"Unsupported HVAC mode: {hvac_mode}")
if hvac_mode == HVAC_MODE_OFF:
return await self.async_turn_off()
if self.hvac_mode == HVAC_MODE_OFF:
await self.async_turn_on()
self._unit = self._airtouch.GetGroups()[self._group_number]
_LOGGER.debug(
"Setting operation mode of %s to %s", self._group_number, hvac_mode
)
self.async_write_ha_state()
@property
def fan_mode(self):
"""Return fan mode of the AC this group belongs to."""
return AT_TO_HA_FAN_SPEED[self._airtouch.acs[self._unit.BelongsToAc].AcFanSpeed]
@property
def fan_modes(self):
"""Return the list of available fan modes."""
airtouch_fan_speeds = self._airtouch.GetSupportedFanSpeedsByGroup(
self._group_number
)
return [AT_TO_HA_FAN_SPEED[speed] for speed in airtouch_fan_speeds]
async def async_set_temperature(self, **kwargs):
"""Set new target temperatures."""
temp = kwargs.get(ATTR_TEMPERATURE)
_LOGGER.debug("Setting temp of %s to %s", self._group_number, str(temp))
self._unit = await self._airtouch.SetGroupToTemperature(
self._group_number, int(temp)
)
self.async_write_ha_state()
async def async_set_fan_mode(self, fan_mode):
"""Set new fan mode."""
if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan mode: {fan_mode}")
_LOGGER.debug("Setting fan mode of %s to %s", self._group_number, fan_mode)
self._unit = await self._airtouch.SetFanSpeedByGroup(
self._group_number, HA_FAN_SPEED_TO_AT[fan_mode]
)
self.async_write_ha_state()
async def async_turn_on(self):
"""Turn on."""
_LOGGER.debug("Turning %s on", self.unique_id)
await self._airtouch.TurnGroupOn(self._group_number)
# in case ac is not on. Airtouch turns itself off if no groups are turned on
# (even if groups turned back on)
await self._airtouch.TurnAcOn(
self._airtouch.GetGroupByGroupNumber(self._group_number).BelongsToAc
)
# this might cause the ac object to be wrong, so force the shared data
# store to update
await self.coordinator.async_request_refresh()
self.async_write_ha_state()
async def async_turn_off(self):
"""Turn off."""
_LOGGER.debug("Turning %s off", self.unique_id)
await self._airtouch.TurnGroupOff(self._group_number)
# this will cause the ac object to be wrong
# (ac turns off automatically if no groups are running)
# so force the shared data store to update
await self.coordinator.async_request_refresh()
self.async_write_ha_state()

50
homeassistant/components/airtouch4/config_flow.py Normal file
View File

@ -0,0 +1,50 @@
"""Config flow for AirTouch4."""
from airtouch4pyapi import AirTouch, AirTouchStatus
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
from .const import DOMAIN
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
class AirtouchConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle an Airtouch config flow."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
if user_input is None:
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
errors = {}
host = user_input[CONF_HOST]
self._async_abort_entries_match({CONF_HOST: host})
airtouch = AirTouch(host)
await airtouch.UpdateInfo()
airtouch_status = airtouch.Status
airtouch_has_groups = bool(
airtouch.Status == AirTouchStatus.OK and airtouch.GetGroups()
)
if airtouch_status != AirTouchStatus.OK:
errors["base"] = "cannot_connect"
elif not airtouch_has_groups:
errors["base"] = "no_units"
if errors:
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors
)
return self.async_create_entry(
title=user_input[CONF_HOST],
data={
CONF_HOST: user_input[CONF_HOST],
},
)

3
homeassistant/components/airtouch4/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Constants for the AirTouch4 integration."""
DOMAIN = "airtouch4"

13
homeassistant/components/airtouch4/manifest.json Normal file
View File

@ -0,0 +1,13 @@
{
"domain": "airtouch4",
"name": "AirTouch 4",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/airtouch4",
"requirements": [
"airtouch4pyapi==1.0.5"
],
"codeowners": [
"@LonePurpleWolf"
],
"iot_class": "local_polling"
}

19
homeassistant/components/airtouch4/strings.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"no_units": "Could not find any AirTouch 4 Groups."
},
"step": {
"user": {
"title": "Setup your AirTouch 4 connection details.",
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
}
}
}
}

17
homeassistant/components/airtouch4/translations/en.json Normal file
View File

@ -0,0 +1,17 @@
{
"config": {
"error": {
"cannot_connect": "Failed to connect",
"no_units": "Could not find any AirTouch 4 Groups."
},
"step": {
"user": {
"title": "Setup your AirTouch 4.",
"data": {
"host": "Host"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -16,6 +16,7 @@ FLOWS = [
"agent_dvr",
"airly",
"airnow",
"airtouch4",
"airvisual",
"alarmdecoder",
"almond",

3
requirements_all.txt
View File

@ -257,6 +257,9 @@ aioymaps==1.1.0
# homeassistant.components.airly
airly==1.1.0
# homeassistant.components.airtouch4
airtouch4pyapi==1.0.5
# homeassistant.components.aladdin_connect
aladdin_connect==0.3

3
requirements_test_all.txt
View File

@ -178,6 +178,9 @@ aioymaps==1.1.0
# homeassistant.components.airly
airly==1.1.0
# homeassistant.components.airtouch4
airtouch4pyapi==1.0.5
# homeassistant.components.ambee
ambee==0.3.0

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

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

123
tests/components/airtouch4/test_config_flow.py Normal file
View File

@ -0,0 +1,123 @@
"""Test the AirTouch 4 config flow."""
from unittest.mock import AsyncMock, Mock, patch
from airtouch4pyapi.airtouch import AirTouch, AirTouchAc, AirTouchGroup, AirTouchStatus
from homeassistant import config_entries
from homeassistant.components.airtouch4.const import DOMAIN
async def test_form(hass):
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] is None
mock_ac = AirTouchAc()
mock_groups = AirTouchGroup()
mock_airtouch = AirTouch("")
mock_airtouch.UpdateInfo = AsyncMock()
mock_airtouch.Status = AirTouchStatus.OK
mock_airtouch.GetAcs = Mock(return_value=[mock_ac])
mock_airtouch.GetGroups = Mock(return_value=[mock_groups])
with patch(
"homeassistant.components.airtouch4.config_flow.AirTouch",
return_value=mock_airtouch,
), patch(
"homeassistant.components.airtouch4.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"host": "0.0.0.1"}
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == "0.0.0.1"
assert result2["data"] == {
"host": "0.0.0.1",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_timeout(hass):
"""Test we handle a connection timeout."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_airtouch = AirTouch("")
mock_airtouch.UpdateInfo = AsyncMock()
mock_airtouch.status = AirTouchStatus.CONNECTION_INTERRUPTED
with patch(
"homeassistant.components.airtouch4.config_flow.AirTouch",
return_value=mock_airtouch,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"host": "0.0.0.1"}
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_library_error_message(hass):
"""Test we handle an unknown error message from the library."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_airtouch = AirTouch("")
mock_airtouch.UpdateInfo = AsyncMock()
mock_airtouch.status = AirTouchStatus.ERROR
with patch(
"homeassistant.components.airtouch4.config_flow.AirTouch",
return_value=mock_airtouch,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"host": "0.0.0.1"}
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_connection_refused(hass):
"""Test we handle a connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_airtouch = AirTouch("")
mock_airtouch.UpdateInfo = AsyncMock()
mock_airtouch.status = AirTouchStatus.NOT_CONNECTED
with patch(
"homeassistant.components.airtouch4.config_flow.AirTouch",
return_value=mock_airtouch,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"host": "0.0.0.1"}
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_no_units(hass):
"""Test we handle no units found."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_ac = AirTouchAc()
mock_airtouch = AirTouch("")
mock_airtouch.UpdateInfo = AsyncMock()
mock_airtouch.Status = AirTouchStatus.OK
mock_airtouch.GetAcs = Mock(return_value=[mock_ac])
mock_airtouch.GetGroups = Mock(return_value=[])
with patch(
"homeassistant.components.airtouch4.config_flow.AirTouch",
return_value=mock_airtouch,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"host": "0.0.0.1"}
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "no_units"}