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

Insteon Device Control Panel (#70834) 

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
pull/71174/head
Tom Harris 2022-04-28 15:35:43 -04:00 committed by Paulus Schoutsen
parent 6d91797366
commit af0d61fb8d
19 changed files with 650 additions and 519 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,11 +9,13 @@ from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_PLATFORM, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import ConfigType
from . import api
from .const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_DIM_STEPS,
CONF_HOUSECODE,
CONF_OVERRIDE,
@ -74,13 +76,19 @@ async def close_insteon_connection(*args):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Insteon platform."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
conf = config[DOMAIN]
conf = dict(config[DOMAIN])
hass.data[DOMAIN][CONF_DEV_PATH] = conf.pop(CONF_DEV_PATH, None)
if not conf:
return True
data, options = convert_yaml_to_config_flow(conf)
if options:
hass.data[DOMAIN] = {}
hass.data[DOMAIN][OPTIONS] = options
# Create a config entry with the connection data
hass.async_create_task(
@ -154,23 +162,30 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
platforms = get_device_platforms(device)
if ON_OFF_EVENTS in platforms:
add_on_off_event_device(hass, device)
create_insteon_device(hass, device, entry.entry_id)
_LOGGER.debug("Insteon device count: %s", len(devices))
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=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} ({devices.modem.cat!r}, 0x{devices.modem.subcat:02x})",
sw_version=f"{devices.modem.firmware:02x} Engine Version: {devices.modem.engine_version}",
)
create_insteon_device(hass, devices.modem, entry.entry_id)
api.async_load_api(hass)
await api.async_register_insteon_frontend(hass)
asyncio.create_task(async_get_device_config(hass, entry))
return True
def create_insteon_device(hass, device, config_entry_id):
"""Create an Insteon device."""
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=config_entry_id, # entry.entry_id,
identifiers={(DOMAIN, str(device.address))},
manufacturer="SmartLabs, Inc",
name=f"{device.description} {device.address}",
model=f"{device.model} ({device.cat!r}, 0x{device.subcat:02x})",
sw_version=f"{device.firmware:02x} Engine Version: {device.engine_version}",
)

45
homeassistant/components/insteon/api/__init__.py
View File

