Insteon scenes (#87803)

* Add Insteon scene support

* Bump to pyinsteon 1.3.1

* Add tests

* Bump Insteon Panel to 0.3.1

* Change docstring
This commit is contained in:
Tom Harris 2023-02-18 09:52:49 -05:00 committed by GitHub
parent 1128041899
commit d84fde8c54
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 879 additions and 3 deletions

View file

@ -28,6 +28,12 @@ from .properties import (
websocket_reset_properties,
websocket_write_properties,
)
from .scenes import (
websocket_delete_scene,
websocket_get_scene,
websocket_get_scenes,
websocket_save_scene,
)
URL_BASE = "/insteon_static"
@ -39,6 +45,11 @@ def async_load_api(hass):
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_scenes)
websocket_api.async_register_command(hass, websocket_get_scene)
websocket_api.async_register_command(hass, websocket_save_scene)
websocket_api.async_register_command(hass, websocket_delete_scene)
websocket_api.async_register_command(hass, websocket_get_aldb)
websocket_api.async_register_command(hass, websocket_change_aldb_record)
websocket_api.async_register_command(hass, websocket_create_aldb_record)

View file

@ -0,0 +1,122 @@
"""Web socket API for Insteon scenes."""
from pyinsteon import devices
from pyinsteon.constants import ResponseStatus
from pyinsteon.managers.scene_manager import (
DeviceLinkSchema,
async_add_or_update_scene,
async_delete_scene,
async_get_scene,
async_get_scenes,
)
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant
from ..const import ID, TYPE
def _scene_to_dict(scene):
"""Return a dictionary mapping of a scene."""
device_dict = {}
for addr, links in scene["devices"].items():
str_addr = str(addr)
device_dict[str_addr] = []
for data in links:
device_dict[str_addr].append(
{
"data1": data.data1,
"data2": data.data2,
"data3": data.data3,
"has_controller": data.has_controller,
"has_responder": data.has_responder,
}
)
return {"name": scene["name"], "group": scene["group"], "devices": device_dict}
@websocket_api.websocket_command({vol.Required(TYPE): "insteon/scenes/get"})
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_get_scenes(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Get all Insteon scenes."""
scenes = await async_get_scenes(work_dir=hass.config.config_dir)
scenes_dict = {
scene_num: _scene_to_dict(scene) for scene_num, scene in scenes.items()
}
connection.send_result(msg[ID], scenes_dict)
@websocket_api.websocket_command(
{vol.Required(TYPE): "insteon/scene/get", vol.Required("scene_id"): int}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_get_scene(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Get an Insteon scene."""
scene_id = msg["scene_id"]
scene = await async_get_scene(scene_num=scene_id, work_dir=hass.config.config_dir)
connection.send_result(msg[ID], _scene_to_dict(scene))
@websocket_api.websocket_command(
{
vol.Required(TYPE): "insteon/scene/save",
vol.Required("name"): str,
vol.Required("scene_id"): int,
vol.Required("links"): DeviceLinkSchema,
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_save_scene(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Save an Insteon scene."""
scene_id = msg["scene_id"]
name = msg["name"]
links = msg["links"]
scene_id, result = await async_add_or_update_scene(
scene_num=scene_id, links=links, name=name, work_dir=hass.config.config_dir
)
await devices.async_save(workdir=hass.config.config_dir)
connection.send_result(
msg[ID], {"scene_id": scene_id, "result": result == ResponseStatus.SUCCESS}
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "insteon/scene/delete",
vol.Required("scene_id"): int,
}
)
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_delete_scene(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
) -> None:
"""Delete an Insteon scene."""
scene_id = msg["scene_id"]
result = await async_delete_scene(
scene_num=scene_id, work_dir=hass.config.config_dir
)
await devices.async_save(workdir=hass.config.config_dir)
connection.send_result(
msg[ID], {"scene_id": scene_id, "result": result == ResponseStatus.SUCCESS}
)

View file

@ -18,7 +18,7 @@
"loggers": ["pyinsteon", "pypubsub"],
"requirements": [
"pyinsteon==1.3.1",
"insteon-frontend-home-assistant==0.2.0"
"insteon-frontend-home-assistant==0.3.1"
],
"usb": [
{

View file

@ -170,6 +170,18 @@ TRIGGER_SCENE_SCHEMA = vol.Schema(
ADD_DEFAULT_LINKS_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})
SCENE_ENTITY_SCHEMA = vol.Schema(
[
{
vol.Required(CONF_ADDRESS): str,
vol.Required("data1"): int,
vol.Required("data2"): int,
vol.Required("data3"): int,
}
]
)
def normalize_byte_entry_to_int(entry: int | bytes | str):
"""Format a hex entry value."""
if isinstance(entry, int):

View file

@ -979,7 +979,7 @@ influxdb==5.3.1
inkbird-ble==0.5.6
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.2.0
insteon-frontend-home-assistant==0.3.1
# homeassistant.components.intellifire
intellifire4py==2.2.2

View file

@ -738,7 +738,7 @@ influxdb==5.3.1
inkbird-ble==0.5.6
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.2.0
insteon-frontend-home-assistant==0.3.1
# homeassistant.components.intellifire
intellifire4py==2.2.2

View file

@ -0,0 +1,547 @@
[
{
"address": "aaaaaa",
"aldb": {
"8191": {
"memory": 8191,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 0,
"target": "111111",
"data1": 0,
"data2": 26,
"data3": 0
},
"8183": {
"memory": 8183,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "111111",
"data1": 0,
"data2": 0,
"data3": 0
},
"8175": {
"memory": 8175,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 0,
"target": "333333",
"data1": 1,
"data2": 66,
"data3": 69
},
"8167": {
"memory": 8167,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 1,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8159": {
"memory": 8159,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 3,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8151": {
"memory": 8151,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 4,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8143": {
"memory": 8143,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 5,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8135": {
"memory": 8135,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 6,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8127": {
"memory": 8127,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 0,
"target": "444444",
"data1": 1,
"data2": 31,
"data3": 65
},
"8119": {
"memory": 8119,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "444444",
"data1": 0,
"data2": 0,
"data3": 0
},
"8111": {
"memory": 8111,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 20,
"target": "111111",
"data1": 0,
"data2": 0,
"data3": 0
},
"8103": {
"memory": 8103,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 20,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"8095": {
"memory": 8095,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 20,
"target": "444444",
"data1": 0,
"data2": 0,
"data3": 0
},
"8087": {
"memory": 8087,
"in_use": false,
"controller": false,
"high_water_mark": true,
"bit5": false,
"bit4": false,
"group": 0,
"target": "000000",
"data1": 0,
"data2": 0,
"data3": 0
}
},
"operating_flags": {},
"properties": {},
"read_write_mode": 2,
"first_mem_addr": 8191
},
{
"address": "111111",
"aldb": {
"4095": {
"memory": 4095,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": false,
"bit4": false,
"group": 0,
"target": "aaaaaa",
"data1": 0,
"data2": 0,
"data3": 0
},
"4087": {
"memory": 4087,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4079": {
"memory": 4079,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 20,
"target": "aaaaaa",
"data1": 255,
"data2": 28,
"data3": 1
},
"4071": {
"memory": 4071,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "333333",
"data1": 255,
"data2": 28,
"data3": 1
},
"4063": {
"memory": 4063,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "333333",
"data1": 0,
"data2": 0,
"data3": 0
},
"4055": {
"memory": 4055,
"in_use": false,
"controller": false,
"high_water_mark": true,
"bit5": false,
"bit4": false,
"group": 0,
"target": "000000",
"data1": 0,
"data2": 0,
"data3": 0
}
},
"operating_flags": {
"program_lock_on": false,
"blink_on_tx_on": false,
"resume_dim_on": false,
"key_beep_on": true,
"blink_on_error_on": false
},
"properties": {
"x10_house": 32,
"x10_unit": 0,
"ramp_rate": 28,
"on_level": 255
},
"read_write_mode": 3,
"first_mem_addr": 4095
},
{
"address": "333333",
"aldb": {
"4095": {
"memory": 4095,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 0,
"target": "aaaaaa",
"data1": 0,
"data2": 0,
"data3": 0
},
"4087": {
"memory": 4087,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4079": {
"memory": 4079,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 3,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4071": {
"memory": 4071,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 4,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4063": {
"memory": 4063,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 5,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4055": {
"memory": 4055,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 6,
"target": "aaaaaa",
"data1": 3,
"data2": 51,
"data3": 165
},
"4047": {
"memory": 4047,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 20,
"target": "aaaaaa",
"data1": 255,
"data2": 0,
"data3": 3
},
"4039": {
"memory": 4039,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "111111",
"data1": 255,
"data2": 28,
"data3": 1
},
"4031": {
"memory": 4031,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "111111",
"data1": 0,
"data2": 0,
"data3": 0
},
"4023": {
"memory": 4023,
"in_use": false,
"controller": false,
"high_water_mark": true,
"bit5": false,
"bit4": false,
"group": 0,
"target": "000000",
"data1": 0,
"data2": 0,
"data3": 0
}
},
"operating_flags": {
"program_lock_on": false,
"blink_on_tx_on": false,
"resume_dim_on": false,
"led_off": false,
"key_beep_on": false,
"rf_disable_on": false,
"powerline_disable_on": false,
"blink_on_error_off": true
},
"properties": {
"led_dimming": 51,
"non_toggle_mask": 0,
"non_toggle_on_off_mask": 0,
"trigger_group_mask": 0,
"on_mask": 0,
"off_mask": 0,
"x10_house": 32,
"x10_unit": 32,
"ramp_rate": 28,
"on_level": 255,
"on_mask_3": 0,
"off_mask_3": 0,
"x10_house_3": 32,
"x10_unit_3": 32,
"ramp_rate_3": 0,
"on_level_3": 0,
"on_mask_4": 0,
"off_mask_4": 0,
"x10_house_4": 32,
"x10_unit_4": 32,
"ramp_rate_4": 0,
"on_level_4": 0,
"on_mask_5": 0,
"off_mask_5": 0,
"x10_house_5": 32,
"x10_unit_5": 32,
"ramp_rate_5": 0,
"on_level_5": 0,
"on_mask_6": 0,
"off_mask_6": 0,
"x10_house_6": 32,
"x10_unit_6": 32,
"ramp_rate_6": 0,
"on_level_6": 0
},
"read_write_mode": 1,
"first_mem_addr": 4095
},
{
"address": "444444",
"aldb": {
"4095": {
"memory": 4095,
"in_use": false,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 0,
"target": "aaaaaa",
"data1": 0,
"data2": 0,
"data3": 0
},
"4087": {
"memory": 4087,
"in_use": true,
"controller": true,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 1,
"target": "aaaaaa",
"data1": 255,
"data2": 28,
"data3": 0
},
"4079": {
"memory": 4079,
"in_use": true,
"controller": false,
"high_water_mark": false,
"bit5": true,
"bit4": false,
"group": 20,
"target": "aaaaaa",
"data1": 255,
"data2": 28,
"data3": 1
},
"4071": {
"memory": 4071,
"in_use": false,
"controller": false,
"high_water_mark": true,
"bit5": false,
"bit4": false,
"group": 0,
"target": "000000",
"data1": 0,
"data2": 0,
"data3": 0
}
},
"operating_flags": {
"program_lock_on": false,
"blink_on_tx_on": false,
"resume_dim_on": false,
"key_beep_on": false,
"blink_on_error_on": false
},
"properties": {
"led_dimming": 0,
"x10_house": 32,
"x10_unit": 32,
"ramp_rate": 28,
"on_level": 255
},
"read_write_mode": 1,
"first_mem_addr": 4095
}
]

View file

@ -0,0 +1,184 @@
"""Test the Insteon Scenes APIs."""
import json
import os
from unittest.mock import AsyncMock, patch
from pyinsteon.constants import ResponseStatus
import pyinsteon.managers.scene_manager
import pytest
from homeassistant.components.insteon.api import async_load_api, scenes
from homeassistant.components.insteon.const import ID, TYPE
from .mock_devices import MockDevices
from tests.common import load_fixture
@pytest.fixture(name="scene_data", scope="session")
def aldb_data_fixture():
"""Load the controller state fixture data."""
return json.loads(load_fixture("insteon/scene_data.json"))
@pytest.fixture(name="remove_json")
def remove_insteon_devices_json(hass):
"""Fixture to remove insteon_devices.json at the end of the test."""
yield
file = os.path.join(hass.config.config_dir, "insteon_devices.json")
if os.path.exists(file):
os.remove(file)
def _scene_to_array(scene):
"""Convert a scene object to a dictionary."""
scene_list = []
for device, links in scene["devices"].items():
for link in links:
link_dict = {}
link_dict["address"] = device.id
link_dict["data1"] = link.data1
link_dict["data2"] = link.data2
link_dict["data3"] = link.data3
scene_list.append(link_dict)
return scene_list
async def _setup(hass, hass_ws_client, scene_data):
"""Set up tests."""
ws_client = await hass_ws_client(hass)
devices = MockDevices()
await devices.async_load()
async_load_api(hass)
for device in scene_data:
addr = device["address"]
aldb = device["aldb"]
devices.fill_aldb(addr, aldb)
return ws_client, devices
async def test_get_scenes(hass, hass_ws_client, scene_data):
"""Test getting all Insteon scenes."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
with patch.object(pyinsteon.managers.scene_manager, "devices", devices):
await ws_client.send_json({ID: 1, TYPE: "insteon/scenes/get"})
msg = await ws_client.receive_json()
result = msg["result"]
assert len(result) == 1
assert len(result["20"]) == 3
async def test_get_scene(hass, hass_ws_client, scene_data):
"""Test getting an Insteon scene."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
with patch.object(pyinsteon.managers.scene_manager, "devices", devices):
await ws_client.send_json({ID: 1, TYPE: "insteon/scene/get", "scene_id": 20})
msg = await ws_client.receive_json()
result = msg["result"]
assert len(result["devices"]) == 3
async def test_save_scene(hass, hass_ws_client, scene_data, remove_json):
"""Test saving an Insteon scene."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
mock_add_or_update_scene = AsyncMock(return_value=(20, ResponseStatus.SUCCESS))
with patch.object(
pyinsteon.managers.scene_manager, "devices", devices
), patch.object(scenes, "async_add_or_update_scene", mock_add_or_update_scene):
scene = await pyinsteon.managers.scene_manager.async_get_scene(20)
scene["devices"]["1a1a1a"] = []
links = _scene_to_array(scene)
await ws_client.send_json(
{
ID: 1,
TYPE: "insteon/scene/save",
"scene_id": 20,
"name": "Some scene name",
"links": links,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["result"]
assert result["scene_id"] == 20
async def test_save_new_scene(hass, hass_ws_client, scene_data, remove_json):
"""Test saving a new Insteon scene."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
mock_add_or_update_scene = AsyncMock(return_value=(21, ResponseStatus.SUCCESS))
with patch.object(
pyinsteon.managers.scene_manager, "devices", devices
), patch.object(scenes, "async_add_or_update_scene", mock_add_or_update_scene):
scene = await pyinsteon.managers.scene_manager.async_get_scene(20)
scene["devices"]["1a1a1a"] = []
links = _scene_to_array(scene)
await ws_client.send_json(
{
ID: 1,
TYPE: "insteon/scene/save",
"scene_id": -1,
"name": "Some scene name",
"links": links,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["result"]
assert result["scene_id"] == 21
async def test_save_scene_error(hass, hass_ws_client, scene_data, remove_json):
"""Test saving an Insteon scene with error."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
mock_add_or_update_scene = AsyncMock(return_value=(20, ResponseStatus.FAILURE))
with patch.object(
pyinsteon.managers.scene_manager, "devices", devices
), patch.object(scenes, "async_add_or_update_scene", mock_add_or_update_scene):
scene = await pyinsteon.managers.scene_manager.async_get_scene(20)
scene["devices"]["1a1a1a"] = []
links = _scene_to_array(scene)
await ws_client.send_json(
{
ID: 1,
TYPE: "insteon/scene/save",
"scene_id": 20,
"name": "Some scene name",
"links": links,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert not result["result"]
assert result["scene_id"] == 20
async def test_delete_scene(hass, hass_ws_client, scene_data, remove_json):
"""Test delete an Insteon scene."""
ws_client, devices = await _setup(hass, hass_ws_client, scene_data)
mock_delete_scene = AsyncMock(return_value=ResponseStatus.SUCCESS)
with patch.object(
pyinsteon.managers.scene_manager, "devices", devices
), patch.object(scenes, "async_delete_scene", mock_delete_scene):
await ws_client.send_json(
{
ID: 1,
TYPE: "insteon/scene/delete",
"scene_id": 20,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["result"]
assert result["scene_id"] == 20