Add Xiaomi miio Alarm Control Panel (#32091)

Co-Authored-By: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
starkillerOG 2020-04-29 01:44:31 +02:00 committed by GitHub
parent 26241980d7
commit e46f1b69ba
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 545 additions and 2 deletions

View file

@ -829,7 +829,17 @@ omit =
homeassistant/components/xfinity/device_tracker.py
homeassistant/components/xiaomi/camera.py
homeassistant/components/xiaomi_aqara/*
homeassistant/components/xiaomi_miio/*
homeassistant/components/xiaomi_miio/__init__.py
homeassistant/components/xiaomi_miio/air_quality.py
homeassistant/components/xiaomi_miio/alarm_control_panel.py
homeassistant/components/xiaomi_miio/device_tracker.py
homeassistant/components/xiaomi_miio/fan.py
homeassistant/components/xiaomi_miio/gateway.py
homeassistant/components/xiaomi_miio/light.py
homeassistant/components/xiaomi_miio/remote.py
homeassistant/components/xiaomi_miio/sensor.py
homeassistant/components/xiaomi_miio/switch.py
homeassistant/components/xiaomi_miio/vacuum.py
homeassistant/components/xiaomi_tv/media_player.py
homeassistant/components/xmpp/notify.py
homeassistant/components/xs1/*

View file

@ -1 +1,69 @@
"""Support for Xiaomi Miio."""
import logging
from homeassistant import config_entries, core
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.helpers import device_registry as dr
from .config_flow import CONF_FLOW_TYPE, CONF_GATEWAY
from .const import DOMAIN
from .gateway import ConnectXiaomiGateway
_LOGGER = logging.getLogger(__name__)
GATEWAY_PLATFORMS = ["alarm_control_panel"]
async def async_setup(hass: core.HomeAssistant, config: dict):
"""Set up the Xiaomi Miio component."""
return True
async def async_setup_entry(
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
):
"""Set up the Xiaomi Miio components from a config entry."""
hass.data[DOMAIN] = {}
if entry.data[CONF_FLOW_TYPE] == CONF_GATEWAY:
if not await async_setup_gateway_entry(hass, entry):
return False
return True
async def async_setup_gateway_entry(
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
):
"""Set up the Xiaomi Gateway component from a config entry."""
host = entry.data[CONF_HOST]
token = entry.data[CONF_TOKEN]
name = entry.title
gateway_id = entry.data["gateway_id"]
# Connect to gateway
gateway = ConnectXiaomiGateway(hass)
if not await gateway.async_connect_gateway(host, token):
return False
gateway_info = gateway.gateway_info
hass.data[DOMAIN][entry.entry_id] = gateway.gateway_device
gateway_model = f"{gateway_info.model}-{gateway_info.hardware_version}"
device_registry = await dr.async_get_registry(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, gateway_info.mac_address)},
identifiers={(DOMAIN, gateway_id)},
manufacturer="Xiaomi",
name=name,
model=gateway_model,
sw_version=gateway_info.firmware_version,
)
for component in GATEWAY_PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True

View file

@ -0,0 +1,150 @@
"""Support for Xiomi Gateway alarm control panels."""
from functools import partial
import logging
from miio import DeviceException
from homeassistant.components.alarm_control_panel import (
SUPPORT_ALARM_ARM_AWAY,
AlarmControlPanelEntity,
)
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMED,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
XIAOMI_STATE_ARMED_VALUE = "on"
XIAOMI_STATE_DISARMED_VALUE = "off"
XIAOMI_STATE_ARMING_VALUE = "oning"
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Xiaomi Gateway Alarm from a config entry."""
entities = []
gateway = hass.data[DOMAIN][config_entry.entry_id]
entity = XiaomiGatewayAlarm(
gateway,
f"{config_entry.title} Alarm",
config_entry.data["model"],
config_entry.data["mac"],
config_entry.data["gateway_id"],
)
entities.append(entity)
async_add_entities(entities)
class XiaomiGatewayAlarm(AlarmControlPanelEntity):
"""Representation of the XiaomiGatewayAlarm."""
def __init__(
self, gateway_device, gateway_name, model, mac_address, gateway_device_id
):
"""Initialize the entity."""
self._gateway = gateway_device
self._name = gateway_name
self._gateway_device_id = gateway_device_id
self._unique_id = f"{model}-{mac_address}"
self._icon = "mdi:shield-home"
self._available = None
self._state = None
@property
def unique_id(self):
"""Return an unique ID."""
return self._unique_id
@property
def device_id(self):
"""Return the device id of the gateway."""
return self._gateway_device_id
@property
def device_info(self):
"""Return the device info of the gateway."""
return {
"identifiers": {(DOMAIN, self._gateway_device_id)},
}
@property
def name(self):
"""Return the name of this entity, if any."""
return self._name
@property
def icon(self):
"""Return the icon to use for device if any."""
return self._icon
@property
def available(self):
"""Return true when state is known."""
return self._available
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def supported_features(self) -> int:
"""Return the list of supported features."""
return SUPPORT_ALARM_ARM_AWAY
async def _try_command(self, mask_error, func, *args, **kwargs):
"""Call a device command handling error messages."""
try:
result = await self.hass.async_add_executor_job(
partial(func, *args, **kwargs)
)
_LOGGER.debug("Response received from miio device: %s", result)
except DeviceException as exc:
_LOGGER.error(mask_error, exc)
async def async_alarm_arm_away(self, code=None):
"""Turn on."""
await self._try_command(
"Turning the alarm on failed: %s", self._gateway.alarm.on
)
async def async_alarm_disarm(self, code=None):
"""Turn off."""
await self._try_command(
"Turning the alarm off failed: %s", self._gateway.alarm.off
)
async def async_update(self):
"""Fetch state from the device."""
try:
state = await self.hass.async_add_executor_job(self._gateway.alarm.status)
except DeviceException as ex:
self._available = False
_LOGGER.error("Got exception while fetching the state: %s", ex)
return
_LOGGER.debug("Got new state: %s", state)
self._available = True
if state == XIAOMI_STATE_ARMED_VALUE:
self._state = STATE_ALARM_ARMED_AWAY
elif state == XIAOMI_STATE_DISARMED_VALUE:
self._state = STATE_ALARM_DISARMED
elif state == XIAOMI_STATE_ARMING_VALUE:
self._state = STATE_ALARM_ARMING
else:
_LOGGER.warning(
"New state (%s) doesn't match expected values: %s/%s/%s",
state,
XIAOMI_STATE_ARMED_VALUE,
XIAOMI_STATE_DISARMED_VALUE,
XIAOMI_STATE_ARMING_VALUE,
)
self._state = None
_LOGGER.debug("State value: %s", self._state)

View file

@ -0,0 +1,82 @@
"""Config flow to configure Xiaomi Miio."""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_TOKEN
# pylint: disable=unused-import
from .const import DOMAIN
from .gateway import ConnectXiaomiGateway
_LOGGER = logging.getLogger(__name__)
CONF_FLOW_TYPE = "config_flow_device"
CONF_GATEWAY = "gateway"
DEFAULT_GATEWAY_NAME = "Xiaomi Gateway"
GATEWAY_CONFIG = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_TOKEN): vol.All(str, vol.Length(min=32, max=32)),
vol.Optional(CONF_NAME, default=DEFAULT_GATEWAY_NAME): str,
}
)
CONFIG_SCHEMA = vol.Schema({vol.Optional(CONF_GATEWAY, default=False): bool})
class XiaomiMiioFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a Xiaomi Miio config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
# Check which device needs to be connected.
if user_input[CONF_GATEWAY]:
return await self.async_step_gateway()
errors["base"] = "no_device_selected"
return self.async_show_form(
step_id="user", data_schema=CONFIG_SCHEMA, errors=errors
)
async def async_step_gateway(self, user_input=None):
"""Handle a flow initialized by the user to configure a gateway."""
errors = {}
if user_input is not None:
host = user_input[CONF_HOST]
token = user_input[CONF_TOKEN]
# Try to connect to a Xiaomi Gateway.
connect_gateway_class = ConnectXiaomiGateway(self.hass)
await connect_gateway_class.async_connect_gateway(host, token)
gateway_info = connect_gateway_class.gateway_info
if gateway_info is not None:
unique_id = f"{gateway_info.model}-{gateway_info.mac_address}-gateway"
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_NAME],
data={
CONF_FLOW_TYPE: CONF_GATEWAY,
CONF_HOST: host,
CONF_TOKEN: token,
"gateway_id": unique_id,
"model": gateway_info.model,
"mac": gateway_info.mac_address,
},
)
errors["base"] = "connect_error"
return self.async_show_form(
step_id="gateway", data_schema=GATEWAY_CONFIG, errors=errors
)

View file

@ -0,0 +1,47 @@
"""Code to handle a Xiaomi Gateway."""
import logging
from miio import DeviceException, gateway
_LOGGER = logging.getLogger(__name__)
class ConnectXiaomiGateway:
"""Class to async connect to a Xiaomi Gateway."""
def __init__(self, hass):
"""Initialize the entity."""
self._hass = hass
self._gateway_device = None
self._gateway_info = None
@property
def gateway_device(self):
"""Return the class containing all connections to the gateway."""
return self._gateway_device
@property
def gateway_info(self):
"""Return the class containing gateway info."""
return self._gateway_info
async def async_connect_gateway(self, host, token):
"""Connect to the Xiaomi Gateway."""
_LOGGER.debug("Initializing with host %s (token %s...)", host, token[:5])
try:
self._gateway_device = gateway.Gateway(host, token)
self._gateway_info = await self._hass.async_add_executor_job(
self._gateway_device.info
)
except DeviceException:
_LOGGER.error(
"DeviceException during setup of xiaomi gateway with host %s", host
)
return False
_LOGGER.debug(
"%s %s %s detected",
self._gateway_info.model,
self._gateway_info.firmware_version,
self._gateway_info.hardware_version,
)
return True

View file

@ -1,6 +1,7 @@
{
"domain": "xiaomi_miio",
"name": "Xiaomi miio",
"name": "Xiaomi Miio",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/xiaomi_miio",
"requirements": ["construct==2.9.45", "python-miio==0.5.0.1"],
"codeowners": ["@rytilahti", "@syssi"]

View file

@ -0,0 +1,29 @@
{
"config": {
"step": {
"user": {
"title": "Xiaomi Miio",
"description": "Select to which device you want to connect.",
"data": {
"gateway": "Connect to a Xiaomi Gateway"
}
},
"gateway": {
"title": "Connect to a Xiaomi Gateway",
"description": "You will need the API Token, see https://www.home-assistant.io/integrations/vacuum.xiaomi_miio/#retrieving-the-access-token for instructions.",
"data": {
"host": "IP adress",
"token": "API Token",
"name": "Name of the Gateway"
}
}
},
"error": {
"connect_error": "Failed to connect, please try again",
"no_device_selected": "No device selected, please select one device."
},
"abort": {
"already_configured": "Device is already configured"
}
}
}

View file

@ -0,0 +1,29 @@
{
"config": {
"step": {
"user": {
"title": "Xiaomi Miio",
"description": "Select to which device you want to connect.",
"data": {
"gateway": "Connect to a Xiaomi Gateway"
}
},
"gateway": {
"title": "Connect to a Xiaomi Gateway",
"description": "You will need the API Token, see https://www.home-assistant.io/integrations/vacuum.xiaomi_miio/#retrieving-the-access-token for instructions.",
"data": {
"host": "IP adress",
"token": "API Token",
"name": "Name of the Gateway"
}
}
},
"error": {
"connect_error": "Failed to connect, please try again",
"no_device_selected": "No device selected, please select one device."
},
"abort": {
"already_configured": "Device is already configured"
}
}
}

View file

@ -141,6 +141,7 @@ FLOWS = [
"withings",
"wled",
"wwlln",
"xiaomi_miio",
"zha",
"zwave"
]

View file

@ -0,0 +1,126 @@
"""Test the Xiaomi Miio config flow."""
from unittest.mock import Mock
from asynctest import patch
from miio import DeviceException
from homeassistant import config_entries
from homeassistant.components.xiaomi_miio import config_flow, const
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_TOKEN
TEST_HOST = "1.2.3.4"
TEST_TOKEN = "12345678901234567890123456789012"
TEST_NAME = "Test_Gateway"
TEST_MODEL = "model5"
TEST_MAC = "AB-CD-EF-GH-IJ-KL"
TEST_GATEWAY_ID = f"{TEST_MODEL}-{TEST_MAC}-gateway"
TEST_HARDWARE_VERSION = "AB123"
TEST_FIRMWARE_VERSION = "1.2.3_456"
def get_mock_info(
model=TEST_MODEL,
mac_address=TEST_MAC,
hardware_version=TEST_HARDWARE_VERSION,
firmware_version=TEST_FIRMWARE_VERSION,
):
"""Return a mock gateway info instance."""
gateway_info = Mock()
gateway_info.model = model
gateway_info.mac_address = mac_address
gateway_info.hardware_version = hardware_version
gateway_info.firmware_version = firmware_version
return gateway_info
async def test_config_flow_step_user_no_device(hass):
"""Test config flow, user step with no device selected."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(result["flow_id"], {},)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {"base": "no_device_selected"}
async def test_config_flow_step_gateway_connect_error(hass):
"""Test config flow, gateway connection error."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {config_flow.CONF_GATEWAY: True},
)
assert result["type"] == "form"
assert result["step_id"] == "gateway"
assert result["errors"] == {}
with patch(
"homeassistant.components.xiaomi_miio.gateway.gateway.Gateway.info",
side_effect=DeviceException({}),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: TEST_HOST, CONF_NAME: TEST_NAME, CONF_TOKEN: TEST_TOKEN},
)
assert result["type"] == "form"
assert result["step_id"] == "gateway"
assert result["errors"] == {"base": "connect_error"}
async def test_config_flow_gateway_success(hass):
"""Test a successful config flow."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {config_flow.CONF_GATEWAY: True},
)
assert result["type"] == "form"
assert result["step_id"] == "gateway"
assert result["errors"] == {}
mock_info = get_mock_info()
with patch(
"homeassistant.components.xiaomi_miio.gateway.gateway.Gateway.info",
return_value=mock_info,
), patch(
"homeassistant.components.xiaomi_miio.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: TEST_HOST, CONF_NAME: TEST_NAME, CONF_TOKEN: TEST_TOKEN},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
config_flow.CONF_FLOW_TYPE: config_flow.CONF_GATEWAY,
CONF_HOST: TEST_HOST,
CONF_TOKEN: TEST_TOKEN,
"gateway_id": TEST_GATEWAY_ID,
"model": TEST_MODEL,
"mac": TEST_MAC,
}