@ -1,7 +1,10 @@
"""Insteon API interface for the frontend."""
from homeassistant.components import websocket_api
from homeassistant.core import callback
from insteon_frontend import get_build_id, locate_dir
from homeassistant.components import panel_custom, websocket_api
from homeassistant.components.insteon.const import CONF_DEV_PATH, DOMAIN
from homeassistant.core import HomeAssistant, callback
from .aldb import (
websocket_add_default_links,
@ -13,7 +16,11 @@ from .aldb import (
websocket_reset_aldb,
websocket_write_aldb,
)
from .device import websocket_get_device
from .device import (
websocket_add_device,
websocket_cancel_add_device,
websocket_get_device,
)
from .properties import (
websocket_change_properties_record,
websocket_get_properties,
@ -22,11 +29,15 @@ from .properties import (
websocket_write_properties,
)
URL_BASE = "/insteon_static"
@callback
def async_load_api(hass):
"""Set up the web socket API."""
websocket_api.async_register_command(hass, websocket_get_device)
websocket_api.async_register_command(hass, websocket_add_device)
websocket_api.async_register_command(hass, websocket_cancel_add_device)
websocket_api.async_register_command(hass, websocket_get_aldb)
websocket_api.async_register_command(hass, websocket_change_aldb_record)
@ -42,3 +53,31 @@ def async_load_api(hass):
websocket_api.async_register_command(hass, websocket_write_properties)
websocket_api.async_register_command(hass, websocket_load_properties)
websocket_api.async_register_command(hass, websocket_reset_properties)
def get_entrypoint(is_dev):
"""Get the entry point for the frontend."""
if is_dev:
return "entrypoint.js"
async def async_register_insteon_frontend(hass: HomeAssistant):
"""Register the Insteon frontend configuration panel."""
# Add to sidepanel if needed
if DOMAIN not in hass.data.get("frontend_panels", {}):
dev_path = hass.data.get(DOMAIN, {}).get(CONF_DEV_PATH)
is_dev = dev_path is not None
path = dev_path if dev_path else locate_dir()
build_id = get_build_id(is_dev)
hass.http.register_static_path(URL_BASE, path, cache_headers=not is_dev)
await panel_custom.async_register_panel(
hass=hass,
frontend_url_path=DOMAIN,
webcomponent_name="insteon-frontend",
sidebar_title=DOMAIN.capitalize(),
sidebar_icon="mdi:power",
module_url=f"{URL_BASE}/entrypoint-{build_id}.js",
embed_iframe=True,
require_admin=True,
)

66
homeassistant/components/insteon/api/device.py
View File

@ -1,17 +1,20 @@
"""API interface to get an Insteon device."""
from pyinsteon import devices
from pyinsteon.constants import DeviceAction
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from ..const import (
DEVICE_ADDRESS,
DEVICE_ID,
DOMAIN,
HA_DEVICE_NOT_FOUND,
ID,
INSTEON_DEVICE_NOT_FOUND,
MULTIPLE,
TYPE,
)
@ -21,6 +24,12 @@ def compute_device_name(ha_device):
return ha_device.name_by_user if ha_device.name_by_user else ha_device.name
async def async_add_devices(address, multiple):
"""Add one or more Insteon devices."""
async for _ in devices.async_add_device(address=address, multiple=multiple):
pass
def get_insteon_device_from_ha_device(ha_device):
"""Return the Insteon device from an HA device."""
for identifier in ha_device.identifiers:
@ -74,3 +83,58 @@ async def websocket_get_device(
"aldb_status": str(device.aldb.status),
}
connection.send_result(msg[ID], device_info)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "insteon/device/add",
vol.Required(MULTIPLE): bool,
vol.Optional(DEVICE_ADDRESS): str,
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_add_device(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Add one or more Insteon devices."""
@callback
def linking_complete(address: str, action: DeviceAction):
"""Forward device events to websocket."""
if action == DeviceAction.COMPLETED:
forward_data = {"type": "linking_stopped", "address": ""}
else:
return
connection.send_message(websocket_api.event_message(msg["id"], forward_data))
@callback
def async_cleanup() -> None:
"""Remove signal listeners."""
devices.unsubscribe(linking_complete)
connection.subscriptions[msg["id"]] = async_cleanup
devices.subscribe(linking_complete)
async for address in devices.async_add_device(
address=msg.get(DEVICE_ADDRESS), multiple=msg[MULTIPLE]
):
forward_data = {"type": "device_added", "address": str(address)}
connection.send_message(websocket_api.event_message(msg["id"], forward_data))
connection.send_result(msg[ID])
@websocket_api.websocket_command({vol.Required(TYPE): "insteon/device/add/cancel"})
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_cancel_add_device(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Cancel the Insteon all-linking process."""
await devices.async_cancel_all_linking()
connection.send_result(msg[ID])

340
homeassistant/components/insteon/api/properties.py
View File

@ -1,17 +1,15 @@
"""Property update methods and schemas."""
from itertools import chain
from pyinsteon import devices
from pyinsteon.constants import RAMP_RATES, ResponseStatus
from pyinsteon.device_types.device_base import Device
from pyinsteon.extended_property import (
NON_TOGGLE_MASK,
NON_TOGGLE_ON_OFF_MASK,
OFF_MASK,
ON_MASK,
RAMP_RATE,
from pyinsteon.config import RADIO_BUTTON_GROUPS, RAMP_RATE_IN_SEC, get_usable_value
from pyinsteon.constants import (
RAMP_RATES_SEC,
PropertyType,
RelayMode,
ResponseStatus,
ToggleMode,
)
from pyinsteon.utils import ramp_rate_to_seconds, seconds_to_ramp_rate
from pyinsteon.device_types.device_base import Device
import voluptuous as vol
import voluptuous_serialize
@ -29,19 +27,12 @@ from ..const import (
)
from .device import notify_device_not_found
TOGGLE_ON_OFF_MODE = "toggle_on_off_mode"
NON_TOGGLE_ON_MODE = "non_toggle_on_mode"
NON_TOGGLE_OFF_MODE = "non_toggle_off_mode"
RADIO_BUTTON_GROUP_PROP = "radio_button_group_"
TOGGLE_PROP = "toggle_"
RAMP_RATE_SECONDS = list(dict.fromkeys(RAMP_RATES.values()))
SHOW_ADVANCED = "show_advanced"
RAMP_RATE_SECONDS = list(dict.fromkeys(RAMP_RATES_SEC))
RAMP_RATE_SECONDS.sort()
TOGGLE_MODES = {TOGGLE_ON_OFF_MODE: 0, NON_TOGGLE_ON_MODE: 1, NON_TOGGLE_OFF_MODE: 2}
TOGGLE_MODES_SCHEMA = {
0: TOGGLE_ON_OFF_MODE,
1: NON_TOGGLE_ON_MODE,
2: NON_TOGGLE_OFF_MODE,
}
RAMP_RATE_LIST = [str(seconds) for seconds in RAMP_RATE_SECONDS]
TOGGLE_MODES = [str(ToggleMode(v)).lower() for v in list(ToggleMode)]
RELAY_MODES = [str(RelayMode(v)).lower() for v in list(RelayMode)]
def _bool_schema(name):
@ -52,239 +43,116 @@ def _byte_schema(name):
return voluptuous_serialize.convert(vol.Schema({vol.Required(name): cv.byte}))[0]
def _ramp_rate_schema(name):
def _float_schema(name):
return voluptuous_serialize.convert(vol.Schema({vol.Required(name): float}))[0]
def _list_schema(name, values):
return voluptuous_serialize.convert(
vol.Schema({vol.Required(name): vol.In(RAMP_RATE_SECONDS)}),
vol.Schema({vol.Required(name): vol.In(values)}),
custom_serializer=cv.custom_serializer,
)[0]
def get_properties(device: Device):
def _multi_select_schema(name, values):
return voluptuous_serialize.convert(
vol.Schema({vol.Optional(name): cv.multi_select(values)}),
custom_serializer=cv.custom_serializer,
)[0]
def _read_only_schema(name, value):
"""Return a constant value schema."""
return voluptuous_serialize.convert(vol.Schema({vol.Required(name): value}))[0]
def get_schema(prop, name, groups):
"""Return the correct shema type."""
if prop.is_read_only:
return _read_only_schema(name, prop.value)
if name == RAMP_RATE_IN_SEC:
return _list_schema(name, RAMP_RATE_LIST)
if name == RADIO_BUTTON_GROUPS:
button_list = {str(group): groups[group].name for group in groups if group != 1}
return _multi_select_schema(name, button_list)
if prop.value_type == bool:
return _bool_schema(name)
if prop.value_type == int:
return _byte_schema(name)
if prop.value_type == float:
return _float_schema(name)
if prop.value_type == ToggleMode:
return _list_schema(name, TOGGLE_MODES)
if prop.value_type == RelayMode:
return _list_schema(name, RELAY_MODES)
return None
def get_properties(device: Device, show_advanced=False):
"""Get the properties of an Insteon device and return the records and schema."""
properties = []
schema = {}
# Limit the properties we manage at this time.
for prop_name in device.operating_flags:
if not device.operating_flags[prop_name].is_read_only:
prop_dict, schema_dict = _get_property(device.operating_flags[prop_name])
properties.append(prop_dict)
schema[prop_name] = schema_dict
mask_found = False
for prop_name in device.properties:
if device.properties[prop_name].is_read_only:
for name, prop in device.configuration.items():
if prop.is_read_only and not show_advanced:
continue
if prop_name == RAMP_RATE:
rr_prop, rr_schema = _get_ramp_rate_property(device.properties[prop_name])
properties.append(rr_prop)
schema[RAMP_RATE] = rr_schema
prop_schema = get_schema(prop, name, device.groups)
if name == "momentary_delay":
print(prop_schema)
if prop_schema is None:
continue
schema[name] = prop_schema
properties.append(property_to_dict(prop))
elif not mask_found and "mask" in prop_name:
mask_found = True
toggle_props, toggle_schema = _get_toggle_properties(device)
properties.extend(toggle_props)
schema.update(toggle_schema)
rb_props, rb_schema = _get_radio_button_properties(device)
properties.extend(rb_props)
schema.update(rb_schema)
else:
prop_dict, schema_dict = _get_property(device.properties[prop_name])
properties.append(prop_dict)
schema[prop_name] = schema_dict
if show_advanced:
for name, prop in device.operating_flags.items():
if prop.property_type != PropertyType.ADVANCED:
continue
prop_schema = get_schema(prop, name, device.groups)
if prop_schema is not None:
schema[name] = prop_schema
properties.append(property_to_dict(prop))
for name, prop in device.properties.items():
if prop.property_type != PropertyType.ADVANCED:
continue
prop_schema = get_schema(prop, name, device.groups)
if prop_schema is not None:
schema[name] = prop_schema
properties.append(property_to_dict(prop))
return properties, schema
def set_property(device, prop_name: str, value):
"""Update a property value."""
if isinstance(value, bool) and prop_name in device.operating_flags:
device.operating_flags[prop_name].new_value = value
elif prop_name == RAMP_RATE:
device.properties[prop_name].new_value = seconds_to_ramp_rate(value)
elif prop_name.startswith(RADIO_BUTTON_GROUP_PROP):
buttons = [int(button) for button in value]
rb_groups = _calc_radio_button_groups(device)
curr_group = int(prop_name[len(RADIO_BUTTON_GROUP_PROP) :])
if len(rb_groups) > curr_group:
removed = [btn for btn in rb_groups[curr_group] if btn not in buttons]
if removed:
device.clear_radio_buttons(removed)
if buttons:
device.set_radio_buttons(buttons)
elif prop_name.startswith(TOGGLE_PROP):
button_name = prop_name[len(TOGGLE_PROP) :]
for button in device.groups:
if device.groups[button].name == button_name:
device.set_toggle_mode(button, int(value))
else:
device.properties[prop_name].new_value = value
def _get_property(prop):
def property_to_dict(prop):
"""Return a property data row."""
value, modified = _get_usable_value(prop)
value = get_usable_value(prop)
modified = value == prop.new_value
if prop.value_type in [ToggleMode, RelayMode] or prop.name == RAMP_RATE_IN_SEC:
value = str(value).lower()
prop_dict = {"name": prop.name, "value": value, "modified": modified}
if isinstance(prop.value, bool):
schema = _bool_schema(prop.name)
return prop_dict
def update_property(device, prop_name, value):
"""Update the value of a device property."""
prop = device.configuration[prop_name]
if prop.value_type == ToggleMode:
toggle_mode = getattr(ToggleMode, value.upper())
prop.new_value = toggle_mode
elif prop.value_type == RelayMode:
relay_mode = getattr(RelayMode, value.upper())
prop.new_value = relay_mode
else:
schema = _byte_schema(prop.name)
return prop_dict, {"name": prop.name, **schema}
def _get_toggle_properties(device):
"""Generate the mask properties for a KPL device."""
props = []
schema = {}
toggle_prop = device.properties[NON_TOGGLE_MASK]
toggle_on_prop = device.properties[NON_TOGGLE_ON_OFF_MASK]
for button in device.groups:
name = f"{TOGGLE_PROP}{device.groups[button].name}"
value, modified = _toggle_button_value(toggle_prop, toggle_on_prop, button)
props.append({"name": name, "value": value, "modified": modified})
toggle_schema = vol.Schema({vol.Required(name): vol.In(TOGGLE_MODES_SCHEMA)})
toggle_schema_dict = voluptuous_serialize.convert(
toggle_schema, custom_serializer=cv.custom_serializer
)
schema[name] = toggle_schema_dict[0]
return props, schema
def _toggle_button_value(non_toggle_prop, toggle_on_prop, button):
"""Determine the toggle value of a button."""
toggle_mask, toggle_modified = _get_usable_value(non_toggle_prop)
toggle_on_mask, toggle_on_modified = _get_usable_value(toggle_on_prop)
bit = button - 1
if not toggle_mask & 1 << bit:
value = 0
else:
if toggle_on_mask & 1 << bit:
value = 1
else:
value = 2
modified = False
if toggle_modified:
curr_bit = non_toggle_prop.value & 1 << bit
new_bit = non_toggle_prop.new_value & 1 << bit
modified = not curr_bit == new_bit
if not modified and value != 0 and toggle_on_modified:
curr_bit = toggle_on_prop.value & 1 << bit
new_bit = toggle_on_prop.new_value & 1 << bit
modified = not curr_bit == new_bit
return value, modified
def _get_radio_button_properties(device):
"""Return the values and schema to set KPL buttons as radio buttons."""
rb_groups = _calc_radio_button_groups(device)
props = []
schema = {}
index = 0
remaining_buttons = []
buttons_in_groups = list(chain.from_iterable(rb_groups))
# Identify buttons not belonging to any group
for button in device.groups:
if button not in buttons_in_groups:
remaining_buttons.append(button)
for rb_group in rb_groups:
name = f"{RADIO_BUTTON_GROUP_PROP}{index}"
button_1 = rb_group[0]
button_str = f"_{button_1}" if button_1 != 1 else ""
on_mask = device.properties[f"{ON_MASK}{button_str}"]
off_mask = device.properties[f"{OFF_MASK}{button_str}"]
modified = on_mask.is_dirty or off_mask.is_dirty
props.append(
{
"name": name,
"modified": modified,
"value": rb_group,
}
)
options = {
button: device.groups[button].name
for button in chain.from_iterable([rb_group, remaining_buttons])
}
rb_schema = vol.Schema({vol.Optional(name): cv.multi_select(options)})
rb_schema_dict = voluptuous_serialize.convert(
rb_schema, custom_serializer=cv.custom_serializer
)
schema[name] = rb_schema_dict[0]
index += 1
if len(remaining_buttons) > 1:
name = f"{RADIO_BUTTON_GROUP_PROP}{index}"
props.append(
{
"name": name,
"modified": False,
"value": [],
}
)
options = {button: device.groups[button].name for button in remaining_buttons}
rb_schema = vol.Schema({vol.Optional(name): cv.multi_select(options)})
rb_schema_dict = voluptuous_serialize.convert(
rb_schema, custom_serializer=cv.custom_serializer
)
schema[name] = rb_schema_dict[0]
return props, schema
def _calc_radio_button_groups(device):
"""Return existing radio button groups."""
rb_groups = []
for button in device.groups:
if button not in list(chain.from_iterable(rb_groups)):
button_str = "" if button == 1 else f"_{button}"
on_mask, _ = _get_usable_value(device.properties[f"{ON_MASK}{button_str}"])
if on_mask != 0:
rb_group = [button]
for bit in list(range(0, button - 1)) + list(range(button, 8)):
if on_mask & 1 << bit:
rb_group.append(bit + 1)
if len(rb_group) > 1:
rb_groups.append(rb_group)
return rb_groups
def _get_ramp_rate_property(prop):
"""Return the value and schema of a ramp rate property."""
rr_prop, _ = _get_property(prop)
rr_prop["value"] = ramp_rate_to_seconds(rr_prop["value"])
return rr_prop, _ramp_rate_schema(prop.name)
def _get_usable_value(prop):
"""Return the current or the modified value of a property."""
value = prop.value if prop.new_value is None else prop.new_value
return value, prop.is_dirty
prop.new_value = value
@websocket_api.websocket_command(
{
vol.Required(TYPE): "insteon/properties/get",
vol.Required(DEVICE_ADDRESS): str,
vol.Required(SHOW_ADVANCED): bool,
}
)
@websocket_api.require_admin
@ -299,7 +167,7 @@ async def websocket_get_properties(
notify_device_not_found(connection, msg, INSTEON_DEVICE_NOT_FOUND)
return
properties, schema = get_properties(device)
properties, schema = get_properties(device, msg[SHOW_ADVANCED])
connection.send_result(msg[ID], {"properties": properties, "schema": schema})
@ -324,7 +192,7 @@ async def websocket_change_properties_record(
notify_device_not_found(connection, msg, INSTEON_DEVICE_NOT_FOUND)
return
set_property(device, msg[PROPERTY_NAME], msg[PROPERTY_VALUE])
update_property(device, msg[PROPERTY_NAME], msg[PROPERTY_VALUE])
connection.send_result(msg[ID])
@ -346,10 +214,9 @@ async def websocket_write_properties(
notify_device_not_found(connection, msg, INSTEON_DEVICE_NOT_FOUND)
return
result1 = await device.async_write_op_flags()
result2 = await device.async_write_ext_properties()
result = await device.async_write_config()
await devices.async_save(workdir=hass.config.config_dir)
if result1 != ResponseStatus.SUCCESS or result2 != ResponseStatus.SUCCESS:
if result != ResponseStatus.SUCCESS:
connection.send_message(
websocket_api.error_message(
msg[ID], "write_failed", "properties not written to device"
@ -377,10 +244,9 @@ async def websocket_load_properties(
notify_device_not_found(connection, msg, INSTEON_DEVICE_NOT_FOUND)
return
result1 = await device.async_read_op_flags()
result2 = await device.async_read_ext_properties()
result, _ = await device.async_read_config(read_aldb=False)
await devices.async_save(workdir=hass.config.config_dir)
if result1 != ResponseStatus.SUCCESS or result2 != ResponseStatus.SUCCESS:
if result != ResponseStatus.SUCCESS:
connection.send_message(
websocket_api.error_message(
msg[ID], "load_failed", "properties not loaded from device"

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

@ -1,8 +1,8 @@
"""Support for Insteon thermostat."""
from __future__ import annotations
from pyinsteon.config import CELSIUS
from pyinsteon.constants import ThermostatMode
from pyinsteon.operating_flag import CELSIUS
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (

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

@ -70,6 +70,7 @@ CONF_DIM_STEPS = "dim_steps"
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"
CONF_DEV_PATH = "dev_path"
PORT_HUB_V1 = 9761
PORT_HUB_V2 = 25105
@ -172,5 +173,6 @@ PROPERTY_NAME = "name"
PROPERTY_VALUE = "value"
HA_DEVICE_NOT_FOUND = "ha_device_not_found"
INSTEON_DEVICE_NOT_FOUND = "insteon_device_not_found"
MULTIPLE = "multiple"
INSTEON_ADDR_REGEX = re.compile(r"([A-Fa-f0-9]{2}\.?[A-Fa-f0-9]{2}\.?[A-Fa-f0-9]{2})$")

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

@ -85,7 +85,7 @@ class InsteonEntity(Entity):
"""Return device information."""
return DeviceInfo(
identifiers={(DOMAIN, str(self._insteon_device.address))},
manufacturer="Smart Home",
manufacturer="SmartLabs, Inc",
model=f"{self._insteon_device.model} ({self._insteon_device.cat!r}, 0x{self._insteon_device.subcat:02x})",
name=f"{self._insteon_device.description} {self._insteon_device.address}",
sw_version=f"{self._insteon_device.firmware:02x} Engine Version: {self._insteon_device.engine_version}",

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

@ -1,5 +1,5 @@
"""Support for Insteon lights via PowerLinc Modem."""
from pyinsteon.extended_property import ON_LEVEL
from pyinsteon.config import ON_LEVEL
from homeassistant.components.light import (
ATTR_BRIGHTNESS,

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

@ -2,13 +2,17 @@
"domain": "insteon",
"name": "Insteon",
"documentation": "https://www.home-assistant.io/integrations/insteon",
"requirements": ["pyinsteon==1.0.13"],
"dependencies": ["http", "websocket_api"],
"requirements": [
"pyinsteon==1.1.0b1",
"insteon-frontend-home-assistant==0.1.0"
],
"codeowners": ["@teharris1"],
"dhcp": [{ "macaddress": "000EF3*" }, { "registered_devices": true }],
"config_flow": true,
"iot_class": "local_push",
"loggers": ["pyinsteon", "pypubsub"],
"after_dependencies": ["usb"],
"after_dependencies": ["panel_custom", "usb"],
"usb": [
{
"vid": "10BF"

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

@ -22,6 +22,7 @@ import homeassistant.helpers.config_validation as cv
from .const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_DIM_STEPS,
CONF_FIRMWARE,
CONF_HOUSECODE,
@ -121,6 +122,7 @@ CONFIG_SCHEMA = vol.Schema(
vol.Optional(CONF_X10): vol.All(
cv.ensure_list_csv, [CONF_X10_SCHEMA]
),
vol.Optional(CONF_DEV_PATH): cv.string,
},
extra=vol.ALLOW_EXTRA,
required=True,

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

@ -4,7 +4,7 @@ import logging
from pyinsteon import devices
from pyinsteon.address import Address
from pyinsteon.constants import ALDBStatus
from pyinsteon.constants import ALDBStatus, DeviceAction
from pyinsteon.events import OFF_EVENT, OFF_FAST_EVENT, ON_EVENT, ON_FAST_EVENT
from pyinsteon.managers.link_manager import (
async_enter_linking_mode,
@ -137,9 +137,10 @@ def register_new_device_callback(hass):
"""Register callback for new Insteon device."""
@callback
def async_new_insteon_device(address=None):
def async_new_insteon_device(address, action: DeviceAction):
"""Detect device from transport to be delegated to platform."""
hass.async_create_task(async_create_new_entities(address))
if action == DeviceAction.ADDED:
hass.async_create_task(async_create_new_entities(address))
async def async_create_new_entities(address):
_LOGGER.debug(

5
requirements_all.txt
View File

@ -881,6 +881,9 @@ influxdb-client==1.24.0
# homeassistant.components.influxdb
influxdb==5.3.1
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.1.0
# homeassistant.components.intellifire
intellifire4py==1.0.2
@ -1544,7 +1547,7 @@ pyialarm==1.9.0
pyicloud==1.0.0
# homeassistant.components.insteon
pyinsteon==1.0.13
pyinsteon==1.1.0b1
# homeassistant.components.intesishome
pyintesishome==1.7.6

5
requirements_test_all.txt
View File

@ -618,6 +618,9 @@ influxdb-client==1.24.0
# homeassistant.components.influxdb
influxdb==5.3.1
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.1.0
# homeassistant.components.intellifire
intellifire4py==1.0.2
@ -1026,7 +1029,7 @@ pyialarm==1.9.0
pyicloud==1.0.0
# homeassistant.components.insteon
pyinsteon==1.0.13
pyinsteon==1.1.0b1
# homeassistant.components.ipma
pyipma==2.0.5

18
tests/components/insteon/fixtures/iolinc_properties.json Normal file
View File

@ -0,0 +1,18 @@
{
"operating_flags": {
"program_lock_on": false,
"blink_on_tx_on": true,
"relay_on_sense_on": false,
"momentary_on": true,
"momentary_on_off_trigger": false,
"x10_off": true,
"sense_sends_off": true,
"momentary_follow_sense": false
},
"properties": {
"prescaler": 1,
"delay": 50,
"x10_house": 32,
"x10_unit": 0
}
}

38
tests/components/insteon/mock_devices.py
View File

@ -1,15 +1,20 @@
"""Mock devices object to test Insteon."""
import asyncio
from unittest.mock import AsyncMock, MagicMock
from pyinsteon.address import Address
from pyinsteon.constants import ALDBStatus, ResponseStatus
from pyinsteon.device_types import (
DimmableLightingControl_KeypadLinc_8,
GeneralController,
GeneralController_RemoteLinc,
Hub,
SensorsActuators_IOLink,
SwitchedLightingControl_SwitchLinc,
)
from pyinsteon.managers.saved_devices_manager import dict_to_aldb_record
from pyinsteon.topics import DEVICE_LIST_CHANGED
from pyinsteon.utils import subscribe_topic
class MockSwitchLinc(SwitchedLightingControl_SwitchLinc):
@ -31,7 +36,10 @@ class MockDevices:
self._connected = connected
self.async_save = AsyncMock()
self.add_x10_device = MagicMock()
self.async_read_config = AsyncMock()
self.set_id = MagicMock()
self.async_add_device_called_with = {}
self.async_cancel_all_linking = AsyncMock()
def __getitem__(self, address):
"""Return a a device from the device address."""
@ -56,18 +64,24 @@ class MockDevices:
addr1 = Address("11.11.11")
addr2 = Address("22.22.22")
addr3 = Address("33.33.33")
addr4 = Address("44.44.44")
self._devices[addr0] = Hub(addr0, 0x03, 0x00, 0x00, "Hub AA.AA.AA", "0")
self._devices[addr1] = MockSwitchLinc(
addr1, 0x02, 0x00, 0x00, "Device 11.11.11", "1"
)
self._devices[addr2] = GeneralController(
self._devices[addr2] = GeneralController_RemoteLinc(
addr2, 0x00, 0x00, 0x00, "Device 22.22.22", "2"
)
self._devices[addr3] = DimmableLightingControl_KeypadLinc_8(
addr3, 0x02, 0x00, 0x00, "Device 33.33.33", "3"
)
self._devices[addr4] = SensorsActuators_IOLink(
addr4, 0x07, 0x00, 0x00, "Device 44.44.44", "4"
)
for device in [self._devices[addr] for addr in [addr1, addr2, addr3]]:
for device in [
self._devices[addr] for addr in [addr1, addr2, addr3, addr4]
]:
device.async_read_config = AsyncMock()
device.aldb.async_write = AsyncMock()
device.aldb.async_load = AsyncMock()
@ -85,7 +99,7 @@ class MockDevices:
return_value=ResponseStatus.SUCCESS
)
for device in [self._devices[addr] for addr in [addr2, addr3]]:
for device in [self._devices[addr] for addr in [addr2, addr3, addr4]]:
device.async_status = AsyncMock()
self._devices[addr1].async_status = AsyncMock(side_effect=AttributeError)
self._devices[addr0].aldb.async_load = AsyncMock()
@ -104,6 +118,7 @@ class MockDevices:
)
self.modem = self._devices[addr0]
self.modem.async_read_config = AsyncMock()
def fill_aldb(self, address, records):
"""Fill the All-Link Database for a device."""
@ -126,3 +141,18 @@ class MockDevices:
value = properties[flag]
if device.properties.get(flag):
device.properties[flag].load(value)
async def async_add_device(self, address=None, multiple=False):
"""Mock the async_add_device method."""
self.async_add_device_called_with = {"address": address, "multiple": multiple}
if multiple:
yield "aa.bb.cc"
await asyncio.sleep(0.01)
yield "bb.cc.dd"
if address:
yield address
await asyncio.sleep(0.01)
def subscribe(self, listener):
"""Mock the subscribe function."""
subscribe_topic(listener, DEVICE_LIST_CHANGED)

51
tests/components/insteon/test_api_device.py
View File

@ -1,6 +1,11 @@
"""Test the device level APIs."""
import asyncio
from unittest.mock import patch
from pyinsteon.constants import DeviceAction
from pyinsteon.topics import DEVICE_LIST_CHANGED
from pyinsteon.utils import publish_topic
from homeassistant.components import insteon
from homeassistant.components.insteon.api import async_load_api
from homeassistant.components.insteon.api.device import (
@ -11,7 +16,7 @@ from homeassistant.components.insteon.api.device import (
TYPE,
async_device_name,
)
from homeassistant.components.insteon.const import DOMAIN
from homeassistant.components.insteon.const import DOMAIN, MULTIPLE
from homeassistant.helpers.device_registry import async_get_registry
from .const import MOCK_USER_INPUT_PLM
@ -137,3 +142,47 @@ async def test_get_ha_device_name(hass, hass_ws_client):
# Test no HA or Insteon device
name = await async_device_name(device_reg, "BB.BB.BB")
assert name == ""
async def test_add_device_api(hass, hass_ws_client):
"""Test adding an Insteon device."""
ws_client, devices, _, _ = await _async_setup(hass, hass_ws_client)
with patch.object(insteon.api.device, "devices", devices):
await ws_client.send_json({ID: 2, TYPE: "insteon/device/add", MULTIPLE: True})
await asyncio.sleep(0.01)
assert devices.async_add_device_called_with.get("address") is None
assert devices.async_add_device_called_with["multiple"] is True
msg = await ws_client.receive_json()
assert msg["event"]["type"] == "device_added"
assert msg["event"]["address"] == "aa.bb.cc"
msg = await ws_client.receive_json()
assert msg["event"]["type"] == "device_added"
assert msg["event"]["address"] == "bb.cc.dd"
publish_topic(
DEVICE_LIST_CHANGED,
address=None,
action=DeviceAction.COMPLETED,
)
msg = await ws_client.receive_json()
assert msg["event"]["type"] == "linking_stopped"
async def test_cancel_add_device(hass, hass_ws_client):
"""Test cancelling adding of a new device."""
ws_client, devices, _, _ = await _async_setup(hass, hass_ws_client)
with patch.object(insteon.api.aldb, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/device/add/cancel",
}
)
msg = await ws_client.receive_json()
assert msg["success"]

467
tests/components/insteon/test_api_properties.py
View File

@ -1,8 +1,11 @@
"""Test the Insteon properties APIs."""
import json
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from pyinsteon.config import MOMENTARY_DELAY, RELAY_MODE, TOGGLE_BUTTON
from pyinsteon.config.extended_property import ExtendedProperty
from pyinsteon.constants import RelayMode, ToggleMode
import pytest
from homeassistant.components import insteon
@ -11,19 +14,12 @@ from homeassistant.components.insteon.api.device import INSTEON_DEVICE_NOT_FOUND
from homeassistant.components.insteon.api.properties import (
DEVICE_ADDRESS,
ID,
NON_TOGGLE_MASK,
NON_TOGGLE_OFF_MODE,
NON_TOGGLE_ON_MODE,
NON_TOGGLE_ON_OFF_MASK,
PROPERTY_NAME,
PROPERTY_VALUE,
RADIO_BUTTON_GROUP_PROP,
TOGGLE_MODES,
TOGGLE_ON_OFF_MODE,
TOGGLE_PROP,
RADIO_BUTTON_GROUPS,
RAMP_RATE_IN_SEC,
SHOW_ADVANCED,
TYPE,
_get_radio_button_properties,
_get_toggle_properties,
)
from .mock_devices import MockDevices
@ -31,43 +27,172 @@ from .mock_devices import MockDevices
from tests.common import load_fixture
@pytest.fixture(name="properties_data", scope="session")
def aldb_data_fixture():
@pytest.fixture(name="kpl_properties_data", scope="session")
def kpl_properties_data_fixture():
"""Load the controller state fixture data."""
return json.loads(load_fixture("insteon/kpl_properties.json"))
async def _setup(hass, hass_ws_client, properties_data):
@pytest.fixture(name="iolinc_properties_data", scope="session")
def iolinc_properties_data_fixture():
"""Load the controller state fixture data."""
return json.loads(load_fixture("insteon/iolinc_properties.json"))
async def _setup(hass, hass_ws_client, address, properties_data):
"""Set up tests."""
ws_client = await hass_ws_client(hass)
devices = MockDevices()
await devices.async_load()
devices.fill_properties("33.33.33", properties_data)
devices.fill_properties(address, properties_data)
async_load_api(hass)
return ws_client, devices
async def test_get_properties(hass, hass_ws_client, properties_data):
async def test_get_properties(
hass, hass_ws_client, kpl_properties_data, iolinc_properties_data
):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{ID: 2, TYPE: "insteon/properties/get", DEVICE_ADDRESS: "33.33.33"}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 54
async def test_change_operating_flag(hass, hass_ws_client, properties_data):
"""Test changing an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
devices.fill_properties("44.44.44", iolinc_properties_data)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "33.33.33",
SHOW_ADVANCED: False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 18
await ws_client.send_json(
{
ID: 3,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 6
await ws_client.send_json(
{
ID: 4,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "33.33.33",
SHOW_ADVANCED: True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 69
await ws_client.send_json(
{
ID: 5,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 14
async def test_get_read_only_properties(hass, hass_ws_client, iolinc_properties_data):
"""Test getting an Insteon device's properties."""
mock_read_only = ExtendedProperty(
"44.44.44", "mock_read_only", bool, is_read_only=True
)
mock_read_only.load(False)
ws_client, devices = await _setup(
hass, hass_ws_client, "44.44.44", iolinc_properties_data
)
device = devices["44.44.44"]
device.configuration["mock_read_only"] = mock_read_only
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 6
await ws_client.send_json(
{
ID: 3,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 15
async def test_get_unknown_properties(hass, hass_ws_client, iolinc_properties_data):
"""Test getting an Insteon device's properties."""
class UnknownType:
"""Mock unknown data type."""
mock_unknown = ExtendedProperty("44.44.44", "mock_unknown", UnknownType)
ws_client, devices = await _setup(
hass, hass_ws_client, "44.44.44", iolinc_properties_data
)
device = devices["44.44.44"]
device.configuration["mock_unknown"] = mock_unknown
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 6
await ws_client.send_json(
{
ID: 3,
TYPE: "insteon/properties/get",
DEVICE_ADDRESS: "44.44.44",
SHOW_ADVANCED: True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["properties"]) == 14
async def test_change_bool_property(hass, hass_ws_client, kpl_properties_data):
"""Test changing a bool type properties."""
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 3,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: "led_off",
@ -79,29 +204,33 @@ async def test_change_operating_flag(hass, hass_ws_client, properties_data):
assert devices["33.33.33"].operating_flags["led_off"].is_dirty
async def test_change_property(hass, hass_ws_client, properties_data):
"""Test changing an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
async def test_change_int_property(hass, hass_ws_client, kpl_properties_data):
"""Test changing a int type properties."""
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
ID: 4,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: "on_mask",
PROPERTY_NAME: "led_dimming",
PROPERTY_VALUE: 100,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert devices["33.33.33"].properties["on_mask"].new_value == 100
assert devices["33.33.33"].properties["on_mask"].is_dirty
assert devices["33.33.33"].properties["led_dimming"].new_value == 100
assert devices["33.33.33"].properties["led_dimming"].is_dirty
async def test_change_ramp_rate_property(hass, hass_ws_client, properties_data):
"""Test changing an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
async def test_change_ramp_rate_property(hass, hass_ws_client, kpl_properties_data):
"""Test changing an Insteon device's ramp rate properties."""
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
@ -109,7 +238,7 @@ async def test_change_ramp_rate_property(hass, hass_ws_client, properties_data):
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: "ramp_rate",
PROPERTY_NAME: RAMP_RATE_IN_SEC,
PROPERTY_VALUE: 4.5,
}
)
@ -119,208 +248,126 @@ async def test_change_ramp_rate_property(hass, hass_ws_client, properties_data):
assert devices["33.33.33"].properties["ramp_rate"].is_dirty
async def test_change_radio_button_group(hass, hass_ws_client, properties_data):
async def test_change_radio_button_group(hass, hass_ws_client, kpl_properties_data):
"""Test changing an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
rb_props, schema = _get_radio_button_properties(devices["33.33.33"])
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
rb_groups = devices["33.33.33"].configuration[RADIO_BUTTON_GROUPS]
# Make sure the baseline is correct
assert rb_props[0]["name"] == f"{RADIO_BUTTON_GROUP_PROP}0"
assert rb_props[0]["value"] == [4, 5]
assert rb_props[1]["value"] == [7, 8]
assert rb_props[2]["value"] == []
assert schema[f"{RADIO_BUTTON_GROUP_PROP}0"]["options"].get(1)
assert schema[f"{RADIO_BUTTON_GROUP_PROP}1"]["options"].get(1)
assert devices["33.33.33"].properties["on_mask"].value == 0
assert devices["33.33.33"].properties["off_mask"].value == 0
assert not devices["33.33.33"].properties["on_mask"].is_dirty
assert not devices["33.33.33"].properties["off_mask"].is_dirty
assert rb_groups.value[0] == [4, 5]
assert rb_groups.value[1] == [7, 8]
# Add button 1 to the group
rb_props[0]["value"].append(1)
new_groups_1 = [[1, 4, 5], [7, 8]]
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: f"{RADIO_BUTTON_GROUP_PROP}0",
PROPERTY_VALUE: rb_props[0]["value"],
PROPERTY_NAME: RADIO_BUTTON_GROUPS,
PROPERTY_VALUE: new_groups_1,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert rb_groups.new_value[0] == [1, 4, 5]
assert rb_groups.new_value[1] == [7, 8]
new_rb_props, _ = _get_radio_button_properties(devices["33.33.33"])
assert 1 in new_rb_props[0]["value"]
assert 4 in new_rb_props[0]["value"]
assert 5 in new_rb_props[0]["value"]
assert schema[f"{RADIO_BUTTON_GROUP_PROP}0"]["options"].get(1)
assert schema[f"{RADIO_BUTTON_GROUP_PROP}1"]["options"].get(1)
assert devices["33.33.33"].properties["on_mask"].new_value == 0x18
assert devices["33.33.33"].properties["off_mask"].new_value == 0x18
assert devices["33.33.33"].properties["on_mask"].is_dirty
assert devices["33.33.33"].properties["off_mask"].is_dirty
# Remove button 5
rb_props[0]["value"].remove(5)
new_groups_2 = [[1, 4], [7, 8]]
await ws_client.send_json(
{
ID: 3,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: f"{RADIO_BUTTON_GROUP_PROP}0",
PROPERTY_VALUE: rb_props[0]["value"],
PROPERTY_NAME: RADIO_BUTTON_GROUPS,
PROPERTY_VALUE: new_groups_2,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
new_rb_props, _ = _get_radio_button_properties(devices["33.33.33"])
assert 1 in new_rb_props[0]["value"]
assert 4 in new_rb_props[0]["value"]
assert 5 not in new_rb_props[0]["value"]
assert schema[f"{RADIO_BUTTON_GROUP_PROP}0"]["options"].get(1)
assert schema[f"{RADIO_BUTTON_GROUP_PROP}1"]["options"].get(1)
assert devices["33.33.33"].properties["on_mask"].new_value == 0x08
assert devices["33.33.33"].properties["off_mask"].new_value == 0x08
assert devices["33.33.33"].properties["on_mask"].is_dirty
assert devices["33.33.33"].properties["off_mask"].is_dirty
# Remove button group 1
rb_props[1]["value"] = []
await ws_client.send_json(
{
ID: 5,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: f"{RADIO_BUTTON_GROUP_PROP}1",
PROPERTY_VALUE: rb_props[1]["value"],
}
)
msg = await ws_client.receive_json()
assert msg["success"]
new_rb_props, _ = _get_radio_button_properties(devices["33.33.33"])
assert len(new_rb_props) == 2
assert new_rb_props[0]["value"] == [1, 4]
assert new_rb_props[1]["value"] == []
assert rb_groups.new_value[0] == [1, 4]
assert rb_groups.new_value[1] == [7, 8]
async def test_create_radio_button_group(hass, hass_ws_client, properties_data):
"""Test changing an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
rb_props, _ = _get_radio_button_properties(devices["33.33.33"])
# Make sure the baseline is correct
assert len(rb_props) == 3
rb_props[0]["value"].append("1")
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: f"{RADIO_BUTTON_GROUP_PROP}2",
PROPERTY_VALUE: ["1", "3"],
}
)
msg = await ws_client.receive_json()
assert msg["success"]
new_rb_props, new_schema = _get_radio_button_properties(devices["33.33.33"])
assert len(new_rb_props) == 4
assert 1 in new_rb_props[0]["value"]
assert new_schema[f"{RADIO_BUTTON_GROUP_PROP}0"]["options"].get(1)
assert not new_schema[f"{RADIO_BUTTON_GROUP_PROP}1"]["options"].get(1)
assert devices["33.33.33"].properties["on_mask"].new_value == 4
assert devices["33.33.33"].properties["off_mask"].new_value == 4
assert devices["33.33.33"].properties["on_mask"].is_dirty
assert devices["33.33.33"].properties["off_mask"].is_dirty
async def test_change_toggle_property(hass, hass_ws_client, properties_data):
async def test_change_toggle_property(hass, hass_ws_client, kpl_properties_data):
"""Update a button's toggle mode."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
device = devices["33.33.33"]
toggle_props, _ = _get_toggle_properties(devices["33.33.33"])
# Make sure the baseline is correct
assert toggle_props[0]["name"] == f"{TOGGLE_PROP}{device.groups[1].name}"
assert toggle_props[0]["value"] == TOGGLE_MODES[TOGGLE_ON_OFF_MODE]
assert toggle_props[1]["value"] == TOGGLE_MODES[NON_TOGGLE_ON_MODE]
assert device.properties[NON_TOGGLE_MASK].value == 2
assert device.properties[NON_TOGGLE_ON_OFF_MASK].value == 2
assert not device.properties[NON_TOGGLE_MASK].is_dirty
assert not device.properties[NON_TOGGLE_ON_OFF_MASK].is_dirty
prop_name = f"{TOGGLE_BUTTON}_c"
toggle_prop = device.configuration[prop_name]
assert toggle_prop.value == ToggleMode.TOGGLE
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: toggle_props[0]["name"],
PROPERTY_VALUE: 1,
PROPERTY_NAME: prop_name,
PROPERTY_VALUE: str(ToggleMode.ON_ONLY).lower(),
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert toggle_prop.new_value == ToggleMode.ON_ONLY
new_toggle_props, _ = _get_toggle_properties(devices["33.33.33"])
assert new_toggle_props[0]["value"] == TOGGLE_MODES[NON_TOGGLE_ON_MODE]
assert device.properties[NON_TOGGLE_MASK].new_value == 3
assert device.properties[NON_TOGGLE_ON_OFF_MASK].new_value == 3
assert device.properties[NON_TOGGLE_MASK].is_dirty
assert device.properties[NON_TOGGLE_ON_OFF_MASK].is_dirty
async def test_change_relay_mode(hass, hass_ws_client, iolinc_properties_data):
"""Update a device's relay mode."""
ws_client, devices = await _setup(
hass, hass_ws_client, "44.44.44", iolinc_properties_data
)
device = devices["44.44.44"]
relay_prop = device.configuration[RELAY_MODE]
assert relay_prop.value == RelayMode.MOMENTARY_A
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 3,
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: toggle_props[0]["name"],
PROPERTY_VALUE: 2,
DEVICE_ADDRESS: "44.44.44",
PROPERTY_NAME: RELAY_MODE,
PROPERTY_VALUE: str(RelayMode.LATCHING).lower(),
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert relay_prop.new_value == RelayMode.LATCHING
new_toggle_props, _ = _get_toggle_properties(devices["33.33.33"])
assert new_toggle_props[0]["value"] == TOGGLE_MODES[NON_TOGGLE_OFF_MODE]
assert device.properties[NON_TOGGLE_MASK].new_value == 3
assert device.properties[NON_TOGGLE_ON_OFF_MASK].new_value is None
assert device.properties[NON_TOGGLE_MASK].is_dirty
assert not device.properties[NON_TOGGLE_ON_OFF_MASK].is_dirty
async def test_change_float_property(hass, hass_ws_client, iolinc_properties_data):
"""Update a float type property."""
ws_client, devices = await _setup(
hass, hass_ws_client, "44.44.44", iolinc_properties_data
)
device = devices["44.44.44"]
delay_prop = device.configuration[MOMENTARY_DELAY]
delay_prop.load(0)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{
ID: 4,
ID: 2,
TYPE: "insteon/properties/change",
DEVICE_ADDRESS: "33.33.33",
PROPERTY_NAME: toggle_props[1]["name"],
PROPERTY_VALUE: 0,
DEVICE_ADDRESS: "44.44.44",
PROPERTY_NAME: MOMENTARY_DELAY,
PROPERTY_VALUE: 1.8,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
new_toggle_props, _ = _get_toggle_properties(devices["33.33.33"])
assert new_toggle_props[1]["value"] == TOGGLE_MODES[TOGGLE_ON_OFF_MODE]
assert device.properties[NON_TOGGLE_MASK].new_value == 1
assert device.properties[NON_TOGGLE_ON_OFF_MASK].new_value == 0
assert device.properties[NON_TOGGLE_MASK].is_dirty
assert device.properties[NON_TOGGLE_ON_OFF_MASK].is_dirty
assert delay_prop.new_value == 1.8
async def test_write_properties(hass, hass_ws_client, properties_data):
async def test_write_properties(hass, hass_ws_client, kpl_properties_data):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
@ -332,9 +379,11 @@ async def test_write_properties(hass, hass_ws_client, properties_data):
assert devices["33.33.33"].async_write_ext_properties.call_count == 1
async def test_write_properties_failure(hass, hass_ws_client, properties_data):
async def test_write_properties_failure(hass, hass_ws_client, kpl_properties_data):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
@ -345,39 +394,48 @@ async def test_write_properties_failure(hass, hass_ws_client, properties_data):
assert msg["error"]["code"] == "write_failed"
async def test_load_properties(hass, hass_ws_client, properties_data):
async def test_load_properties(hass, hass_ws_client, kpl_properties_data):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
device = devices["33.33.33"]
device.async_read_config = AsyncMock(return_value=(1, 1))
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{ID: 2, TYPE: "insteon/properties/load", DEVICE_ADDRESS: "33.33.33"}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert devices["33.33.33"].async_read_op_flags.call_count == 1
assert devices["33.33.33"].async_read_ext_properties.call_count == 1
assert devices["33.33.33"].async_read_config.call_count == 1
async def test_load_properties_failure(hass, hass_ws_client, properties_data):
async def test_load_properties_failure(hass, hass_ws_client, kpl_properties_data):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
device = devices["33.33.33"]
device.async_read_config = AsyncMock(return_value=(0, 0))
with patch.object(insteon.api.properties, "devices", devices):
await ws_client.send_json(
{ID: 2, TYPE: "insteon/properties/load", DEVICE_ADDRESS: "22.22.22"}
{ID: 2, TYPE: "insteon/properties/load", DEVICE_ADDRESS: "33.33.33"}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "load_failed"
async def test_reset_properties(hass, hass_ws_client, properties_data):
async def test_reset_properties(hass, hass_ws_client, kpl_properties_data):
"""Test getting an Insteon device's properties."""
ws_client, devices = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
device = devices["33.33.33"]
device.operating_flags["led_off"].new_value = True
device.configuration["led_off"].new_value = True
device.properties["on_mask"].new_value = 100
assert device.operating_flags["led_off"].is_dirty
assert device.properties["on_mask"].is_dirty
@ -391,20 +449,23 @@ async def test_reset_properties(hass, hass_ws_client, properties_data):
assert not device.properties["on_mask"].is_dirty
async def test_bad_address(hass, hass_ws_client, properties_data):
async def test_bad_address(hass, hass_ws_client, kpl_properties_data):
"""Test for a bad Insteon address."""
ws_client, _ = await _setup(hass, hass_ws_client, properties_data)
ws_client, devices = await _setup(
hass, hass_ws_client, "33.33.33", kpl_properties_data
)
ws_id = 0
for call in ["get", "write", "load", "reset"]:
ws_id += 1
await ws_client.send_json(
{
ID: ws_id,
TYPE: f"insteon/properties/{call}",
DEVICE_ADDRESS: "99.99.99",
}
)
params = {
ID: ws_id,
TYPE: f"insteon/properties/{call}",
DEVICE_ADDRESS: "99.99.99",
}
if call == "get":
params[SHOW_ADVANCED] = False
await ws_client.send_json(params)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["message"] == INSTEON_DEVICE_NOT_FOUND

50
tests/components/insteon/test_config_flow.py
View File

@ -2,10 +2,8 @@
from unittest.mock import patch
import voluptuous_serialize
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import dhcp, usb
from homeassistant.components import usb
from homeassistant.components.insteon.config_flow import (
HUB1,
HUB2,
@ -39,7 +37,6 @@ from homeassistant.const import (
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from .const import (
MOCK_HOSTNAME,
@ -651,48 +648,3 @@ async def test_discovery_via_usb_already_setup(hass):
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
async def test_discovery_via_dhcp_hubv1(hass):
"""Test usb flow."""
await _test_dhcp(hass, HUB1)
async def test_discovery_via_dhcp_hubv2(hass):
"""Test usb flow."""
await _test_dhcp(hass, HUB2)
async def _test_dhcp(hass, modem_type):
"""Test the dhcp discovery for a moddem type."""
discovery_info = dhcp.DhcpServiceInfo(
ip="11.22.33.44", hostname="", macaddress="00:0e:f3:aa:bb:cc"
)
result = await hass.config_entries.flow.async_init(
"insteon",
context={"source": config_entries.SOURCE_DHCP},
data=discovery_info,
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch("homeassistant.components.insteon.config_flow.async_connect"), patch(
"homeassistant.components.insteon.async_setup_entry", return_value=True
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"modem_type": modem_type}
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
schema = voluptuous_serialize.convert(
result2["data_schema"],
custom_serializer=cv.custom_serializer,
)
for field in schema:
if field["name"] == "host":
assert field.get("default") == "11.22.33.44"
break

22
tests/components/insteon/test_init.py
View File

@ -7,6 +7,7 @@ from pyinsteon.address import Address
from homeassistant.components import insteon
from homeassistant.components.insteon.const import (
CONF_CAT,
CONF_DEV_PATH,
CONF_OVERRIDE,
CONF_SUBCAT,
CONF_X10,
@ -222,3 +223,24 @@ async def test_setup_entry_failed_connection(hass: HomeAssistant, caplog):
{},
)
assert "Could not connect to Insteon modem" in caplog.text
async def test_import_frontend_dev_url(hass: HomeAssistant):
"""Test importing a dev_url config entry."""
config = {}
config[DOMAIN] = {CONF_DEV_PATH: "/some/path"}
with patch.object(
insteon, "async_connect", new=mock_successful_connection
), patch.object(insteon, "close_insteon_connection"), patch.object(
insteon, "devices", new=MockDevices()
), patch(
PATCH_CONNECTION, new=mock_successful_connection
):
assert await async_setup_component(
hass,
insteon.DOMAIN,
config,
)
await hass.async_block_till_done()
await asyncio.sleep(0.01)