273 lines
9.2 KiB
Python
273 lines
9.2 KiB
Python
import homeassistant.helpers.config_validation as cv
|
|
import voluptuous as vol
|
|
from homeassistant.config_entries import ConfigFlow, OptionsFlow, ConfigEntry
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers.aiohttp_client import async_create_clientsession
|
|
|
|
from . import DOMAIN
|
|
from .core import utils
|
|
from .core.xiaomi_cloud import MiCloud
|
|
|
|
ACTIONS = {
|
|
"cloud": "Add Mi Cloud Account",
|
|
"token": "Add Gateway using Token"
|
|
}
|
|
|
|
SERVERS = {
|
|
"cn": "China",
|
|
"de": "Europe",
|
|
"i2": "India",
|
|
"ru": "Russia",
|
|
"sg": "Singapore",
|
|
"us": "United States"
|
|
}
|
|
|
|
OPT_DEBUG = {
|
|
"true": "Basic logs",
|
|
"mqtt": "MQTT logs",
|
|
"zigbee": "Zigbee logs",
|
|
}
|
|
|
|
|
|
class FlowHandler(ConfigFlow, domain=DOMAIN):
|
|
VERSION = 1
|
|
cloud = None
|
|
|
|
async def async_step_user(self, user_input=None):
|
|
if user_input is not None:
|
|
if user_input["action"] == "cloud":
|
|
return await self.async_step_cloud()
|
|
elif user_input["action"] == "token":
|
|
return await self.async_step_token()
|
|
else:
|
|
device = next(
|
|
device for device in self.hass.data[DOMAIN]["devices"]
|
|
if device["did"] == user_input["action"]
|
|
)
|
|
return self.async_show_form(
|
|
step_id="token",
|
|
data_schema=vol.Schema({
|
|
vol.Required("host", default=device["localip"]): str,
|
|
vol.Required("token", default=device["token"]): str,
|
|
vol.Required("telnet_cmd"): str,
|
|
}),
|
|
)
|
|
|
|
if DOMAIN in self.hass.data and "devices" in self.hass.data[DOMAIN]:
|
|
for device in self.hass.data[DOMAIN]["devices"]:
|
|
if (device["model"] in utils.SUPPORTED_MODELS and
|
|
device["did"] not in ACTIONS):
|
|
name = f"Add {device['name']} ({device['localip']})"
|
|
ACTIONS[device["did"]] = name
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=vol.Schema({
|
|
vol.Required("action", default="cloud"): vol.In(ACTIONS)
|
|
})
|
|
)
|
|
|
|
async def async_step_cloud(self, user_input=None, error=None):
|
|
if user_input:
|
|
if not user_input["servers"]:
|
|
return await self.async_step_cloud(error="no_servers")
|
|
|
|
session = async_create_clientsession(self.hass)
|
|
cloud = MiCloud(session)
|
|
if await cloud.login(
|
|
user_input["username"], user_input["password"]
|
|
):
|
|
user_input.update(cloud.auth)
|
|
return self.async_create_entry(
|
|
title=user_input["username"], data=user_input
|
|
)
|
|
|
|
else:
|
|
return await self.async_step_cloud(error="cant_login")
|
|
|
|
return self.async_show_form(
|
|
step_id="cloud",
|
|
data_schema=vol.Schema({
|
|
vol.Required("username"): str,
|
|
vol.Required("password"): str,
|
|
vol.Required("servers", default=["cn"]):
|
|
cv.multi_select(SERVERS)
|
|
}),
|
|
errors={"base": error} if error else None
|
|
)
|
|
|
|
async def async_step_token(self, user_input=None, error=None):
|
|
"""GUI > Configuration > Integrations > Plus > Xiaomi Gateway 3"""
|
|
if user_input is not None:
|
|
error = await utils.check_gateway(**user_input)
|
|
if error:
|
|
return await self.async_step_token(error=error)
|
|
|
|
return self.async_create_entry(
|
|
title=user_input["host"], data=user_input
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="token",
|
|
data_schema=vol.Schema({
|
|
vol.Required("host"): str,
|
|
vol.Required("token"): str,
|
|
vol.Required("telnet_cmd"): str,
|
|
}),
|
|
errors={"base": error} if error else None
|
|
)
|
|
|
|
@staticmethod
|
|
@callback
|
|
def async_get_options_flow(entry: ConfigEntry):
|
|
return OptionsFlowHandler(entry)
|
|
|
|
|
|
TITLE = "Xiaomi Gateway 3"
|
|
|
|
ZHA_NOTIFICATION = """For **ZHA**, goto:
|
|
|
|
Configuration > Integrations > Add Integration > Zigbee Home Automation:
|
|
|
|
- Radio Type: **EZSP**
|
|
- Path: `socket://{0}:8888`
|
|
|
|
For **zigbee2mqtt**, goto:
|
|
|
|
Supervisor > Zigbee2mqtt > Configuration:
|
|
|
|
```
|
|
serial:
|
|
port: 'tcp://{0}:8888'
|
|
adapter: ezsp
|
|
```
|
|
"""
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class OptionsFlowHandler(OptionsFlow):
|
|
def __init__(self, entry: ConfigEntry):
|
|
self.entry = entry
|
|
|
|
async def async_step_init(self, user_input=None):
|
|
if "servers" in self.entry.data:
|
|
return await self.async_step_cloud()
|
|
else:
|
|
return await self.async_step_user()
|
|
|
|
async def async_step_cloud(self, user_input=None):
|
|
if user_input is not None:
|
|
did = user_input["did"]
|
|
device = next(
|
|
device for device in self.hass.data[DOMAIN]["devices"]
|
|
if device["did"] == did
|
|
)
|
|
|
|
if device["pid"] != 6:
|
|
device_info = (
|
|
f"Name: {device['name']}\n"
|
|
f"Model: {device['model']}\n"
|
|
f"IP: {device['localip']}\n"
|
|
f"MAC: {device['mac']}\n"
|
|
f"Token: {device['token']}"
|
|
)
|
|
else:
|
|
bindkey = await utils.get_bindkey(
|
|
self.hass.data[DOMAIN]["cloud"], did
|
|
)
|
|
device_info = (
|
|
f"Name: {device['name']}\n"
|
|
f"Model: {device['model']}\n"
|
|
f"MAC: {device['mac']}\n"
|
|
f"Bindkey: {bindkey}\n"
|
|
)
|
|
|
|
if device["model"] == "lumi.gateway.v3":
|
|
device_info += "\nLAN key: " + await utils.get_lan_key(
|
|
device["localip"], device["token"]
|
|
)
|
|
elif ".vacuum." in device["model"]:
|
|
device_info += "\nRooms: " + await utils.get_room_mapping(
|
|
self.hass.data[DOMAIN]["cloud"],
|
|
device["localip"], device["token"],
|
|
)
|
|
elif device["model"] == "yeelink.light.bslamp2":
|
|
device_info += "\nLAN mode: " + await utils.enable_bslamp2_lan(
|
|
device["localip"], device["token"]
|
|
)
|
|
|
|
elif not self.hass.data[DOMAIN].get("devices"):
|
|
device_info = "No devices in account"
|
|
else:
|
|
device_info = "Choose a device from the list"
|
|
|
|
devices = {}
|
|
for device in self.hass.data[DOMAIN].get("devices", []):
|
|
# 0 - wifi, 6 - ble, 8 - wifi+ble
|
|
if device["pid"] in (0, 8):
|
|
info = device["localip"]
|
|
elif device["pid"] == 6:
|
|
info = "BLE"
|
|
else:
|
|
continue
|
|
devices[device["did"]] = f"{device['name']} ({info})"
|
|
|
|
return self.async_show_form(
|
|
step_id="cloud",
|
|
data_schema=vol.Schema({
|
|
vol.Required("did"): vol.In(devices)
|
|
}),
|
|
description_placeholders={
|
|
"device_info": device_info
|
|
}
|
|
)
|
|
|
|
async def async_step_user(self, user_input=None):
|
|
if user_input:
|
|
old_mode = self.entry.options.get("zha", False)
|
|
new_mode = user_input["zha"]
|
|
if new_mode != old_mode:
|
|
host = user_input["host"]
|
|
|
|
# change zigbee firmware if needed
|
|
if not await utils.update_zigbee_firmware(
|
|
self.hass, host, new_mode
|
|
):
|
|
raise Exception("Can't update zigbee firmware")
|
|
|
|
if new_mode is True:
|
|
self.hass.components.persistent_notification.async_create(
|
|
ZHA_NOTIFICATION.format(host), TITLE
|
|
)
|
|
|
|
return self.async_create_entry(title='', data=user_input)
|
|
|
|
host = self.entry.options["host"]
|
|
token = self.entry.options["token"]
|
|
telnet_cmd = self.entry.options.get("telnet_cmd", "")
|
|
ble = self.entry.options.get("ble", True)
|
|
stats = self.entry.options.get("stats", False)
|
|
debug = self.entry.options.get("debug", [])
|
|
buzzer = self.entry.options.get("buzzer", False)
|
|
memory = self.entry.options.get("memory", False)
|
|
zha = self.entry.options.get("zha", False)
|
|
|
|
# filter only supported items
|
|
debug = [k for k in debug if k in OPT_DEBUG]
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=vol.Schema({
|
|
vol.Required("host", default=host): str,
|
|
vol.Required("token", default=token): str,
|
|
vol.Optional("telnet_cmd", default=telnet_cmd): str,
|
|
vol.Required("ble", default=ble): bool,
|
|
vol.Optional("stats", default=stats): bool,
|
|
vol.Optional("debug", default=debug):
|
|
cv.multi_select(OPT_DEBUG),
|
|
vol.Optional("buzzer", default=buzzer): bool,
|
|
vol.Optional("memory", default=memory): bool,
|
|
vol.Optional("zha", default=zha): bool,
|
|
}),
|
|
)
|