ZHA backup/restore config flow (#77044)

pull/77572/head
puddly 2022-08-30 12:40:16 -04:00 committed by GitHub
parent e48d493db4
commit f78b39bdbf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 1736 additions and 314 deletions

View File

@ -1,21 +1,39 @@
"""Config flow for ZHA."""
from __future__ import annotations
import collections
import contextlib
import copy
import json
import logging
import os
from typing import Any
import serial.tools.list_ports
import voluptuous as vol
from zigpy.application import ControllerApplication
import zigpy.backups
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.exceptions import NetworkNotFormed
from homeassistant import config_entries
from homeassistant.components import onboarding, usb, zeroconf
from homeassistant.components.file_upload import process_uploaded_file
from homeassistant.const import CONF_NAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowHandler, FlowResult
from homeassistant.helpers.selector import FileSelector, FileSelectorConfig
from homeassistant.util import dt
from .core.const import (
CONF_BAUDRATE,
CONF_DATABASE,
CONF_FLOWCONTROL,
CONF_RADIO_TYPE,
CONF_ZIGPY,
DATA_ZHA,
DATA_ZHA_CONFIG,
DEFAULT_DATABASE_NAME,
DOMAIN,
RadioType,
)
@ -27,24 +45,184 @@ SUPPORTED_PORT_SETTINGS = (
)
DECONZ_DOMAIN = "deconz"
# Only the common radio types will be autoprobed, ordered by new device popularity.
# XBee takes too long to probe since it scans through all possible bauds and likely has
# very few users to begin with.
AUTOPROBE_RADIOS = (
RadioType.ezsp,
RadioType.znp,
RadioType.deconz,
RadioType.zigate,
)
class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
FORMATION_STRATEGY = "formation_strategy"
FORMATION_FORM_NEW_NETWORK = "form_new_network"
FORMATION_REUSE_SETTINGS = "reuse_settings"
FORMATION_CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup"
FORMATION_UPLOAD_MANUAL_BACKUP = "upload_manual_backup"
VERSION = 3
CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup"
OVERWRITE_COORDINATOR_IEEE = "overwrite_coordinator_ieee"
def __init__(self):
UPLOADED_BACKUP_FILE = "uploaded_backup_file"
_LOGGER = logging.getLogger(__name__)
def _format_backup_choice(
backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True
) -> str:
"""Format network backup info into a short piece of text."""
if not pan_ids:
return dt.as_local(backup.backup_time).strftime("%c")
identifier = (
# PAN ID
f"{str(backup.network_info.pan_id)[2:]}"
# EPID
f":{str(backup.network_info.extended_pan_id).replace(':', '')}"
).lower()
return f"{dt.as_local(backup.backup_time).strftime('%c')} ({identifier})"
def _allow_overwrite_ezsp_ieee(
backup: zigpy.backups.NetworkBackup,
) -> zigpy.backups.NetworkBackup:
"""Return a new backup with the flag to allow overwriting the EZSP EUI64."""
new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
new_stack_specific.setdefault("ezsp", {})[
"i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it"
] = True
return backup.replace(
network_info=backup.network_info.replace(stack_specific=new_stack_specific)
)
def _prevent_overwrite_ezsp_ieee(
backup: zigpy.backups.NetworkBackup,
) -> zigpy.backups.NetworkBackup:
"""Return a new backup without the flag to allow overwriting the EZSP EUI64."""
if "ezsp" not in backup.network_info.stack_specific:
return backup
new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
new_stack_specific.setdefault("ezsp", {}).pop(
"i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it", None
)
return backup.replace(
network_info=backup.network_info.replace(stack_specific=new_stack_specific)
)
class BaseZhaFlow(FlowHandler):
"""Mixin for common ZHA flow steps and forms."""
def __init__(self) -> None:
"""Initialize flow instance."""
self._device_path = None
self._device_settings = None
self._radio_type = None
self._title = None
super().__init__()
async def async_step_user(self, user_input=None):
"""Handle a zha config flow start."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
self._device_path: str | None = None
self._device_settings: dict[str, Any] | None = None
self._radio_type: RadioType | None = None
self._title: str | None = None
self._current_settings: zigpy.backups.NetworkBackup | None = None
self._backups: list[zigpy.backups.NetworkBackup] = []
self._chosen_backup: zigpy.backups.NetworkBackup | None = None
@contextlib.asynccontextmanager
async def _connect_zigpy_app(self) -> ControllerApplication:
"""Connect to the radio with the current config and then clean up."""
assert self._radio_type is not None
config = self.hass.data.get(DATA_ZHA, {}).get(DATA_ZHA_CONFIG, {})
app_config = config.get(CONF_ZIGPY, {}).copy()
database_path = config.get(
CONF_DATABASE,
self.hass.config.path(DEFAULT_DATABASE_NAME),
)
# Don't create `zigbee.db` if it doesn't already exist
if not await self.hass.async_add_executor_job(os.path.exists, database_path):
database_path = None
app_config[CONF_DATABASE] = database_path
app_config[CONF_DEVICE] = self._device_settings
app_config = self._radio_type.controller.SCHEMA(app_config)
app = await self._radio_type.controller.new(
app_config, auto_form=False, start_radio=False
)
try:
await app.connect()
yield app
finally:
await app.disconnect()
async def _restore_backup(
self, backup: zigpy.backups.NetworkBackup, **kwargs: Any
) -> None:
"""Restore the provided network backup, passing through kwargs."""
if self._current_settings is not None and self._current_settings.supersedes(
self._chosen_backup
):
return
async with self._connect_zigpy_app() as app:
await app.backups.restore_backup(backup, **kwargs)
async def _detect_radio_type(self) -> bool:
"""Probe all radio types on the current port."""
for radio in AUTOPROBE_RADIOS:
_LOGGER.debug("Attempting to probe radio type %s", radio)
dev_config = radio.controller.SCHEMA_DEVICE(
{CONF_DEVICE_PATH: self._device_path}
)
probe_result = await radio.controller.probe(dev_config)
if not probe_result:
continue
# Radio library probing can succeed and return new device settings
if isinstance(probe_result, dict):
dev_config = probe_result
self._radio_type = radio
self._device_settings = dev_config
return True
return False
async def _async_create_radio_entity(self) -> FlowResult:
"""Create a config entity with the current flow state."""
assert self._title is not None
assert self._radio_type is not None
assert self._device_path is not None
assert self._device_settings is not None
device_settings = self._device_settings.copy()
device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job(
usb.get_serial_by_id, self._device_path
)
return self.async_create_entry(
title=self._title,
data={
CONF_DEVICE: device_settings,
CONF_RADIO_TYPE: self._radio_type.name,
},
)
async def async_step_choose_serial_port(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose a serial port."""
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
list_of_ports = [
f"{p}, s/n: {p.serial_number or 'n/a'}"
@ -53,48 +231,329 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
]
if not list_of_ports:
return await self.async_step_pick_radio()
return await self.async_step_manual_pick_radio_type()
list_of_ports.append(CONF_MANUAL_PATH)
if user_input is not None:
user_selection = user_input[CONF_DEVICE_PATH]
if user_selection == CONF_MANUAL_PATH:
return await self.async_step_pick_radio()
return await self.async_step_manual_pick_radio_type()
port = ports[list_of_ports.index(user_selection)]
dev_path = await self.hass.async_add_executor_job(
usb.get_serial_by_id, port.device
self._device_path = port.device
if not await self._detect_radio_type():
# Did not autodetect anything, proceed to manual selection
return await self.async_step_manual_pick_radio_type()
self._title = (
f"{port.description}, s/n: {port.serial_number or 'n/a'}"
f" - {port.manufacturer}"
if port.manufacturer
else ""
)
auto_detected_data = await detect_radios(dev_path)
if auto_detected_data is not None:
title = f"{port.description}, s/n: {port.serial_number or 'n/a'}"
title += f" - {port.manufacturer}" if port.manufacturer else ""
return self.async_create_entry(
title=title,
data=auto_detected_data,
return await self.async_step_choose_formation_strategy()
# Pre-select the currently configured port
default_port = vol.UNDEFINED
if self._device_path is not None:
for description, port in zip(list_of_ports, ports):
if port.device == self._device_path:
default_port = description
break
else:
default_port = CONF_MANUAL_PATH
schema = vol.Schema(
{
vol.Required(CONF_DEVICE_PATH, default=default_port): vol.In(
list_of_ports
)
}
)
return self.async_show_form(step_id="choose_serial_port", data_schema=schema)
# did not detect anything
self._device_path = dev_path
return await self.async_step_pick_radio()
schema = vol.Schema({vol.Required(CONF_DEVICE_PATH): vol.In(list_of_ports)})
return self.async_show_form(step_id="user", data_schema=schema)
async def async_step_pick_radio(self, user_input=None):
"""Select radio type."""
async def async_step_manual_pick_radio_type(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manually select the radio type."""
if user_input is not None:
self._radio_type = RadioType.get_by_description(user_input[CONF_RADIO_TYPE])
return await self.async_step_port_config()
return await self.async_step_manual_port_config()
# Pre-select the current radio type
default = vol.UNDEFINED
if self._radio_type is not None:
default = self._radio_type.description
schema = {
vol.Required(CONF_RADIO_TYPE, default=default): vol.In(RadioType.list())
}
schema = {vol.Required(CONF_RADIO_TYPE): vol.In(sorted(RadioType.list()))}
return self.async_show_form(
step_id="pick_radio",
step_id="manual_pick_radio_type",
data_schema=vol.Schema(schema),
)
async def async_step_manual_port_config(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Enter port settings specific for this type of radio."""
assert self._radio_type is not None
errors = {}
if user_input is not None:
self._title = user_input[CONF_DEVICE_PATH]
self._device_path = user_input[CONF_DEVICE_PATH]
self._device_settings = user_input.copy()
if await self._radio_type.controller.probe(user_input):
return await self.async_step_choose_formation_strategy()
errors["base"] = "cannot_connect"
schema = {
vol.Required(
CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED
): str
}
source = self.context.get("source")
for param, value in self._radio_type.controller.SCHEMA_DEVICE.schema.items():
if param not in SUPPORTED_PORT_SETTINGS:
continue
if source == config_entries.SOURCE_ZEROCONF and param == CONF_BAUDRATE:
value = 115200
param = vol.Required(CONF_BAUDRATE, default=value)
elif self._device_settings is not None and param in self._device_settings:
param = vol.Required(str(param), default=self._device_settings[param])
schema[param] = value
return self.async_show_form(
step_id="manual_port_config",
data_schema=vol.Schema(schema),
errors=errors,
)
async def _async_load_network_settings(self) -> None:
"""Connect to the radio and load its current network settings."""
async with self._connect_zigpy_app() as app:
# Check if the stick has any settings and load them
try:
await app.load_network_info()
except NetworkNotFormed:
pass
else:
self._current_settings = zigpy.backups.NetworkBackup(
network_info=app.state.network_info,
node_info=app.state.node_info,
)
# The list of backups will always exist
self._backups = app.backups.backups.copy()
async def async_step_choose_formation_strategy(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose how to deal with the current radio's settings."""
await self._async_load_network_settings()
strategies = []
# Check if we have any automatic backups *and* if the backups differ from
# the current radio settings, if they exist (since restoring would be redundant)
if self._backups and (
self._current_settings is None
or any(
not backup.is_compatible_with(self._current_settings)
for backup in self._backups
)
):
strategies.append(CHOOSE_AUTOMATIC_BACKUP)
if self._current_settings is not None:
strategies.append(FORMATION_REUSE_SETTINGS)
strategies.append(FORMATION_UPLOAD_MANUAL_BACKUP)
strategies.append(FORMATION_FORM_NEW_NETWORK)
return self.async_show_menu(
step_id="choose_formation_strategy",
menu_options=strategies,
)
async def async_step_reuse_settings(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Reuse the existing network settings on the stick."""
return await self._async_create_radio_entity()
async def async_step_form_new_network(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Form a brand new network."""
async with self._connect_zigpy_app() as app:
await app.form_network()
return await self._async_create_radio_entity()
def _parse_uploaded_backup(
self, uploaded_file_id: str
) -> zigpy.backups.NetworkBackup:
"""Read and parse an uploaded backup JSON file."""
with process_uploaded_file(self.hass, uploaded_file_id) as file_path:
contents = file_path.read_text()
return zigpy.backups.NetworkBackup.from_dict(json.loads(contents))
async def async_step_upload_manual_backup(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Upload and restore a coordinator backup JSON file."""
errors = {}
if user_input is not None:
try:
self._chosen_backup = await self.hass.async_add_executor_job(
self._parse_uploaded_backup, user_input[UPLOADED_BACKUP_FILE]
)
except ValueError:
errors["base"] = "invalid_backup_json"
else:
return await self.async_step_maybe_confirm_ezsp_restore()
return self.async_show_form(
step_id="upload_manual_backup",
data_schema=vol.Schema(
{
vol.Required(UPLOADED_BACKUP_FILE): FileSelector(
FileSelectorConfig(accept=".json,application/json")
)
}
),
errors=errors,
)
async def async_step_choose_automatic_backup(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose an automatic backup."""
if self.show_advanced_options:
# Always show the PAN IDs when in advanced mode
choices = [
_format_backup_choice(backup, pan_ids=True) for backup in self._backups
]
else:
# Only show the PAN IDs for multiple backups taken on the same day
num_backups_on_date = collections.Counter(
backup.backup_time.date() for backup in self._backups
)
choices = [
_format_backup_choice(
backup, pan_ids=(num_backups_on_date[backup.backup_time.date()] > 1)
)
for backup in self._backups
]
if user_input is not None:
index = choices.index(user_input[CHOOSE_AUTOMATIC_BACKUP])
self._chosen_backup = self._backups[index]
return await self.async_step_maybe_confirm_ezsp_restore()
return self.async_show_form(
step_id="choose_automatic_backup",
data_schema=vol.Schema(
{
vol.Required(CHOOSE_AUTOMATIC_BACKUP, default=choices[0]): vol.In(
choices
),
}
),
)
async def async_step_maybe_confirm_ezsp_restore(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm restore for EZSP radios that require permanent IEEE writes."""
assert self._chosen_backup is not None
if self._radio_type != RadioType.ezsp:
await self._restore_backup(self._chosen_backup)
return await self._async_create_radio_entity()
# We have no way to partially load network settings if no network is formed
if self._current_settings is None:
# Since we are going to be restoring the backup anyways, write it to the
# radio without overwriting the IEEE but don't take a backup with these
# temporary settings
temp_backup = _prevent_overwrite_ezsp_ieee(self._chosen_backup)
await self._restore_backup(temp_backup, create_new=False)
await self._async_load_network_settings()
assert self._current_settings is not None
if (
self._current_settings.node_info.ieee == self._chosen_backup.node_info.ieee
or not self._current_settings.network_info.metadata["ezsp"][
"can_write_custom_eui64"
]
):
# No point in prompting the user if the backup doesn't have a new IEEE
# address or if there is no way to overwrite the IEEE address a second time
await self._restore_backup(self._chosen_backup)
return await self._async_create_radio_entity()
if user_input is not None:
backup = self._chosen_backup
if user_input[OVERWRITE_COORDINATOR_IEEE]:
backup = _allow_overwrite_ezsp_ieee(backup)
# If the user declined to overwrite the IEEE *and* we wrote the backup to
# their empty radio above, restoring it again would be redundant.
await self._restore_backup(backup)
return await self._async_create_radio_entity()
return self.async_show_form(
step_id="maybe_confirm_ezsp_restore",
data_schema=vol.Schema(
{vol.Required(OVERWRITE_COORDINATOR_IEEE, default=True): bool}
),
)
class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 3
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> config_entries.OptionsFlow:
"""Create the options flow."""
return ZhaOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a zha config flow start."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
return await self.async_step_choose_serial_port(user_input)
async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult:
"""Handle usb discovery."""
vid = discovery_info.vid
@ -118,9 +577,8 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
# If they already have a discovery for deconz
# we ignore the usb discovery as they probably
# want to use it there instead
# If they already have a discovery for deconz we ignore the usb discovery as
# they probably want to use it there instead
if self.hass.config_entries.flow.async_progress_by_handler(DECONZ_DOMAIN):
return self.async_abort(reason="not_zha_device")
for entry in self.hass.config_entries.async_entries(DECONZ_DOMAIN):
@ -140,19 +598,18 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = {CONF_NAME: self._title}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):
"""Confirm a USB discovery."""
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm a discovery."""
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
auto_detected_data = await detect_radios(self._device_path)
if auto_detected_data is None:
if not await self._detect_radio_type():
# This path probably will not happen now that we have
# more precise USB matching unless there is a problem
# with the device
return self.async_abort(reason="usb_probe_failed")
return self.async_create_entry(
title=self._title,
data=auto_detected_data,
)
return await self.async_step_choose_formation_strategy()
return self.async_show_form(
step_id="confirm",
@ -188,61 +645,22 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
self.context["title_placeholders"] = {
CONF_NAME: node_name,
}
self.context["title_placeholders"] = {CONF_NAME: node_name}
self._title = device_path
self._device_path = device_path
if "efr32" in radio_type:
self._radio_type = RadioType.ezsp.name
self._radio_type = RadioType.ezsp
elif "zigate" in radio_type:
self._radio_type = RadioType.zigate.name
self._radio_type = RadioType.zigate
else:
self._radio_type = RadioType.znp.name
self._radio_type = RadioType.znp
return await self.async_step_port_config()
return await self.async_step_manual_port_config()
async def async_step_port_config(self, user_input=None):
"""Enter port settings specific for this type of radio."""
errors = {}
app_cls = RadioType[self._radio_type].controller
if user_input is not None:
self._device_path = user_input.get(CONF_DEVICE_PATH)
if await app_cls.probe(user_input):
serial_by_id = await self.hass.async_add_executor_job(
usb.get_serial_by_id, user_input[CONF_DEVICE_PATH]
)
user_input[CONF_DEVICE_PATH] = serial_by_id
return self.async_create_entry(
title=user_input[CONF_DEVICE_PATH],
data={CONF_DEVICE: user_input, CONF_RADIO_TYPE: self._radio_type},
)
errors["base"] = "cannot_connect"
schema = {
vol.Required(
CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED
): str
}
radio_schema = app_cls.SCHEMA_DEVICE.schema
if isinstance(radio_schema, vol.Schema):
radio_schema = radio_schema.schema
source = self.context.get("source")
for param, value in radio_schema.items():
if param in SUPPORTED_PORT_SETTINGS:
schema[param] = value
if source == config_entries.SOURCE_ZEROCONF and param == CONF_BAUDRATE:
schema[param] = 115200
return self.async_show_form(
step_id="port_config",
data_schema=vol.Schema(schema),
errors=errors,
)
async def async_step_hardware(self, data=None):
async def async_step_hardware(
self, data: dict[str, Any] | None = None
) -> FlowResult:
"""Handle hardware flow."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -250,40 +668,39 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="invalid_hardware_data")
if data.get("radio_type") != "efr32":
return self.async_abort(reason="invalid_hardware_data")
self._radio_type = RadioType.ezsp.name
app_cls = RadioType[self._radio_type].controller
self._radio_type = RadioType.ezsp
schema = {
vol.Required(
CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED
): str
}
radio_schema = app_cls.SCHEMA_DEVICE.schema
radio_schema = self._radio_type.controller.SCHEMA_DEVICE.schema
assert not isinstance(radio_schema, vol.Schema)
for param, value in radio_schema.items():
if param in SUPPORTED_PORT_SETTINGS:
schema[param] = value
try:
self._device_settings = vol.Schema(schema)(data.get("port"))
device_settings = vol.Schema(schema)(data.get("port"))
except vol.Invalid:
return self.async_abort(reason="invalid_hardware_data")
self._title = data.get("name", data["port"]["path"])
self._device_path = device_settings.pop(CONF_DEVICE_PATH)
self._device_settings = device_settings
self._set_confirm_only()
return await self.async_step_confirm_hardware()
async def async_step_confirm_hardware(self, user_input=None):
async def async_step_confirm_hardware(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm a hardware discovery."""
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
return self.async_create_entry(
title=self._title,
data={
CONF_DEVICE: self._device_settings,
CONF_RADIO_TYPE: self._radio_type,
},
)
return await self._async_create_radio_entity()
return self.async_show_form(
step_id="confirm_hardware",
@ -291,14 +708,65 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
async def detect_radios(dev_path: str) -> dict[str, Any] | None:
"""Probe all radio types on the device port."""
for radio in RadioType:
dev_config = radio.controller.SCHEMA_DEVICE({CONF_DEVICE_PATH: dev_path})
probe_result = await radio.controller.probe(dev_config)
if probe_result:
if isinstance(probe_result, dict):
return {CONF_RADIO_TYPE: radio.name, CONF_DEVICE: probe_result}
return {CONF_RADIO_TYPE: radio.name, CONF_DEVICE: dev_config}
class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow):
"""Handle an options flow."""
return None
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
super().__init__()
self.config_entry = config_entry
self._device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
self._device_settings = config_entry.data[CONF_DEVICE]
self._radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]]
self._title = config_entry.title
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Launch the options flow."""
if user_input is not None:
try:
await self.hass.config_entries.async_unload(self.config_entry.entry_id)
except config_entries.OperationNotAllowed:
# ZHA is not running
pass
return await self.async_step_choose_serial_port()
return self.async_show_form(step_id="init")
async def _async_create_radio_entity(self):
"""Re-implementation of the base flow's final step to update the config."""
device_settings = self._device_settings.copy()
device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job(
usb.get_serial_by_id, self._device_path
)
# Avoid creating both `.options` and `.data` by directly writing `data` here
self.hass.config_entries.async_update_entry(
entry=self.config_entry,
data={
CONF_DEVICE: device_settings,
CONF_RADIO_TYPE: self._radio_type.name,
},
options=self.config_entry.options,
)
# Reload ZHA after we finish
await self.hass.config_entries.async_setup(self.config_entry.entry_id)
# Intentionally do not set `data` to avoid creating `options`, we set it above
return self.async_create_entry(title=self._title, data={})
def async_remove(self):
"""Maybe reload ZHA if the flow is aborted."""
if self.config_entry.state not in (
config_entries.ConfigEntryState.SETUP_ERROR,
config_entries.ConfigEntryState.NOT_LOADED,
):
return
self.hass.async_create_task(
self.hass.config_entries.async_setup(self.config_entry.entry_id)
)

View File

@ -236,14 +236,14 @@ _ControllerClsType = type[zigpy.application.ControllerApplication]
class RadioType(enum.Enum):
"""Possible options for radio type."""
znp = (
"ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2",
zigpy_znp.zigbee.application.ControllerApplication,
)
ezsp = (
"EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis",
bellows.zigbee.application.ControllerApplication,
)
znp = (
"ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2",
zigpy_znp.zigbee.application.ControllerApplication,
)
deconz = (
"deCONZ = dresden elektronik deCONZ protocol: ConBee I/II, RaspBee I/II",
zigpy_deconz.zigbee.application.ControllerApplication,
@ -263,11 +263,11 @@ class RadioType(enum.Enum):
return [e.description for e in RadioType]
@classmethod
def get_by_description(cls, description: str) -> str:
def get_by_description(cls, description: str) -> RadioType:
"""Get radio by description."""
for radio in cls:
if radio.description == description:
return radio.name
return radio
raise ValueError
def __init__(self, description: str, controller_cls: _ControllerClsType) -> None:

View File

@ -93,6 +93,7 @@
"name": "*zigate*"
}
],
"dependencies": ["file_upload"],
"after_dependencies": ["onboarding", "usb", "zeroconf"],
"iot_class": "local_polling",
"loggers": [

View File

@ -2,10 +2,10 @@
"config": {
"flow_title": "{name}",
"step": {
"user": {
"title": "ZHA",
"choose_serial_port": {
"title": "Select a Serial Port",
"data": { "path": "Serial Device Path" },
"description": "Select serial port for Zigbee radio"
"description": "Select the serial port for your Zigbee radio"
},
"confirm": {
"description": "Do you want to setup {name}?"
@ -13,23 +13,55 @@
"confirm_hardware": {
"description": "Do you want to setup {name}?"
},
"pick_radio": {
"manual_pick_radio_type": {
"data": { "radio_type": "Radio Type" },
"title": "Radio Type",
"description": "Pick a type of your Zigbee radio"
"description": "Pick your Zigbee radio type"
},
"port_config": {
"title": "Settings",
"description": "Enter port specific settings",
"manual_port_config": {
"title": "Serial Port Settings",
"description": "Enter the serial port settings",
"data": {
"path": "Serial device path",
"baudrate": "port speed",
"flow_control": "data flow control"
}
},
"choose_formation_strategy": {
"title": "Network Formation",
"description": "Choose the network settings for your radio.",
"menu_options": {
"form_new_network": "Erase network settings and form a new network",
"reuse_settings": "Keep radio network settings",
"choose_automatic_backup": "Restore an automatic backup",
"upload_manual_backup": "Upload a manual backup"
}
},
"choose_automatic_backup": {
"title": "Restore Automatic Backup",
"description": "Restore your network settings from an automatic backup",
"data": {
"choose_automatic_backup": "Choose an automatic backup"
}
},
"upload_manual_backup": {
"title": "Upload a Manual Backup",
"description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.",
"data": {
"uploaded_backup_file": "Upload a file"
}
},
"maybe_confirm_ezsp_restore": {
"title": "Overwrite Radio IEEE Address",
"description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.",
"data": {
"overwrite_coordinator_ieee": "Permanently replace the radio IEEE address"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_backup_json": "Invalid backup JSON"
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
@ -37,6 +69,78 @@
"usb_probe_failed": "Failed to probe the usb device"
}
},
"options": {
"flow_title": "[%key:component::zha::config::flow_title%]",
"step": {
"init": {
"title": "Reconfigure ZHA",
"description": "ZHA will be stopped. Do you wish to continue?"
},
"choose_serial_port": {
"title": "[%key:component::zha::config::step::choose_serial_port::title%]",
"data": {
"path": "[%key:component::zha::config::step::choose_serial_port::data::path%]"
},
"description": "[%key:component::zha::config::step::choose_serial_port::description%]"
},
"manual_pick_radio_type": {
"data": {
"radio_type": "[%key:component::zha::config::step::manual_pick_radio_type::data::radio_type%]"
},
"title": "[%key:component::zha::config::step::manual_pick_radio_type::title%]",
"description": "[%key:component::zha::config::step::manual_pick_radio_type::description%]"
},
"manual_port_config": {
"title": "[%key:component::zha::config::step::manual_port_config::title%]",
"description": "[%key:component::zha::config::step::manual_port_config::description%]",
"data": {
"path": "[%key:component::zha::config::step::manual_port_config::data::path%]",
"baudrate": "[%key:component::zha::config::step::manual_port_config::data::baudrate%]",
"flow_control": "[%key:component::zha::config::step::manual_port_config::data::flow_control%]"
}
},
"choose_formation_strategy": {
"title": "[%key:component::zha::config::step::choose_formation_strategy::title%]",
"description": "[%key:component::zha::config::step::choose_formation_strategy::description%]",
"menu_options": {
"form_new_network": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::form_new_network%]",
"reuse_settings": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::reuse_settings%]",
"choose_automatic_backup": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::choose_automatic_backup%]",
"upload_manual_backup": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::upload_manual_backup%]"
}
},
"choose_automatic_backup": {
"title": "[%key:component::zha::config::step::choose_automatic_backup::title%]",
"description": "[%key:component::zha::config::step::choose_automatic_backup::description%]",
"data": {
"choose_automatic_backup": "[%key:component::zha::config::step::choose_automatic_backup::data::choose_automatic_backup%]"
}
},
"upload_manual_backup": {
"title": "[%key:component::zha::config::step::upload_manual_backup::title%]",
"description": "[%key:component::zha::config::step::upload_manual_backup::description%]",
"data": {
"uploaded_backup_file": "[%key:component::zha::config::step::upload_manual_backup::data::uploaded_backup_file%]"
}
},
"maybe_confirm_ezsp_restore": {
"title": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::title%]",
"description": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::description%]",
"data": {
"overwrite_coordinator_ieee": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::data::overwrite_coordinator_ieee%]"
}
}
},
"error": {
"cannot_connect": "[%key:component::zha::config::error::cannot_connect%]",
"invalid_backup_json": "[%key:component::zha::config::error::invalid_backup_json%]"
},
"abort": {
"single_instance_allowed": "[%key:component::zha::config::abort::single_instance_allowed%]",
"not_zha_device": "[%key:component::zha::config::abort::not_zha_device%]",
"usb_probe_failed": "[%key:component::zha::config::abort::usb_probe_failed%]"
}
},
"config_panel": {
"zha_options": {
"title": "Global Options",

View File

@ -6,38 +6,70 @@
"usb_probe_failed": "Failed to probe the usb device"
},
"error": {
"cannot_connect": "Failed to connect"
"cannot_connect": "Failed to connect",
"invalid_backup_json": "Invalid backup JSON"
},
"flow_title": "{name}",
"step": {
"choose_automatic_backup": {
"data": {
"choose_automatic_backup": "Choose an automatic backup"
},
"description": "Restore your network settings from an automatic backup",
"title": "Restore Automatic Backup"
},
"choose_formation_strategy": {
"description": "Choose the network settings for your radio.",
"menu_options": {
"choose_automatic_backup": "Restore an automatic backup",
"form_new_network": "Erase network settings and form a new network",
"reuse_settings": "Keep radio network settings",
"upload_manual_backup": "Upload a manual backup"
},
"title": "Network Formation"
},
"choose_serial_port": {
"data": {
"path": "Serial Device Path"
},
"description": "Select the serial port for your Zigbee radio",
"title": "Select a Serial Port"
},
"confirm": {
"description": "Do you want to setup {name}?"
},
"confirm_hardware": {
"description": "Do you want to setup {name}?"
},
"pick_radio": {
"manual_pick_radio_type": {
"data": {
"radio_type": "Radio Type"
},
"description": "Pick a type of your Zigbee radio",
"description": "Pick your Zigbee radio type",
"title": "Radio Type"
},
"port_config": {
"manual_port_config": {
"data": {
"baudrate": "port speed",
"flow_control": "data flow control",
"path": "Serial device path"
},
"description": "Enter port specific settings",
"title": "Settings"
"description": "Enter the serial port settings",
"title": "Serial Port Settings"
},
"user": {
"maybe_confirm_ezsp_restore": {
"data": {
"path": "Serial Device Path"
"overwrite_coordinator_ieee": "Permanently replace the radio IEEE address"
},
"description": "Select serial port for Zigbee radio",
"title": "ZHA"
"description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.",
"title": "Overwrite Radio IEEE Address"
},
"upload_manual_backup": {
"data": {
"uploaded_backup_file": "Upload a file"
},
"description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.",
"title": "Upload a Manual Backup"
}
}
},
@ -114,5 +146,77 @@
"remote_button_short_release": "\"{subtype}\" button released",
"remote_button_triple_press": "\"{subtype}\" button triple clicked"
}
},
"options": {
"abort": {
"not_zha_device": "This device is not a zha device",
"single_instance_allowed": "Already configured. Only a single configuration possible.",
"usb_probe_failed": "Failed to probe the usb device"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_backup_json": "Invalid backup JSON"
},
"flow_title": "{name}",
"step": {
"choose_automatic_backup": {
"data": {
"choose_automatic_backup": "Choose an automatic backup"
},
"description": "Restore your network settings from an automatic backup",
"title": "Restore Automatic Backup"
},
"choose_formation_strategy": {
"description": "Choose the network settings for your radio.",
"menu_options": {
"choose_automatic_backup": "Restore an automatic backup",
"form_new_network": "Erase network settings and form a new network",
"reuse_settings": "Keep radio network settings",
"upload_manual_backup": "Upload a manual backup"
},
"title": "Network Formation"
},
"choose_serial_port": {
"data": {
"path": "Serial Device Path"
},
"description": "Select the serial port for your Zigbee radio",
"title": "Select a Serial Port"
},
"init": {
"description": "ZHA will be stopped. Do you wish to continue?",
"title": "Reconfigure ZHA"
},
"manual_pick_radio_type": {
"data": {
"radio_type": "Radio Type"
},
"description": "Pick your Zigbee radio type",
"title": "Radio Type"
},
"manual_port_config": {
"data": {
"baudrate": "port speed",
"flow_control": "data flow control",
"path": "Serial device path"
},
"description": "Enter the serial port settings",
"title": "Serial Port Settings"
},
"maybe_confirm_ezsp_restore": {
"data": {
"overwrite_coordinator_ieee": "Permanently replace the radio IEEE address"
},
"description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.",
"title": "Overwrite Radio IEEE Address"
},
"upload_manual_backup": {
"data": {
"uploaded_backup_file": "Upload a file"
},
"description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.",
"title": "Upload a Manual Backup"
}
}
}
}

File diff suppressed because it is too large Load Diff