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

Add config flow to insteon component (#36467) 

* Squashed

* Fix requirements_all

* Update homeassistant/components/insteon/__init__.py

Only update options if the result is to create the entry.

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/insteon/__init__.py

No return value needed.

Co-authored-by: J. Nick Koston <nick@koston.org>

* Ref RESULT_TYPE_CREATE_ENTRY correctly

* Return result back to import config process

* Make DOMAIN ref more clear

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/38778/head
Tom Harris 2020-08-11 19:04:44 -04:00 committed by GitHub
parent 6bdb2f3d11
commit b1fd931cdc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 1740 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.coveragerc
View File

@ -394,7 +394,18 @@ omit =
homeassistant/components/ihc/*
homeassistant/components/imap/sensor.py
homeassistant/components/imap_email_content/sensor.py
homeassistant/components/insteon/*
homeassistant/components/insteon/__init__.py
homeassistant/components/insteon/binary_sensor.py
homeassistant/components/insteon/climate.py
homeassistant/components/insteon/const.py
homeassistant/components/insteon/cover.py
homeassistant/components/insteon/fan.py
homeassistant/components/insteon/insteon_entity.py
homeassistant/components/insteon/ipdb.py
homeassistant/components/insteon/light.py
homeassistant/components/insteon/schemas.py
homeassistant/components/insteon/switch.py
homeassistant/components/insteon/utils.py
homeassistant/components/incomfort/*
homeassistant/components/intesishome/*
homeassistant/components/ios/*

110
homeassistant/components/insteon/__init__.py
View File

@ -4,24 +4,16 @@ import logging
from pyinsteon import async_close, async_connect, devices
from homeassistant.const import (
CONF_HOST,
CONF_PLATFORM,
CONF_PORT,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_PLATFORM, EVENT_HOMEASSISTANT_STOP
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY
from homeassistant.exceptions import ConfigEntryNotReady
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_FIRMWARE,
CONF_HOUSECODE,
CONF_HUB_PASSWORD,
CONF_HUB_USERNAME,
CONF_HUB_VERSION,
CONF_IP_PORT,
CONF_OVERRIDE,
CONF_PRODUCT_KEY,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
@ -29,7 +21,7 @@ from .const import (
INSTEON_COMPONENTS,
ON_OFF_EVENTS,
)
from .schemas import CONFIG_SCHEMA # noqa F440
from .schemas import convert_yaml_to_config_flow
from .utils import (
add_on_off_event_device,
async_register_services,
@ -63,10 +55,10 @@ async def async_id_unknown_devices(config_dir):
await devices.async_save(workdir=config_dir)
async def async_setup_platforms(hass, config):
async def async_setup_platforms(hass, config_entry):
"""Initiate the connection and services."""
tasks = [
hass.helpers.discovery.async_load_platform(component, DOMAIN, {}, config)
hass.config_entries.async_forward_entry_setup(config_entry, component)
for component in INSTEON_COMPONENTS
]
await asyncio.gather(*tasks)
@ -78,12 +70,26 @@ async def async_setup_platforms(hass, config):
add_on_off_event_device(hass, device)
_LOGGER.debug("Insteon device count: %s", len(devices))
register_new_device_callback(hass, config)
register_new_device_callback(hass)
async_register_services(hass)
device_registry = await hass.helpers.device_registry.async_get_registry()
device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, str(devices.modem.address))},
manufacturer="Smart Home",
name=f"{devices.modem.description} {devices.modem.address}",
model=f"{devices.modem.model} (0x{devices.modem.cat:02x}, 0x{devices.modem.subcat:02x})",
sw_version=f"{devices.modem.firmware:02x} Engine Version: {devices.modem.engine_version}",
)
# Make a copy of addresses due to edge case where the list of devices could change during status update
# Cannot be done concurrently due to issues with the underlying protocol.
for address in devices:
await devices[address].async_status()
for address in list(devices):
try:
await devices[address].async_status()
except AttributeError:
pass
await async_id_unknown_devices(hass.config.config_dir)
@ -92,57 +98,57 @@ async def close_insteon_connection(*args):
await async_close()
async def async_import_config(hass, conf):
"""Set up all of the config imported from yaml."""
data, options = convert_yaml_to_config_flow(conf)
# Create a config entry with the connection data
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
# If this is the first time we ran, update the config options
if result["type"] == RESULT_TYPE_CREATE_ENTRY and options:
entry = result["result"]
hass.config_entries.async_update_entry(
entry=entry, options=options,
)
return result
async def async_setup(hass, config):
"""Set up the connection to the modem."""
"""Set up the Insteon platform."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
port = conf.get(CONF_PORT)
host = conf.get(CONF_HOST)
ip_port = conf.get(CONF_IP_PORT)
username = conf.get(CONF_HUB_USERNAME)
password = conf.get(CONF_HUB_PASSWORD)
hub_version = conf.get(CONF_HUB_VERSION)
hass.async_create_task(async_import_config(hass, conf))
return True
if host:
_LOGGER.info("Connecting to Insteon Hub on %s:%d", host, ip_port)
else:
_LOGGER.info("Connecting to Insteon PLM on %s", port)
try:
await async_connect(
device=port,
host=host,
port=ip_port,
username=username,
password=password,
hub_version=hub_version,
)
except ConnectionError:
_LOGGER.error("Could not connect to Insteon modem")
return False
_LOGGER.info("Connection to Insteon modem successful")
async def async_setup_entry(hass, entry):
"""Set up an Insteon entry."""
if not devices.modem:
try:
await async_connect(**entry.data)
except ConnectionError as exception:
_LOGGER.error("Could not connect to Insteon modem")
raise ConfigEntryNotReady from exception
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, close_insteon_connection)
conf = config[DOMAIN]
overrides = conf.get(CONF_OVERRIDE, [])
x10_devices = conf.get(CONF_X10, [])
await devices.async_load(
workdir=hass.config.config_dir, id_devices=0, load_modem_aldb=0
)
for device_override in overrides:
for device_override in entry.options.get(CONF_OVERRIDE, []):
# Override the device default capabilities for a specific address
address = device_override.get("address")
if not devices.get(address):
cat = device_override[CONF_CAT]
subcat = device_override[CONF_SUBCAT]
firmware = device_override.get(CONF_FIRMWARE)
if firmware is None:
firmware = device_override.get(CONF_PRODUCT_KEY, 0)
devices.set_id(address, cat, subcat, firmware)
devices.set_id(address, cat, subcat, 0)
for device in x10_devices:
for device in entry.options.get(CONF_X10, []):
housecode = device.get(CONF_HOUSECODE)
unitcode = device.get(CONF_UNITCODE)
x10_type = "on_off"
@ -156,5 +162,5 @@ async def async_setup(hass, config):
)
device = devices.add_x10_device(housecode, unitcode, x10_type, steps)
asyncio.create_task(async_setup_platforms(hass, config))
asyncio.create_task(async_setup_platforms(hass, entry))
return True

25
homeassistant/components/insteon/binary_sensor.py
View File

@ -26,10 +26,12 @@ from homeassistant.components.binary_sensor import (
DEVICE_CLASS_PROBLEM,
DEVICE_CLASS_SAFETY,
DEVICE_CLASS_SMOKE,
DOMAIN,
DOMAIN as BINARY_SENSOR_DOMAIN,
BinarySensorEntity,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
@ -50,11 +52,22 @@ SENSOR_TYPES = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the INSTEON entity class for the hass platform."""
async_add_insteon_entities(
hass, DOMAIN, InsteonBinarySensorEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon binary sensors from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass,
BINARY_SENSOR_DOMAIN,
InsteonBinarySensorEntity,
async_add_entities,
discovery_info,
)
signal = f"{SIGNAL_ADD_ENTITIES}_{BINARY_SENSOR_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonBinarySensorEntity(InsteonEntity, BinarySensorEntity):

25
homeassistant/components/insteon/climate.py
View File

@ -13,7 +13,7 @@ from homeassistant.components.climate.const import (
CURRENT_HVAC_FAN,
CURRENT_HVAC_HEAT,
CURRENT_HVAC_IDLE,
DOMAIN,
DOMAIN as CLIMATE_DOMAIN,
HVAC_MODE_AUTO,
HVAC_MODE_COOL,
HVAC_MODE_FAN_ONLY,
@ -26,7 +26,9 @@ from homeassistant.components.climate.const import (
SUPPORT_TARGET_TEMPERATURE_RANGE,
)
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
@ -62,11 +64,22 @@ SUPPORTED_FEATURES = (
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Insteon platform."""
async_add_insteon_entities(
hass, DOMAIN, InsteonClimateEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon climate entities from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass,
CLIMATE_DOMAIN,
InsteonClimateEntity,
async_add_entities,
discovery_info,
)
signal = f"{SIGNAL_ADD_ENTITIES}_{CLIMATE_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonClimateEntity(InsteonEntity, ClimateEntity):

317
homeassistant/components/insteon/config_flow.py Normal file
View File

@ -0,0 +1,317 @@
"""Test config flow for Insteon."""
import logging
from pyinsteon import async_connect
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
# pylint: disable=unused-import
from .const import (
CONF_HOUSECODE,
CONF_HUB_VERSION,
CONF_OVERRIDE,
CONF_UNITCODE,
CONF_X10,
DOMAIN,
SIGNAL_ADD_DEVICE_OVERRIDE,
SIGNAL_ADD_X10_DEVICE,
SIGNAL_REMOVE_DEVICE_OVERRIDE,
SIGNAL_REMOVE_X10_DEVICE,
)
from .schemas import (
add_device_override,
add_x10_device,
build_device_override_schema,
build_hub_schema,
build_plm_schema,
build_remove_override_schema,
build_remove_x10_schema,
build_x10_schema,
)
STEP_PLM = "plm"
STEP_HUB_V1 = "hubv1"
STEP_HUB_V2 = "hubv2"
STEP_CHANGE_HUB_CONFIG = "change_hub_config"
STEP_ADD_X10 = "add_x10"
STEP_ADD_OVERRIDE = "add_override"
STEP_REMOVE_OVERRIDE = "remove_override"
STEP_REMOVE_X10 = "remove_x10"
MODEM_TYPE = "modem_type"
PLM = "PowerLinc Modem (PLM)"
HUB1 = "Hub version 1 (pre-2014)"
HUB2 = "Hub version 2"
_LOGGER = logging.getLogger(__name__)
def _only_one_selected(*args):
"""Test if only one item is True."""
return sum(args) == 1
async def _async_connect(**kwargs):
"""Connect to the Insteon modem."""
try:
await async_connect(**kwargs)
_LOGGER.info("Connected to Insteon modem.")
return True
except ConnectionError:
_LOGGER.error("Could not connect to Insteon modem.")
return False
def _remove_override(address, options):
"""Remove a device override from config."""
new_options = {}
if options.get(CONF_X10):
new_options[CONF_X10] = options.get(CONF_X10)
new_overrides = []
for override in options[CONF_OVERRIDE]:
if override[CONF_ADDRESS] != address:
new_overrides.append(override)
if new_overrides:
new_options[CONF_OVERRIDE] = new_overrides
return new_options
def _remove_x10(device, options):
"""Remove an X10 device from the config."""
housecode = device[11].lower()
unitcode = int(device[24:])
new_options = {}
if options.get(CONF_OVERRIDE):
new_options[CONF_OVERRIDE] = options.get(CONF_OVERRIDE)
new_x10 = []
for existing_device in options[CONF_X10]:
if (
existing_device[CONF_HOUSECODE].lower() != housecode
or existing_device[CONF_UNITCODE] != unitcode
):
new_x10.append(existing_device)
if new_x10:
new_options[CONF_X10] = new_x10
return new_options, housecode, unitcode
class InsteonFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Insteon config flow handler."""
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Define the config flow to handle options."""
return InsteonOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
"""For backward compatibility."""
return await self.async_step_init(user_input=user_input)
async def async_step_init(self, user_input=None):
"""Init the config flow."""
errors = {}
if self._async_current_entries():
return self.async_abort(reason="already_configured")
if user_input is not None:
selection = user_input.get(MODEM_TYPE)
if selection == PLM:
return await self.async_step_plm()
if selection == HUB1:
return await self.async_step_hubv1()
return await self.async_step_hubv2()
modem_types = [PLM, HUB1, HUB2]
data_schema = vol.Schema({vol.Required(MODEM_TYPE): vol.In(modem_types)})
return self.async_show_form(
step_id="init", data_schema=data_schema, errors=errors
)
async def async_step_plm(self, user_input=None):
"""Set up the PLM modem type."""
errors = {}
if user_input is not None:
if await _async_connect(**user_input):
return self.async_create_entry(title="", data=user_input)
errors["base"] = "cannot_connect"
schema_defaults = user_input if user_input is not None else {}
data_schema = build_plm_schema(**schema_defaults)
return self.async_show_form(
step_id=STEP_PLM, data_schema=data_schema, errors=errors
)
async def async_step_hubv1(self, user_input=None):
"""Set up the Hub v1 modem type."""
return await self._async_setup_hub(hub_version=1, user_input=user_input)
async def async_step_hubv2(self, user_input=None):
"""Set up the Hub v2 modem type."""
return await self._async_setup_hub(hub_version=2, user_input=user_input)
async def _async_setup_hub(self, hub_version, user_input):
"""Set up the Hub versions 1 and 2."""
errors = {}
if user_input is not None:
user_input[CONF_HUB_VERSION] = hub_version
if await _async_connect(**user_input):
return self.async_create_entry(title="", data=user_input)
user_input.pop(CONF_HUB_VERSION)
errors["base"] = "cannot_connect"
schema_defaults = user_input if user_input is not None else {}
data_schema = build_hub_schema(hub_version=hub_version, **schema_defaults)
step_id = STEP_HUB_V2 if hub_version == 2 else STEP_HUB_V1
return self.async_show_form(
step_id=step_id, data_schema=data_schema, errors=errors
)
async def async_step_import(self, import_info):
"""Import a yaml entry as a config entry."""
if self._async_current_entries():
return self.async_abort(reason="already_configured")
if not await _async_connect(**import_info):
return self.async_abort(reason="cannot_connect")
return self.async_create_entry(title="", data=import_info)
class InsteonOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle an Insteon options flow."""
def __init__(self, config_entry):
"""Init the InsteonOptionsFlowHandler class."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Init the options config flow."""
errors = {}
if user_input is not None:
change_hub_config = user_input.get(STEP_CHANGE_HUB_CONFIG, False)
device_override = user_input.get(STEP_ADD_OVERRIDE, False)
x10_device = user_input.get(STEP_ADD_X10, False)
remove_override = user_input.get(STEP_REMOVE_OVERRIDE, False)
remove_x10 = user_input.get(STEP_REMOVE_X10, False)
if _only_one_selected(
change_hub_config,
device_override,
x10_device,
remove_override,
remove_x10,
):
if change_hub_config:
return await self.async_step_change_hub_config()
if device_override:
return await self.async_step_add_override()
if x10_device:
return await self.async_step_add_x10()
if remove_override:
return await self.async_step_remove_override()
if remove_x10:
return await self.async_step_remove_x10()
errors["base"] = "select_single"
data_schema = {
vol.Optional(STEP_ADD_OVERRIDE): bool,
vol.Optional(STEP_ADD_X10): bool,
}
if self.config_entry.data.get(CONF_HOST):
data_schema[vol.Optional(STEP_CHANGE_HUB_CONFIG)] = bool
options = {**self.config_entry.options}
if options.get(CONF_OVERRIDE):
data_schema[vol.Optional(STEP_REMOVE_OVERRIDE)] = bool
if options.get(CONF_X10):
data_schema[vol.Optional(STEP_REMOVE_X10)] = bool
return self.async_show_form(
step_id="init", data_schema=vol.Schema(data_schema), errors=errors
)
async def async_step_change_hub_config(self, user_input=None):
"""Change the Hub configuration."""
if user_input is not None:
data = {
**self.config_entry.data,
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
}
if self.config_entry.data[CONF_HUB_VERSION] == 2:
data[CONF_USERNAME] = user_input[CONF_USERNAME]
data[CONF_PASSWORD] = user_input[CONF_PASSWORD]
self.hass.config_entries.async_update_entry(self.config_entry, data=data)
return self.async_create_entry(
title="", data={**self.config_entry.options},
)
data_schema = build_hub_schema(**self.config_entry.data)
return self.async_show_form(
step_id=STEP_CHANGE_HUB_CONFIG, data_schema=data_schema
)
async def async_step_add_override(self, user_input=None):
"""Add a device override."""
errors = {}
if user_input is not None:
try:
data = add_device_override({**self.config_entry.options}, user_input)
async_dispatcher_send(self.hass, SIGNAL_ADD_DEVICE_OVERRIDE, user_input)
return self.async_create_entry(title="", data=data)
except ValueError:
errors["base"] = "input_error"
schema_defaults = user_input if user_input is not None else {}
data_schema = build_device_override_schema(**schema_defaults)
return self.async_show_form(
step_id=STEP_ADD_OVERRIDE, data_schema=data_schema, errors=errors
)
async def async_step_add_x10(self, user_input=None):
"""Add an X10 device."""
errors = {}
if user_input is not None:
options = add_x10_device({**self.config_entry.options}, user_input)
async_dispatcher_send(self.hass, SIGNAL_ADD_X10_DEVICE, user_input)
return self.async_create_entry(title="", data=options)
schema_defaults = user_input if user_input is not None else {}
data_schema = build_x10_schema(**schema_defaults)
return self.async_show_form(
step_id=STEP_ADD_X10, data_schema=data_schema, errors=errors
)
async def async_step_remove_override(self, user_input=None):
"""Remove a device override."""
errors = {}
options = self.config_entry.options
if user_input is not None:
options = _remove_override(user_input[CONF_ADDRESS], options)
async_dispatcher_send(
self.hass, SIGNAL_REMOVE_DEVICE_OVERRIDE, user_input[CONF_ADDRESS],
)
return self.async_create_entry(title="", data=options)
data_schema = build_remove_override_schema(options[CONF_OVERRIDE])
return self.async_show_form(
step_id=STEP_REMOVE_OVERRIDE, data_schema=data_schema, errors=errors
)
async def async_step_remove_x10(self, user_input=None):
"""Remove an X10 device."""
errors = {}
options = self.config_entry.options
if user_input is not None:
options, housecode, unitcode = _remove_x10(user_input[CONF_DEVICE], options)
async_dispatcher_send(
self.hass, SIGNAL_REMOVE_X10_DEVICE, housecode, unitcode
)
return self.async_create_entry(title="", data=options)
data_schema = build_remove_x10_schema(options[CONF_X10])
return self.async_show_form(
step_id=STEP_REMOVE_X10, data_schema=data_schema, errors=errors
)

16
homeassistant/components/insteon/const.py
View File

@ -43,6 +43,12 @@ INSTEON_COMPONENTS = [
"switch",
]
X10_PLATFORMS = [
"binary_sensor",
"switch",
"light",
]
CONF_IP_PORT = "ip_port"
CONF_HUB_USERNAME = "username"
CONF_HUB_PASSWORD = "password"
@ -61,6 +67,9 @@ CONF_X10_ALL_UNITS_OFF = "x10_all_units_off"
CONF_X10_ALL_LIGHTS_ON = "x10_all_lights_on"
CONF_X10_ALL_LIGHTS_OFF = "x10_all_lights_off"
PORT_HUB_V1 = 9761
PORT_HUB_V2 = 25105
SRV_ADD_ALL_LINK = "add_all_link"
SRV_DEL_ALL_LINK = "delete_all_link"
SRV_LOAD_ALDB = "load_all_link_database"
@ -82,6 +91,13 @@ SRV_ADD_DEFAULT_LINKS = "add_default_links"
SIGNAL_LOAD_ALDB = "load_aldb"
SIGNAL_PRINT_ALDB = "print_aldb"
SIGNAL_SAVE_DEVICES = "save_devices"
SIGNAL_ADD_ENTITIES = "insteon_add_entities"
SIGNAL_ADD_DEFAULT_LINKS = "add_default_links"
SIGNAL_ADD_DEVICE_OVERRIDE = "add_device_override"
SIGNAL_REMOVE_DEVICE_OVERRIDE = "insteon_remove_device_override"
SIGNAL_REMOVE_ENTITY = "insteon_remove_entity"
SIGNAL_ADD_X10_DEVICE = "insteon_add_x10_device"
SIGNAL_REMOVE_X10_DEVICE = "insteon_remove_x10_device"
SIGNAL_ADD_DEFAULT_LINKS = "add_default_links"
HOUSECODES = [

21
homeassistant/components/insteon/cover.py
View File

@ -4,13 +4,15 @@ import math
from homeassistant.components.cover import (
ATTR_POSITION,
DOMAIN,
DOMAIN as COVER_DOMAIN,
SUPPORT_CLOSE,
SUPPORT_OPEN,
SUPPORT_SET_POSITION,
CoverEntity,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
@ -19,11 +21,18 @@ _LOGGER = logging.getLogger(__name__)
SUPPORTED_FEATURES = SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_SET_POSITION
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Insteon platform."""
async_add_insteon_entities(
hass, DOMAIN, InsteonCoverEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon covers from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass, COVER_DOMAIN, InsteonCoverEntity, async_add_entities, discovery_info
)
signal = f"{SIGNAL_ADD_ENTITIES}_{COVER_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonCoverEntity(InsteonEntity, CoverEntity):

21
homeassistant/components/insteon/fan.py
View File

@ -4,7 +4,7 @@ import logging
from pyinsteon.constants import FanSpeed
from homeassistant.components.fan import (
DOMAIN,
DOMAIN as FAN_DOMAIN,
SPEED_HIGH,
SPEED_LOW,
SPEED_MEDIUM,
@ -12,7 +12,9 @@ from homeassistant.components.fan import (
SUPPORT_SET_SPEED,
FanEntity,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
@ -26,11 +28,18 @@ SPEED_TO_VALUE = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the INSTEON entity class for the hass platform."""
async_add_insteon_entities(
hass, DOMAIN, InsteonFanEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon fans from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass, FAN_DOMAIN, InsteonFanEntity, async_add_entities, discovery_info
)
signal = f"{SIGNAL_ADD_ENTITIES}_{FAN_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonFanEntity(InsteonEntity, FanEntity):

30
homeassistant/components/insteon/insteon_entity.py
View File

@ -1,6 +1,8 @@
"""Insteon base entity."""
import logging
from pyinsteon import devices
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
@ -9,9 +11,11 @@ from homeassistant.helpers.dispatcher import (
from homeassistant.helpers.entity import Entity
from .const import (
DOMAIN,
SIGNAL_ADD_DEFAULT_LINKS,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
SIGNAL_REMOVE_ENTITY,
SIGNAL_SAVE_DEVICES,
STATE_NAME_LABEL_MAP,
)
@ -74,6 +78,18 @@ class InsteonEntity(Entity):
"""Provide attributes for display on device card."""
return {"insteon_address": self.address, "insteon_group": self.group}
@property
def device_info(self):
"""Return device information."""
return {
"identifiers": {(DOMAIN, str(self._insteon_device.address))},
"name": f"{self._insteon_device.description} {self._insteon_device.address}",
"model": f"{self._insteon_device.model} (0x{self._insteon_device.cat:02x}, 0x{self._insteon_device.subcat:02x})",
"sw_version": f"{self._insteon_device.firmware:02x} Engine Version: {self._insteon_device.engine_version}",
"manufacturer": "Smart Home",
"via_device": (DOMAIN, str(devices.modem.address)),
}
@callback
def async_entity_update(self, name, address, value, group):
"""Receive notification from transport that new data exists."""
@ -101,6 +117,20 @@ class InsteonEntity(Entity):
async_dispatcher_connect(
self.hass, default_links_signal, self._async_add_default_links
)
remove_signal = f"{self._insteon_device.address.id}_{SIGNAL_REMOVE_ENTITY}"
self.async_on_remove(
async_dispatcher_connect(self.hass, remove_signal, self.async_remove)
)
async def async_will_remove_from_hass(self):
"""Unsubscribe to INSTEON update events."""
_LOGGER.debug(
"Remove tracking updates for device %s group %d name %s",
self.address,
self.group,
self._insteon_device_group.name,
)
self._insteon_device_group.unsubscribe(self.async_entity_update)
async def _async_read_aldb(self, reload):
"""Call device load process and print to log."""

21
homeassistant/components/insteon/light.py
View File

@ -3,11 +3,13 @@ import logging
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
DOMAIN,
DOMAIN as LIGHT_DOMAIN,
SUPPORT_BRIGHTNESS,
LightEntity,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
@ -16,11 +18,18 @@ _LOGGER = logging.getLogger(__name__)
MAX_BRIGHTNESS = 255
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Insteon component."""
async_add_insteon_entities(
hass, DOMAIN, InsteonDimmerEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon lights from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass, LIGHT_DOMAIN, InsteonDimmerEntity, async_add_entities, discovery_info
)
signal = f"{SIGNAL_ADD_ENTITIES}_{LIGHT_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonDimmerEntity(InsteonEntity, LightEntity):

3
homeassistant/components/insteon/manifest.json
View File

@ -3,5 +3,6 @@
"name": "Insteon",
"documentation": "https://www.home-assistant.io/integrations/insteon",
"requirements": ["pyinsteon==1.0.7"],
"codeowners": ["@teharris1"]
"codeowners": ["@teharris1"],
"config_flow": true
}

185
homeassistant/components/insteon/schemas.py
View File

@ -1,15 +1,20 @@
"""Schemas used by insteon component."""
from binascii import Error as HexError, unhexlify
from typing import Dict
from pyinsteon.address import Address
from pyinsteon.constants import HC_LOOKUP
import voluptuous as vol
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_ENTITY_ID,
CONF_HOST,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_PORT,
CONF_USERNAME,
ENTITY_MATCH_ALL,
)
import homeassistant.helpers.config_validation as cv
@ -34,12 +39,15 @@ from .const import (
CONF_X10_ALL_UNITS_OFF,
DOMAIN,
HOUSECODES,
PORT_HUB_V1,
PORT_HUB_V2,
SRV_ALL_LINK_GROUP,
SRV_ALL_LINK_MODE,
SRV_CONTROLLER,
SRV_HOUSECODE,
SRV_LOAD_DB_RELOAD,
SRV_RESPONDER,
X10_PLATFORMS,
)
@ -51,7 +59,7 @@ def set_default_port(schema: Dict) -> Dict:
if not ip_port:
hub_version = schema.get(CONF_HUB_VERSION)
# Found hub_version but not ip_port
schema[CONF_IP_PORT] = 9761 if hub_version == 1 else 25105
schema[CONF_IP_PORT] = PORT_HUB_V1 if hub_version == 1 else PORT_HUB_V2
return schema
@ -150,3 +158,176 @@ TRIGGER_SCENE_SCHEMA = vol.Schema(
ADD_DEFAULT_LINKS_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})
def normalize_byte_entry_to_int(entry: [int, bytes, str]):
"""Format a hex entry value."""
if isinstance(entry, int):
if entry in range(0, 256):
return entry
raise ValueError("Must be single byte")
if isinstance(entry, str):
if entry[0:2].lower() == "0x":
entry = entry[2:]
if len(entry) != 2:
raise ValueError("Not a valid hex code")
try:
entry = unhexlify(entry)
except HexError:
raise ValueError("Not a valid hex code")
return int.from_bytes(entry, byteorder="big")
def add_device_override(config_data, new_override):
"""Add a new device override."""
try:
address = str(Address(new_override[CONF_ADDRESS]))
cat = normalize_byte_entry_to_int(new_override[CONF_CAT])
subcat = normalize_byte_entry_to_int(new_override[CONF_SUBCAT])
except ValueError:
raise ValueError("Incorrect values")
overrides = config_data.get(CONF_OVERRIDE, [])
curr_override = {}
# If this address has an override defined, remove it
for override in overrides:
if override[CONF_ADDRESS] == address:
curr_override = override
break
if curr_override:
overrides.remove(curr_override)
curr_override[CONF_ADDRESS] = address
curr_override[CONF_CAT] = cat
curr_override[CONF_SUBCAT] = subcat
overrides.append(curr_override)
config_data[CONF_OVERRIDE] = overrides
return config_data
def add_x10_device(config_data, new_x10):
"""Add a new X10 device to X10 device list."""
curr_device = {}
x10_devices = config_data.get(CONF_X10, [])
for x10_device in x10_devices:
if (
x10_device[CONF_HOUSECODE] == new_x10[CONF_HOUSECODE]
and x10_device[CONF_UNITCODE] == new_x10[CONF_UNITCODE]
):
curr_device = x10_device
break
if curr_device:
x10_devices.remove(curr_device)
curr_device[CONF_HOUSECODE] = new_x10[CONF_HOUSECODE]
curr_device[CONF_UNITCODE] = new_x10[CONF_UNITCODE]
curr_device[CONF_PLATFORM] = new_x10[CONF_PLATFORM]
curr_device[CONF_DIM_STEPS] = new_x10[CONF_DIM_STEPS]
x10_devices.append(curr_device)
config_data[CONF_X10] = x10_devices
return config_data
def build_device_override_schema(
address=vol.UNDEFINED,
cat=vol.UNDEFINED,
subcat=vol.UNDEFINED,
firmware=vol.UNDEFINED,
):
"""Build the device override schema for config flow."""
return vol.Schema(
{
vol.Required(CONF_ADDRESS, default=address): str,
vol.Optional(CONF_CAT, default=cat): str,
vol.Optional(CONF_SUBCAT, default=subcat): str,
}
)
def build_x10_schema(
housecode=vol.UNDEFINED,
unitcode=vol.UNDEFINED,
platform=vol.UNDEFINED,
dim_steps=22,
):
"""Build the X10 schema for config flow."""
return vol.Schema(
{
vol.Required(CONF_HOUSECODE, default=housecode): vol.In(HC_LOOKUP.keys()),
vol.Required(CONF_UNITCODE, default=unitcode): vol.In(range(1, 17)),
vol.Required(CONF_PLATFORM, default=platform): vol.In(X10_PLATFORMS),
vol.Optional(CONF_DIM_STEPS, default=dim_steps): vol.In(range(1, 255)),
}
)
def build_plm_schema(device=vol.UNDEFINED):
"""Build the PLM schema for config flow."""
return vol.Schema({vol.Required(CONF_DEVICE, default=device): str})
def build_hub_schema(
hub_version,
host=vol.UNDEFINED,
port=PORT_HUB_V2,
username=vol.UNDEFINED,
password=vol.UNDEFINED,
):
"""Build the Hub v2 schema for config flow."""
schema = {
vol.Required(CONF_HOST, default=host): str,
vol.Required(CONF_PORT, default=port): int,
}
if hub_version == 2:
schema[vol.Required(CONF_USERNAME, default=username)] = str
schema[vol.Required(CONF_PASSWORD, default=password)] = str
return vol.Schema(schema)
def build_remove_override_schema(data):
"""Build the schema to remove device overrides in config flow options."""
selection = []
for override in data:
selection.append(override[CONF_ADDRESS])
return vol.Schema({vol.Required(CONF_ADDRESS): vol.In(selection)})
def build_remove_x10_schema(data):
"""Build the schema to remove an X10 device in config flow options."""
selection = []
for device in data:
housecode = device[CONF_HOUSECODE].upper()
unitcode = device[CONF_UNITCODE]
selection.append(f"Housecode: {housecode}, Unitcode: {unitcode}")
return vol.Schema({vol.Required(CONF_DEVICE): vol.In(selection)})
def convert_yaml_to_config_flow(yaml_config):
"""Convert the YAML based configuration to a config flow configuration."""
config = {}
if yaml_config.get(CONF_HOST):
hub_version = yaml_config.get(CONF_HUB_VERSION, 2)
default_port = PORT_HUB_V2 if hub_version == 2 else PORT_HUB_V1
config[CONF_HOST] = yaml_config.get(CONF_HOST)
config[CONF_PORT] = yaml_config.get(CONF_PORT, default_port)
config[CONF_HUB_VERSION] = hub_version
if hub_version == 2:
config[CONF_USERNAME] = yaml_config[CONF_USERNAME]
config[CONF_PASSWORD] = yaml_config[CONF_PASSWORD]
else:
config[CONF_DEVICE] = yaml_config[CONF_PORT]
options = {}
for old_override in yaml_config.get(CONF_OVERRIDE, []):
override = {}
override[CONF_ADDRESS] = str(Address(old_override[CONF_ADDRESS]))
override[CONF_CAT] = normalize_byte_entry_to_int(old_override[CONF_CAT])
override[CONF_SUBCAT] = normalize_byte_entry_to_int(old_override[CONF_SUBCAT])
options = add_device_override(options, override)
for x10_device in yaml_config.get(CONF_X10, []):
options = add_x10_device(options, x10_device)
return config, options

115
homeassistant/components/insteon/strings.json Normal file
View File

@ -0,0 +1,115 @@
{
"config": {
"step": {
"init": {
"title": "Insteon",
"description": "Select the Insteon modem type.",
"data": {
"plm": "PowerLink Modem (PLM)",
"hubv1": "Hub Version 1 (Pre-2014)",
"hubv2": "Hub Version 2"
}
},
"plm": {
"title": "Insteon PLM",
"description": "Configure the Insteon PowerLink Modem (PLM).",
"data": {
"device": "PLM device (i.e. /dev/ttyUSB0 or COM3)"
}
},
"hub1": {
"title": "Insteon Hub Version 1",
"description": "Configure the Insteon Hub Version 1 (pre-2014).",
"data": {
"host": "Hub IP address",
"port": "IP port"
}
},
"hub2": {
"title": "Insteon Hub Version 2",
"description": "Configure the Insteon Hub Version 2.",
"data": {
"host": "Hub IP address",
"username": "Username",
"password": "Password",
"port": "IP port"
}
}
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"select_single": "Select one option."
},
"abort": {
"cannot_connect": "Unable to connect to the Insteon modem",
"already_configured": "An Insteon modem connection is already configured"
}
},
"options": {
"step": {
"init": {
"title": "Insteon",
"description": "Select an option to configure.",
"data": {
"change_hub_config": "Change the Hub configuration.",
"add_override": "Add a device override.",
"add_x10": "Add an X10 device.",
"remove_override": "Remove a device override.",
"remove_x10": "Remove an X10 device."
}
},
"change_hub_config": {
"title": "Insteon",
"description": "Change the Insteon Hub connection information. You must restart Home Assistant after making this change. This does not change the configuration of the Hub itself. To change the configuration in the Hub use the Hub app.",
"data": {
"host": "New host name or IP address",
"username": "New username",
"password": "New password",
"port": "New port number"
}
},
"add_override": {
"title": "Insteon",
"description": "Add a device override.",
"data": {
"address": "Device address (i.e. 1a2b3c)",
"cat": "Device category (i.e. 0x10)",
"subcat": "Device subcategory (i.e. 0x0a)"
}
},
"add_x10": {
"title": "Insteon",
"description": "Change the Insteon Hub password.",
"data": {
"housecode": "Housecode (a - p)",
"unitcode": "Unitcode (1 - 16)",
"platform": "Platform",
"steps": "Dimmer steps (for light devices only, default 22)"
}
},
"remove_override": {
"title": "Insteon",
"description": "Remove a device override",
"data": {
"address": "Select a device address to remove"
}
},
"remove_x10": {
"title": "Insteon",
"description": "Remove an X10 device",
"data": {
"address": "Select a device address to remove"
}
}
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"select_single": "Select one option.",
"input_error": "Invalid entries, please check your values."
},
"abort": {
"cannot_connect": "Unable to connect to the Insteon modem",
"already_configured": "An Insteon modem connection is already configured"
}
}
}

21
homeassistant/components/insteon/switch.py
View File

@ -1,19 +1,28 @@
"""Support for INSTEON dimmers via PowerLinc Modem."""
import logging
from homeassistant.components.switch import DOMAIN, SwitchEntity
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN, SwitchEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import SIGNAL_ADD_ENTITIES
from .insteon_entity import InsteonEntity
from .utils import async_add_insteon_entities
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the INSTEON entity class for the hass platform."""
async_add_insteon_entities(
hass, DOMAIN, InsteonSwitchEntity, async_add_entities, discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Insteon switches from a config entry."""
def add_entities(discovery_info=None):
"""Add the Insteon entities for the platform."""
async_add_insteon_entities(
hass, SWITCH_DOMAIN, InsteonSwitchEntity, async_add_entities, discovery_info
)
signal = f"{SIGNAL_ADD_ENTITIES}_{SWITCH_DOMAIN}"
async_dispatcher_connect(hass, signal, add_entities)
add_entities()
class InsteonSwitchEntity(InsteonEntity, SwitchEntity):

115
homeassistant/components/insteon/translations/en.json Normal file
View File

@ -0,0 +1,115 @@
{
"config": {
"abort": {
"already_configured": "An Insteon modem connection is already configured",
"cannot_connect": "Unable to connect to the Insteon modem"
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"select_single": "Select one option."
},
"step": {
"hub1": {
"data": {
"host": "Hub IP address",
"port": "IP port"
},
"description": "Configure the Insteon Hub Version 1 (pre-2014).",
"title": "Insteon Hub Version 1"
},
"hub2": {
"data": {
"host": "Hub IP address",
"password": "Password",
"port": "IP port",
"username": "Username"
},
"description": "Configure the Insteon Hub Version 2.",
"title": "Insteon Hub Version 2"
},
"init": {
"data": {
"hubv1": "Hub Version 1 (Pre-2014)",
"hubv2": "Hub Version 2",
"plm": "PowerLink Modem (PLM)"
},
"description": "Select the Insteon modem type.",
"title": "Insteon"
},
"plm": {
"data": {
"device": "PLM device (i.e. /dev/ttyUSB0 or COM3)"
},
"description": "Configure the Insteon PowerLink Modem (PLM).",
"title": "Insteon PLM"
}
}
},
"options": {
"abort": {
"already_configured": "An Insteon modem connection is already configured",
"cannot_connect": "Unable to connect to the Insteon modem"
},
"error": {
"cannot_connect": "Failed to connect to the Insteon modem, please try again.",
"input_error": "Invalid entries, please check your values.",
"select_single": "Select one option."
},
"step": {
"add_override": {
"data": {
"address": "Device address (i.e. 1a2b3c)",
"cat": "Device category (i.e. 0x10)",
"subcat": "Device subcategory (i.e. 0x0a)"
},
"description": "Add a device override.",
"title": "Insteon"
},
"add_x10": {
"data": {
"housecode": "Housecode (a - p)",
"platform": "Platform",
"steps": "Dimmer steps (for light devices only, default 22)",
"unitcode": "Unitcode (1 - 16)"
},
"description": "Change the Insteon Hub password.",
"title": "Insteon"
},
"change_hub_config": {
"data": {
"host": "New host name or IP address",
"password": "New password",
"port": "New port number",
"username": "New username"
},
"description": "Change the Insteon Hub connection information. You must restart Home Assistant after making this change. This does not change the configuration of the Hub itself. To change the configuration in the Hub use the Hub app.",
"title": "Insteon"
},
"init": {
"data": {
"add_override": "Add a device override.",
"add_x10": "Add an X10 device.",
"change_hub_config": "Change the Hub configuration.",
"remove_override": "Remove a device override.",
"remove_x10": "Remove an X10 device."
},
"description": "Select an option to configure.",
"title": "Insteon"
},
"remove_override": {
"data": {
"address": "Select a device address to remove"
},
"description": "Remove a device override",
"title": "Insteon"
},
"remove_x10": {
"data": {
"address": "Select a device address to remove"
},
"description": "Remove an X10 device",
"title": "Insteon"
}
}
}
}

105
homeassistant/components/insteon/utils.py
View File

@ -3,6 +3,7 @@ import asyncio
import logging
from pyinsteon import devices
from pyinsteon.address import Address
from pyinsteon.constants import ALDBStatus
from pyinsteon.events import OFF_EVENT, OFF_FAST_EVENT, ON_EVENT, ON_FAST_EVENT
from pyinsteon.managers.link_manager import (
@ -18,10 +19,15 @@ from pyinsteon.managers.x10_manager import (
async_x10_all_lights_on,
async_x10_all_units_off,
)
from pyinsteon.x10_address import create as create_x10_address
from homeassistant.const import CONF_ADDRESS, CONF_ENTITY_ID, ENTITY_MATCH_ALL
from homeassistant.const import (
CONF_ADDRESS,
CONF_ENTITY_ID,
CONF_PLATFORM,
ENTITY_MATCH_ALL,
)
from homeassistant.core import callback
from homeassistant.helpers import discovery
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
@ -29,6 +35,11 @@ from homeassistant.helpers.dispatcher import (
)
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_SUBCAT,
CONF_UNITCODE,
DOMAIN,
EVENT_CONF_BUTTON,
EVENT_GROUP_OFF,
@ -37,8 +48,14 @@ from .const import (
EVENT_GROUP_ON_FAST,
ON_OFF_EVENTS,
SIGNAL_ADD_DEFAULT_LINKS,
SIGNAL_ADD_DEVICE_OVERRIDE,
SIGNAL_ADD_ENTITIES,
SIGNAL_ADD_X10_DEVICE,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
SIGNAL_REMOVE_DEVICE_OVERRIDE,
SIGNAL_REMOVE_ENTITY,
SIGNAL_REMOVE_X10_DEVICE,
SIGNAL_SAVE_DEVICES,
SRV_ADD_ALL_LINK,
SRV_ADD_DEFAULT_LINKS,
@ -116,9 +133,8 @@ def add_on_off_event_device(hass, device):
)
def register_new_device_callback(hass, config):
def register_new_device_callback(hass):
"""Register callback for new Insteon device."""
new_device_lock = asyncio.Lock()
@callback
def async_new_insteon_device(address=None):
@ -129,27 +145,17 @@ def register_new_device_callback(hass, config):
_LOGGER.debug(
"Adding new INSTEON device to Home Assistant with address %s", address
)
async with new_device_lock:
await devices.async_save(workdir=hass.config.config_dir)
await devices.async_save(workdir=hass.config.config_dir)
device = devices[address]
await device.async_status()
platforms = get_device_platforms(device)
tasks = []
for platform in platforms:
if platform == ON_OFF_EVENTS:
add_on_off_event_device(hass, device)
else:
tasks.append(
discovery.async_load_platform(
hass,
platform,
DOMAIN,
discovered={"address": device.address.id},
hass_config=config,
)
)
await asyncio.gather(*tasks)
signal = f"{SIGNAL_ADD_ENTITIES}_{platform}"
dispatcher_send(hass, signal, {"address": device.address})
devices.subscribe(async_new_insteon_device, force_strong_ref=True)
@ -158,6 +164,8 @@ def register_new_device_callback(hass, config):
def async_register_services(hass):
"""Register services used by insteon component."""
save_lock = asyncio.Lock()
async def async_srv_add_all_link(service):
"""Add an INSTEON All-Link between two devices."""
group = service.data.get(SRV_ALL_LINK_GROUP)
@ -192,8 +200,9 @@ def async_register_services(hass):
async def async_srv_save_devices():
"""Write the Insteon device configuration to file."""
_LOGGER.debug("Saving Insteon devices")
await devices.async_save(hass.config.config_dir)
async with save_lock:
_LOGGER.debug("Saving Insteon devices")
await devices.async_save(hass.config.config_dir)
def print_aldb(service):
"""Print the All-Link Database for a device."""
@ -241,6 +250,56 @@ def async_register_services(hass):
signal = f"{entity_id}_{SIGNAL_ADD_DEFAULT_LINKS}"
async_dispatcher_send(hass, signal)
async def async_add_device_override(override):
"""Remove an Insten device and associated entities."""
address = Address(override[CONF_ADDRESS])
await async_remove_device(address)
devices.set_id(address, override[CONF_CAT], override[CONF_SUBCAT], 0)
await async_srv_save_devices()
async def async_remove_device_override(address):
"""Remove an Insten device and associated entities."""
address = Address(address)
await async_remove_device(address)
devices.set_id(address, None, None, None)
await devices.async_identify_device(address)
await async_srv_save_devices()
@callback
def async_add_x10_device(x10_config):
"""Add X10 device."""
housecode = x10_config[CONF_HOUSECODE]
unitcode = x10_config[CONF_UNITCODE]
platform = x10_config[CONF_PLATFORM]
steps = x10_config.get(CONF_DIM_STEPS, 22)
x10_type = "on_off"
if platform == "light":
x10_type = "dimmable"
elif platform == "binary_sensor":
x10_type = "sensor"
_LOGGER.debug(
"Adding X10 device to Insteon: %s %d %s", housecode, unitcode, x10_type
)
# This must be run in the event loop
devices.add_x10_device(housecode, unitcode, x10_type, steps)
async def async_remove_x10_device(housecode, unitcode):
"""Remove an X10 device and associated entities."""
address = create_x10_address(housecode, unitcode)
devices.pop(address)
await async_remove_device(address)
async def async_remove_device(address):
"""Remove the device and all entities from hass."""
signal = f"{address.id}_{SIGNAL_REMOVE_ENTITY}"
async_dispatcher_send(hass, signal)
dev_registry = await hass.helpers.device_registry.async_get_registry()
device = dev_registry.async_get_device(
identifiers={(DOMAIN, str(address))}, connections=set()
)
if device:
dev_registry.async_remove_device(device.id)
hass.services.async_register(
DOMAIN, SRV_ADD_ALL_LINK, async_srv_add_all_link, schema=ADD_ALL_LINK_SCHEMA
)
@ -286,6 +345,14 @@ def async_register_services(hass):
schema=ADD_DEFAULT_LINKS_SCHEMA,
)
async_dispatcher_connect(hass, SIGNAL_SAVE_DEVICES, async_srv_save_devices)
async_dispatcher_connect(
hass, SIGNAL_ADD_DEVICE_OVERRIDE, async_add_device_override
)
async_dispatcher_connect(
hass, SIGNAL_REMOVE_DEVICE_OVERRIDE, async_remove_device_override
)
async_dispatcher_connect(hass, SIGNAL_ADD_X10_DEVICE, async_add_x10_device)
async_dispatcher_connect(hass, SIGNAL_REMOVE_X10_DEVICE, async_remove_x10_device)
_LOGGER.debug("Insteon Services registered")

1
homeassistant/generated/config_flows.py
View File

@ -85,6 +85,7 @@ FLOWS = [
"iaqualink",
"icloud",
"ifttt",
"insteon",
"ios",
"ipma",
"ipp",

3
requirements_test_all.txt
View File

@ -664,6 +664,9 @@ pyhomematic==0.1.68
# homeassistant.components.icloud
pyicloud==0.9.7
# homeassistant.components.insteon
pyinsteon==1.0.7
# homeassistant.components.ipma
pyipma==2.0.5

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

@ -0,0 +1 @@
"""Test for the Insteon integration."""

703
tests/components/insteon/test_config_flow.py Normal file
View File

@ -0,0 +1,703 @@
"""Test the config flow for the Insteon integration."""
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.insteon import async_import_config
from homeassistant.components.insteon.config_flow import (
HUB1,
HUB2,
MODEM_TYPE,
PLM,
STEP_ADD_OVERRIDE,
STEP_ADD_X10,
STEP_CHANGE_HUB_CONFIG,
STEP_HUB_V2,
STEP_REMOVE_OVERRIDE,
STEP_REMOVE_X10,
)
from homeassistant.components.insteon.const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_HUB_VERSION,
CONF_OVERRIDE,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
DOMAIN,
X10_PLATFORMS,
)
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_HOST,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.helpers.typing import HomeAssistantType
from tests.async_mock import patch
from tests.common import MockConfigEntry
MOCK_HOSTNAME = "1.1.1.1"
MOCK_DEVICE = "/dev/ttyUSB55"
MOCK_USERNAME = "test-username"
MOCK_PASSWORD = "test-password"
MOCK_PORT = 4567
MOCK_ADDRESS = "1a2b3c"
MOCK_CAT = 0x02
MOCK_SUBCAT = 0x1A
MOCK_HOUSECODE = "c"
MOCK_UNITCODE = 6
MOCK_X10_PLATFORM = X10_PLATFORMS[2]
MOCK_X10_STEPS = 10
MOCK_USER_INPUT_PLM = {
CONF_DEVICE: MOCK_DEVICE,
}
MOCK_USER_INPUT_HUB_V2 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
CONF_PORT: MOCK_PORT,
}
MOCK_USER_INPUT_HUB_V1 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_PORT: MOCK_PORT,
}
MOCK_DEVICE_OVERRIDE_CONFIG = {
CONF_ADDRESS: MOCK_ADDRESS,
CONF_CAT: MOCK_CAT,
CONF_SUBCAT: MOCK_SUBCAT,
}
MOCK_X10_CONFIG = {
CONF_HOUSECODE: MOCK_HOUSECODE,
CONF_UNITCODE: MOCK_UNITCODE,
CONF_PLATFORM: MOCK_X10_PLATFORM,
CONF_DIM_STEPS: MOCK_X10_STEPS,
}
MOCK_IMPORT_CONFIG_PLM = {CONF_PORT: MOCK_DEVICE}
MOCK_IMPORT_MINIMUM_HUB_V2 = {
CONF_HOST: MOCK_HOSTNAME,
CONF_USERNAME: MOCK_USERNAME,
CONF_PASSWORD: MOCK_PASSWORD,
}
MOCK_IMPORT_MINIMUM_HUB_V1 = {CONF_HOST: MOCK_HOSTNAME, CONF_HUB_VERSION: 1}
MOCK_IMPORT_FULL_CONFIG_PLM = MOCK_IMPORT_CONFIG_PLM.copy()
MOCK_IMPORT_FULL_CONFIG_PLM[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_PLM[CONF_X10] = [MOCK_X10_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V2 = MOCK_USER_INPUT_HUB_V2.copy()
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_HUB_VERSION] = 2
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V2[CONF_X10] = [MOCK_X10_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V1 = MOCK_USER_INPUT_HUB_V1.copy()
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_HUB_VERSION] = 1
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_OVERRIDE] = [MOCK_DEVICE_OVERRIDE_CONFIG]
MOCK_IMPORT_FULL_CONFIG_HUB_V1[CONF_X10] = [MOCK_X10_CONFIG]
PATCH_CONNECTION = "homeassistant.components.insteon.config_flow.async_connect"
PATCH_ASYNC_SETUP = "homeassistant.components.insteon.async_setup"
PATCH_ASYNC_SETUP_ENTRY = "homeassistant.components.insteon.async_setup_entry"
async def mock_successful_connection(*args, **kwargs):
"""Return a successful connection."""
return True
async def mock_failed_connection(*args, **kwargs):
"""Return a failed connection."""
raise ConnectionError("Connection failed")
async def _init_form(hass, modem_type):
"""Run the init form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {MODEM_TYPE: modem_type},
)
return result2
async def _device_form(hass, flow_id, connection, user_input):
"""Test the PLM, Hub v1 or Hub v2 form."""
with patch(PATCH_CONNECTION, new=connection,), patch(
PATCH_ASYNC_SETUP, return_value=True
) as mock_setup, patch(
PATCH_ASYNC_SETUP_ENTRY, return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(flow_id, user_input)
return result, mock_setup, mock_setup_entry
async def test_form_select_modem(hass: HomeAssistantType):
"""Test we get a modem form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, HUB2)
assert result["step_id"] == STEP_HUB_V2
assert result["type"] == "form"
async def test_fail_on_existing(hass: HomeAssistantType):
"""Test we fail if the integration is already configured."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
assert config_entry.state == config_entries.ENTRY_STATE_NOT_LOADED
result = await hass.config_entries.flow.async_init(
DOMAIN,
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_form_select_plm(hass: HomeAssistantType):
"""Test we set up the PLM correctly."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, PLM)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_PLM
)
assert result2["type"] == "create_entry"
assert result2["data"] == MOCK_USER_INPUT_PLM
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_select_hub_v1(hass: HomeAssistantType):
"""Test we set up the Hub v1 correctly."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, HUB1)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V1
)
assert result2["type"] == "create_entry"
assert result2["data"] == {
**MOCK_USER_INPUT_HUB_V1,
CONF_HUB_VERSION: 1,
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_select_hub_v2(hass: HomeAssistantType):
"""Test we set up the Hub v2 correctly."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, HUB2)
result2, mock_setup, mock_setup_entry = await _device_form(
hass, result["flow_id"], mock_successful_connection, MOCK_USER_INPUT_HUB_V2
)
assert result2["type"] == "create_entry"
assert result2["data"] == {
**MOCK_USER_INPUT_HUB_V2,
CONF_HUB_VERSION: 2,
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_failed_connection_plm(hass: HomeAssistantType):
"""Test a failed connection with the PLM."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, PLM)
result2, _, _ = await _device_form(
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_PLM
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_failed_connection_hub(hass: HomeAssistantType):
"""Test a failed connection with a Hub."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _init_form(hass, HUB2)
result2, _, _ = await _device_form(
hass, result["flow_id"], mock_failed_connection, MOCK_USER_INPUT_HUB_V2
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def _import_config(hass, config):
"""Run the import step."""
with patch(PATCH_CONNECTION, new=mock_successful_connection,), patch(
PATCH_ASYNC_SETUP, return_value=True
), patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
return await async_import_config(hass, config)
async def test_import_plm(hass: HomeAssistantType):
"""Test importing a minimum PLM config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_CONFIG_PLM)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data == MOCK_USER_INPUT_PLM
async def test_import_plm_full(hass: HomeAssistantType):
"""Test importing a full PLM config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_CONFIG_PLM)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data == MOCK_USER_INPUT_PLM
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_OVERRIDE][0][CONF_CAT] == MOCK_CAT
assert entry.options[CONF_OVERRIDE][0][CONF_SUBCAT] == MOCK_SUBCAT
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
assert entry.options[CONF_X10][0][CONF_UNITCODE] == MOCK_UNITCODE
assert entry.options[CONF_X10][0][CONF_PLATFORM] == MOCK_X10_PLATFORM
assert entry.options[CONF_X10][0][CONF_DIM_STEPS] == MOCK_X10_STEPS
async def test_import_full_hub_v1(hass: HomeAssistantType):
"""Test importing a full Hub v1 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_CONFIG_HUB_V1)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == MOCK_PORT
assert entry.data[CONF_HUB_VERSION] == 1
assert CONF_USERNAME not in entry.data
assert CONF_PASSWORD not in entry.data
assert CONF_OVERRIDE not in entry.data
assert CONF_X10 not in entry.data
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
async def test_import_full_hub_v2(hass: HomeAssistantType):
"""Test importing a full Hub v2 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_FULL_CONFIG_HUB_V2)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == MOCK_PORT
assert entry.data[CONF_USERNAME] == MOCK_USERNAME
assert entry.data[CONF_PASSWORD] == MOCK_PASSWORD
assert entry.data[CONF_HUB_VERSION] == 2
assert CONF_OVERRIDE not in entry.data
assert CONF_X10 not in entry.data
assert entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert entry.options[CONF_X10][0][CONF_HOUSECODE] == MOCK_HOUSECODE
async def test_import_min_hub_v2(hass: HomeAssistantType):
"""Test importing a minimum Hub v2 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == 25105
assert entry.data[CONF_USERNAME] == MOCK_USERNAME
assert entry.data[CONF_PASSWORD] == MOCK_PASSWORD
assert entry.data[CONF_HUB_VERSION] == 2
async def test_import_min_hub_v1(hass: HomeAssistantType):
"""Test importing a minimum Hub v1 config from yaml."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V1)
assert result["type"] == "create_entry"
assert hass.config_entries.async_entries(DOMAIN)
for entry in hass.config_entries.async_entries(DOMAIN):
assert entry.data[CONF_HOST] == MOCK_HOSTNAME
assert entry.data[CONF_PORT] == 9761
assert entry.data[CONF_HUB_VERSION] == 1
async def test_import_existing(hass: HomeAssistantType):
"""Test we fail on an existing config imported."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
assert config_entry.state == config_entries.ENTRY_STATE_NOT_LOADED
result = await _import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_import_failed_connection(hass: HomeAssistantType):
"""Test a failed connection on import."""
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(PATCH_CONNECTION, new=mock_failed_connection,), patch(
PATCH_ASYNC_SETUP, return_value=True
), patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
result = await async_import_config(hass, MOCK_IMPORT_MINIMUM_HUB_V2)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def _options_init_form(hass, entry_id, step):
"""Run the init options form."""
with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True):
result = await hass.config_entries.options.async_init(entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"], {step: True},
)
return result2
async def _options_form(hass, flow_id, user_input):
"""Test an options form."""
with patch(PATCH_ASYNC_SETUP_ENTRY, return_value=True) as mock_setup_entry:
result = await hass.config_entries.options.async_configure(flow_id, user_input)
return result, mock_setup_entry
async def test_options_change_hub_config(hass: HomeAssistantType):
"""Test changing Hub v2 config."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(
hass, config_entry.entry_id, STEP_CHANGE_HUB_CONFIG
)
user_input = {
CONF_HOST: "2.3.4.5",
CONF_PORT: 9999,
CONF_USERNAME: "new username",
CONF_PASSWORD: "new password",
}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {}
assert config_entry.data == {**user_input, CONF_HUB_VERSION: 2}
async def test_options_add_device_override(hass: HomeAssistantType):
"""Test adding a device override."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_OVERRIDE)
user_input = {
CONF_ADDRESS: "1a2b3c",
CONF_CAT: "0x04",
CONF_SUBCAT: "0xaa",
}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_OVERRIDE]) == 1
assert config_entry.options[CONF_OVERRIDE][0][CONF_ADDRESS] == "1A.2B.3C"
assert config_entry.options[CONF_OVERRIDE][0][CONF_CAT] == 4
assert config_entry.options[CONF_OVERRIDE][0][CONF_SUBCAT] == 170
result2 = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_OVERRIDE)
user_input = {
CONF_ADDRESS: "4d5e6f",
CONF_CAT: "05",
CONF_SUBCAT: "bb",
}
await _options_form(hass, result2["flow_id"], user_input)
assert len(config_entry.options[CONF_OVERRIDE]) == 2
assert config_entry.options[CONF_OVERRIDE][1][CONF_ADDRESS] == "4D.5E.6F"
assert config_entry.options[CONF_OVERRIDE][1][CONF_CAT] == 5
assert config_entry.options[CONF_OVERRIDE][1][CONF_SUBCAT] == 187
async def test_options_remove_device_override(hass: HomeAssistantType):
"""Test removing a device override."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={
CONF_OVERRIDE: [
{CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 6, CONF_SUBCAT: 100},
{CONF_ADDRESS: "4D.5E.6F", CONF_CAT: 7, CONF_SUBCAT: 200},
]
},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_OVERRIDE)
user_input = {CONF_ADDRESS: "1A.2B.3C"}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_OVERRIDE]) == 1
async def test_options_remove_device_override_with_x10(hass: HomeAssistantType):
"""Test removing a device override when an X10 device is configured."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={
CONF_OVERRIDE: [
{CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 6, CONF_SUBCAT: 100},
{CONF_ADDRESS: "4D.5E.6F", CONF_CAT: 7, CONF_SUBCAT: 200},
],
CONF_X10: [
{
CONF_HOUSECODE: "d",
CONF_UNITCODE: 5,
CONF_PLATFORM: "light",
CONF_DIM_STEPS: 22,
}
],
},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_OVERRIDE)
user_input = {CONF_ADDRESS: "1A.2B.3C"}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_OVERRIDE]) == 1
assert len(config_entry.options[CONF_X10]) == 1
async def test_options_add_x10_device(hass: HomeAssistantType):
"""Test adding an X10 device."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_X10)
user_input = {
CONF_HOUSECODE: "c",
CONF_UNITCODE: 12,
CONF_PLATFORM: "light",
CONF_DIM_STEPS: 18,
}
result2, _ = await _options_form(hass, result["flow_id"], user_input)
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_X10]) == 1
assert config_entry.options[CONF_X10][0][CONF_HOUSECODE] == "c"
assert config_entry.options[CONF_X10][0][CONF_UNITCODE] == 12
assert config_entry.options[CONF_X10][0][CONF_PLATFORM] == "light"
assert config_entry.options[CONF_X10][0][CONF_DIM_STEPS] == 18
result = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_X10)
user_input = {
CONF_HOUSECODE: "d",
CONF_UNITCODE: 10,
CONF_PLATFORM: "binary_sensor",
CONF_DIM_STEPS: 15,
}
result3, _ = await _options_form(hass, result["flow_id"], user_input)
assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_X10]) == 2
assert config_entry.options[CONF_X10][1][CONF_HOUSECODE] == "d"
assert config_entry.options[CONF_X10][1][CONF_UNITCODE] == 10
assert config_entry.options[CONF_X10][1][CONF_PLATFORM] == "binary_sensor"
assert config_entry.options[CONF_X10][1][CONF_DIM_STEPS] == 15
async def test_options_remove_x10_device(hass: HomeAssistantType):
"""Test removing an X10 device."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={
CONF_X10: [
{
CONF_HOUSECODE: "C",
CONF_UNITCODE: 4,
CONF_PLATFORM: "light",
CONF_DIM_STEPS: 18,
},
{
CONF_HOUSECODE: "D",
CONF_UNITCODE: 10,
CONF_PLATFORM: "binary_sensor",
CONF_DIM_STEPS: 15,
},
]
},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_X10)
for device in config_entry.options[CONF_X10]:
housecode = device[CONF_HOUSECODE].upper()
unitcode = device[CONF_UNITCODE]
print(f"Housecode: {housecode}, Unitcode: {unitcode}")
user_input = {CONF_DEVICE: "Housecode: C, Unitcode: 4"}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_X10]) == 1
async def test_options_remove_x10_device_with_override(hass: HomeAssistantType):
"""Test removing an X10 device when a device override is configured."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={
CONF_X10: [
{
CONF_HOUSECODE: "C",
CONF_UNITCODE: 4,
CONF_PLATFORM: "light",
CONF_DIM_STEPS: 18,
},
{
CONF_HOUSECODE: "D",
CONF_UNITCODE: 10,
CONF_PLATFORM: "binary_sensor",
CONF_DIM_STEPS: 15,
},
],
CONF_OVERRIDE: [{CONF_ADDRESS: "1A.2B.3C", CONF_CAT: 1, CONF_SUBCAT: 18}],
},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_REMOVE_X10)
for device in config_entry.options[CONF_X10]:
housecode = device[CONF_HOUSECODE].upper()
unitcode = device[CONF_UNITCODE]
print(f"Housecode: {housecode}, Unitcode: {unitcode}")
user_input = {CONF_DEVICE: "Housecode: C, Unitcode: 4"}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(config_entry.options[CONF_X10]) == 1
assert len(config_entry.options[CONF_OVERRIDE]) == 1
async def test_options_dup_selection(hass: HomeAssistantType):
"""Test if a duplicate selection was made in options."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result["flow_id"], {STEP_ADD_OVERRIDE: True, STEP_ADD_X10: True},
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["errors"] == {"base": "select_single"}
async def test_options_override_bad_data(hass: HomeAssistantType):
"""Test for bad data in a device override."""
config_entry = MockConfigEntry(
domain=DOMAIN,
entry_id="abcde12345",
data={**MOCK_USER_INPUT_HUB_V2, CONF_HUB_VERSION: 2},
options={},
)
config_entry.add_to_hass(hass)
result = await _options_init_form(hass, config_entry.entry_id, STEP_ADD_OVERRIDE)
user_input = {
CONF_ADDRESS: "zzzzzz",
CONF_CAT: "bad",
CONF_SUBCAT: "data",
}
result, _ = await _options_form(hass, result["flow_id"], user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "input_error"}