719 lines
23 KiB
Python
719 lines
23 KiB
Python
"""Support for INSTEON Modems (PLM and Hub)."""
|
|
import collections
|
|
import logging
|
|
from typing import Dict
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.const import (
|
|
CONF_ADDRESS,
|
|
CONF_ENTITY_ID,
|
|
CONF_HOST,
|
|
CONF_PLATFORM,
|
|
CONF_PORT,
|
|
ENTITY_MATCH_ALL,
|
|
EVENT_HOMEASSISTANT_STOP,
|
|
)
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers import discovery
|
|
from homeassistant.helpers.dispatcher import dispatcher_send, async_dispatcher_connect
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.entity import Entity
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DOMAIN = "insteon"
|
|
INSTEON_ENTITIES = "entities"
|
|
|
|
CONF_IP_PORT = "ip_port"
|
|
CONF_HUB_USERNAME = "username"
|
|
CONF_HUB_PASSWORD = "password"
|
|
CONF_HUB_VERSION = "hub_version"
|
|
CONF_OVERRIDE = "device_override"
|
|
CONF_PLM_HUB_MSG = "Must configure either a PLM port or a Hub host"
|
|
CONF_CAT = "cat"
|
|
CONF_SUBCAT = "subcat"
|
|
CONF_FIRMWARE = "firmware"
|
|
CONF_PRODUCT_KEY = "product_key"
|
|
CONF_X10 = "x10_devices"
|
|
CONF_HOUSECODE = "housecode"
|
|
CONF_UNITCODE = "unitcode"
|
|
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"
|
|
|
|
SRV_ADD_ALL_LINK = "add_all_link"
|
|
SRV_DEL_ALL_LINK = "delete_all_link"
|
|
SRV_LOAD_ALDB = "load_all_link_database"
|
|
SRV_PRINT_ALDB = "print_all_link_database"
|
|
SRV_PRINT_IM_ALDB = "print_im_all_link_database"
|
|
SRV_X10_ALL_UNITS_OFF = "x10_all_units_off"
|
|
SRV_X10_ALL_LIGHTS_OFF = "x10_all_lights_off"
|
|
SRV_X10_ALL_LIGHTS_ON = "x10_all_lights_on"
|
|
SRV_ALL_LINK_GROUP = "group"
|
|
SRV_ALL_LINK_MODE = "mode"
|
|
SRV_LOAD_DB_RELOAD = "reload"
|
|
SRV_CONTROLLER = "controller"
|
|
SRV_RESPONDER = "responder"
|
|
SRV_HOUSECODE = "housecode"
|
|
SRV_SCENE_ON = "scene_on"
|
|
SRV_SCENE_OFF = "scene_off"
|
|
|
|
SIGNAL_LOAD_ALDB = "load_aldb"
|
|
SIGNAL_PRINT_ALDB = "print_aldb"
|
|
|
|
HOUSECODES = [
|
|
"a",
|
|
"b",
|
|
"c",
|
|
"d",
|
|
"e",
|
|
"f",
|
|
"g",
|
|
"h",
|
|
"i",
|
|
"j",
|
|
"k",
|
|
"l",
|
|
"m",
|
|
"n",
|
|
"o",
|
|
"p",
|
|
]
|
|
|
|
BUTTON_PRESSED_STATE_NAME = "onLevelButton"
|
|
EVENT_BUTTON_ON = "insteon.button_on"
|
|
EVENT_BUTTON_OFF = "insteon.button_off"
|
|
EVENT_CONF_BUTTON = "button"
|
|
|
|
|
|
def set_default_port(schema: Dict) -> Dict:
|
|
"""Set the default port based on the Hub version."""
|
|
# If the ip_port is found do nothing
|
|
# If it is not found the set the default
|
|
ip_port = schema.get(CONF_IP_PORT)
|
|
if not ip_port:
|
|
hub_version = schema.get(CONF_HUB_VERSION)
|
|
# Found hub_version but not ip_port
|
|
if hub_version == 1:
|
|
schema[CONF_IP_PORT] = 9761
|
|
else:
|
|
schema[CONF_IP_PORT] = 25105
|
|
return schema
|
|
|
|
|
|
CONF_DEVICE_OVERRIDE_SCHEMA = vol.All(
|
|
cv.deprecated(CONF_PLATFORM),
|
|
vol.Schema(
|
|
{
|
|
vol.Required(CONF_ADDRESS): cv.string,
|
|
vol.Optional(CONF_CAT): cv.byte,
|
|
vol.Optional(CONF_SUBCAT): cv.byte,
|
|
vol.Optional(CONF_FIRMWARE): cv.byte,
|
|
vol.Optional(CONF_PRODUCT_KEY): cv.byte,
|
|
vol.Optional(CONF_PLATFORM): cv.string,
|
|
}
|
|
),
|
|
)
|
|
|
|
|
|
CONF_X10_SCHEMA = vol.All(
|
|
vol.Schema(
|
|
{
|
|
vol.Required(CONF_HOUSECODE): cv.string,
|
|
vol.Required(CONF_UNITCODE): vol.Range(min=1, max=16),
|
|
vol.Required(CONF_PLATFORM): cv.string,
|
|
vol.Optional(CONF_DIM_STEPS): vol.Range(min=2, max=255),
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
CONFIG_SCHEMA = vol.Schema(
|
|
{
|
|
DOMAIN: vol.All(
|
|
vol.Schema(
|
|
{
|
|
vol.Exclusive(
|
|
CONF_PORT, "plm_or_hub", msg=CONF_PLM_HUB_MSG
|
|
): cv.string,
|
|
vol.Exclusive(
|
|
CONF_HOST, "plm_or_hub", msg=CONF_PLM_HUB_MSG
|
|
): cv.string,
|
|
vol.Optional(CONF_IP_PORT): cv.port,
|
|
vol.Optional(CONF_HUB_USERNAME): cv.string,
|
|
vol.Optional(CONF_HUB_PASSWORD): cv.string,
|
|
vol.Optional(CONF_HUB_VERSION, default=2): vol.In([1, 2]),
|
|
vol.Optional(CONF_OVERRIDE): vol.All(
|
|
cv.ensure_list_csv, [CONF_DEVICE_OVERRIDE_SCHEMA]
|
|
),
|
|
vol.Optional(CONF_X10_ALL_UNITS_OFF): vol.In(HOUSECODES),
|
|
vol.Optional(CONF_X10_ALL_LIGHTS_ON): vol.In(HOUSECODES),
|
|
vol.Optional(CONF_X10_ALL_LIGHTS_OFF): vol.In(HOUSECODES),
|
|
vol.Optional(CONF_X10): vol.All(
|
|
cv.ensure_list_csv, [CONF_X10_SCHEMA]
|
|
),
|
|
},
|
|
extra=vol.ALLOW_EXTRA,
|
|
required=True,
|
|
),
|
|
cv.has_at_least_one_key(CONF_PORT, CONF_HOST),
|
|
set_default_port,
|
|
)
|
|
},
|
|
extra=vol.ALLOW_EXTRA,
|
|
)
|
|
|
|
|
|
ADD_ALL_LINK_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(SRV_ALL_LINK_GROUP): vol.Range(min=0, max=255),
|
|
vol.Required(SRV_ALL_LINK_MODE): vol.In([SRV_CONTROLLER, SRV_RESPONDER]),
|
|
}
|
|
)
|
|
|
|
|
|
DEL_ALL_LINK_SCHEMA = vol.Schema(
|
|
{vol.Required(SRV_ALL_LINK_GROUP): vol.Range(min=0, max=255)}
|
|
)
|
|
|
|
|
|
LOAD_ALDB_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(CONF_ENTITY_ID): vol.Any(cv.entity_id, ENTITY_MATCH_ALL),
|
|
vol.Optional(SRV_LOAD_DB_RELOAD, default=False): cv.boolean,
|
|
}
|
|
)
|
|
|
|
|
|
PRINT_ALDB_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})
|
|
|
|
|
|
X10_HOUSECODE_SCHEMA = vol.Schema({vol.Required(SRV_HOUSECODE): vol.In(HOUSECODES)})
|
|
|
|
|
|
TRIGGER_SCENE_SCHEMA = vol.Schema(
|
|
{vol.Required(SRV_ALL_LINK_GROUP): vol.Range(min=0, max=255)}
|
|
)
|
|
|
|
|
|
STATE_NAME_LABEL_MAP = {
|
|
"keypadButtonA": "Button A",
|
|
"keypadButtonB": "Button B",
|
|
"keypadButtonC": "Button C",
|
|
"keypadButtonD": "Button D",
|
|
"keypadButtonE": "Button E",
|
|
"keypadButtonF": "Button F",
|
|
"keypadButtonG": "Button G",
|
|
"keypadButtonH": "Button H",
|
|
"keypadButtonMain": "Main",
|
|
"onOffButtonA": "Button A",
|
|
"onOffButtonB": "Button B",
|
|
"onOffButtonC": "Button C",
|
|
"onOffButtonD": "Button D",
|
|
"onOffButtonE": "Button E",
|
|
"onOffButtonF": "Button F",
|
|
"onOffButtonG": "Button G",
|
|
"onOffButtonH": "Button H",
|
|
"onOffButtonMain": "Main",
|
|
"fanOnLevel": "Fan",
|
|
"lightOnLevel": "Light",
|
|
"coolSetPoint": "Cool Set",
|
|
"heatSetPoint": "HeatSet",
|
|
"statusReport": "Status",
|
|
"generalSensor": "Sensor",
|
|
"motionSensor": "Motion",
|
|
"lightSensor": "Light",
|
|
"batterySensor": "Battery",
|
|
"dryLeakSensor": "Dry",
|
|
"wetLeakSensor": "Wet",
|
|
"heartbeatLeakSensor": "Heartbeat",
|
|
"openClosedRelay": "Relay",
|
|
"openClosedSensor": "Sensor",
|
|
"lightOnOff": "Light",
|
|
"outletTopOnOff": "Top",
|
|
"outletBottomOnOff": "Bottom",
|
|
"coverOpenLevel": "Cover",
|
|
}
|
|
|
|
|
|
async def async_setup(hass, config):
|
|
"""Set up the connection to the modem."""
|
|
import insteonplm
|
|
|
|
ipdb = IPDB()
|
|
insteon_modem = None
|
|
|
|
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)
|
|
overrides = conf.get(CONF_OVERRIDE, [])
|
|
x10_devices = conf.get(CONF_X10, [])
|
|
x10_all_units_off_housecode = conf.get(CONF_X10_ALL_UNITS_OFF)
|
|
x10_all_lights_on_housecode = conf.get(CONF_X10_ALL_LIGHTS_ON)
|
|
x10_all_lights_off_housecode = conf.get(CONF_X10_ALL_LIGHTS_OFF)
|
|
|
|
@callback
|
|
def async_new_insteon_device(device):
|
|
"""Detect device from transport to be delegated to platform."""
|
|
for state_key in device.states:
|
|
platform_info = ipdb[device.states[state_key]]
|
|
if platform_info and platform_info.platform:
|
|
platform = platform_info.platform
|
|
|
|
if platform == "on_off_events":
|
|
device.states[state_key].register_updates(_fire_button_on_off_event)
|
|
|
|
else:
|
|
_LOGGER.info(
|
|
"New INSTEON device: %s (%s) %s",
|
|
device.address,
|
|
device.states[state_key].name,
|
|
platform,
|
|
)
|
|
|
|
hass.async_create_task(
|
|
discovery.async_load_platform(
|
|
hass,
|
|
platform,
|
|
DOMAIN,
|
|
discovered={
|
|
"address": device.address.id,
|
|
"state_key": state_key,
|
|
},
|
|
hass_config=config,
|
|
)
|
|
)
|
|
|
|
def add_all_link(service):
|
|
"""Add an INSTEON All-Link between two devices."""
|
|
group = service.data.get(SRV_ALL_LINK_GROUP)
|
|
mode = service.data.get(SRV_ALL_LINK_MODE)
|
|
link_mode = 1 if mode.lower() == SRV_CONTROLLER else 0
|
|
insteon_modem.start_all_linking(link_mode, group)
|
|
|
|
def del_all_link(service):
|
|
"""Delete an INSTEON All-Link between two devices."""
|
|
group = service.data.get(SRV_ALL_LINK_GROUP)
|
|
insteon_modem.start_all_linking(255, group)
|
|
|
|
def load_aldb(service):
|
|
"""Load the device All-Link database."""
|
|
entity_id = service.data[CONF_ENTITY_ID]
|
|
reload = service.data[SRV_LOAD_DB_RELOAD]
|
|
if entity_id.lower() == ENTITY_MATCH_ALL:
|
|
for entity_id in hass.data[DOMAIN].get(INSTEON_ENTITIES):
|
|
_send_load_aldb_signal(entity_id, reload)
|
|
else:
|
|
_send_load_aldb_signal(entity_id, reload)
|
|
|
|
def _send_load_aldb_signal(entity_id, reload):
|
|
"""Send the load All-Link database signal to INSTEON entity."""
|
|
signal = f"{entity_id}_{SIGNAL_LOAD_ALDB}"
|
|
dispatcher_send(hass, signal, reload)
|
|
|
|
def print_aldb(service):
|
|
"""Print the All-Link Database for a device."""
|
|
# For now this sends logs to the log file.
|
|
# Furture direction is to create an INSTEON control panel.
|
|
entity_id = service.data[CONF_ENTITY_ID]
|
|
signal = f"{entity_id}_{SIGNAL_PRINT_ALDB}"
|
|
dispatcher_send(hass, signal)
|
|
|
|
def print_im_aldb(service):
|
|
"""Print the All-Link Database for a device."""
|
|
# For now this sends logs to the log file.
|
|
# Furture direction is to create an INSTEON control panel.
|
|
print_aldb_to_log(insteon_modem.aldb)
|
|
|
|
def x10_all_units_off(service):
|
|
"""Send the X10 All Units Off command."""
|
|
housecode = service.data.get(SRV_HOUSECODE)
|
|
insteon_modem.x10_all_units_off(housecode)
|
|
|
|
def x10_all_lights_off(service):
|
|
"""Send the X10 All Lights Off command."""
|
|
housecode = service.data.get(SRV_HOUSECODE)
|
|
insteon_modem.x10_all_lights_off(housecode)
|
|
|
|
def x10_all_lights_on(service):
|
|
"""Send the X10 All Lights On command."""
|
|
housecode = service.data.get(SRV_HOUSECODE)
|
|
insteon_modem.x10_all_lights_on(housecode)
|
|
|
|
def scene_on(service):
|
|
"""Trigger an INSTEON scene ON."""
|
|
group = service.data.get(SRV_ALL_LINK_GROUP)
|
|
insteon_modem.trigger_group_on(group)
|
|
|
|
def scene_off(service):
|
|
"""Trigger an INSTEON scene ON."""
|
|
group = service.data.get(SRV_ALL_LINK_GROUP)
|
|
insteon_modem.trigger_group_off(group)
|
|
|
|
def _register_services():
|
|
hass.services.register(
|
|
DOMAIN, SRV_ADD_ALL_LINK, add_all_link, schema=ADD_ALL_LINK_SCHEMA
|
|
)
|
|
hass.services.register(
|
|
DOMAIN, SRV_DEL_ALL_LINK, del_all_link, schema=DEL_ALL_LINK_SCHEMA
|
|
)
|
|
hass.services.register(
|
|
DOMAIN, SRV_LOAD_ALDB, load_aldb, schema=LOAD_ALDB_SCHEMA
|
|
)
|
|
hass.services.register(
|
|
DOMAIN, SRV_PRINT_ALDB, print_aldb, schema=PRINT_ALDB_SCHEMA
|
|
)
|
|
hass.services.register(DOMAIN, SRV_PRINT_IM_ALDB, print_im_aldb, schema=None)
|
|
hass.services.register(
|
|
DOMAIN,
|
|
SRV_X10_ALL_UNITS_OFF,
|
|
x10_all_units_off,
|
|
schema=X10_HOUSECODE_SCHEMA,
|
|
)
|
|
hass.services.register(
|
|
DOMAIN,
|
|
SRV_X10_ALL_LIGHTS_OFF,
|
|
x10_all_lights_off,
|
|
schema=X10_HOUSECODE_SCHEMA,
|
|
)
|
|
hass.services.register(
|
|
DOMAIN,
|
|
SRV_X10_ALL_LIGHTS_ON,
|
|
x10_all_lights_on,
|
|
schema=X10_HOUSECODE_SCHEMA,
|
|
)
|
|
hass.services.register(
|
|
DOMAIN, SRV_SCENE_ON, scene_on, schema=TRIGGER_SCENE_SCHEMA
|
|
)
|
|
hass.services.register(
|
|
DOMAIN, SRV_SCENE_OFF, scene_off, schema=TRIGGER_SCENE_SCHEMA
|
|
)
|
|
_LOGGER.debug("Insteon Services registered")
|
|
|
|
def _fire_button_on_off_event(address, group, val):
|
|
# Firing an event when a button is pressed.
|
|
device = insteon_modem.devices[address.hex]
|
|
state_name = device.states[group].name
|
|
button = (
|
|
"" if state_name == BUTTON_PRESSED_STATE_NAME else state_name[-1].lower()
|
|
)
|
|
schema = {CONF_ADDRESS: address.hex}
|
|
if button != "":
|
|
schema[EVENT_CONF_BUTTON] = button
|
|
if val:
|
|
event = EVENT_BUTTON_ON
|
|
else:
|
|
event = EVENT_BUTTON_OFF
|
|
_LOGGER.debug(
|
|
"Firing event %s with address %s and button %s", event, address.hex, button
|
|
)
|
|
hass.bus.fire(event, schema)
|
|
|
|
if host:
|
|
_LOGGER.info("Connecting to Insteon Hub on %s", host)
|
|
conn = await insteonplm.Connection.create(
|
|
host=host,
|
|
port=ip_port,
|
|
username=username,
|
|
password=password,
|
|
hub_version=hub_version,
|
|
loop=hass.loop,
|
|
workdir=hass.config.config_dir,
|
|
)
|
|
else:
|
|
_LOGGER.info("Looking for Insteon PLM on %s", port)
|
|
conn = await insteonplm.Connection.create(
|
|
device=port, loop=hass.loop, workdir=hass.config.config_dir
|
|
)
|
|
|
|
insteon_modem = conn.protocol
|
|
|
|
for device_override in overrides:
|
|
#
|
|
# Override the device default capabilities for a specific address
|
|
#
|
|
address = device_override.get("address")
|
|
for prop in device_override:
|
|
if prop in [CONF_CAT, CONF_SUBCAT]:
|
|
insteon_modem.devices.add_override(address, prop, device_override[prop])
|
|
elif prop in [CONF_FIRMWARE, CONF_PRODUCT_KEY]:
|
|
insteon_modem.devices.add_override(
|
|
address, CONF_PRODUCT_KEY, device_override[prop]
|
|
)
|
|
|
|
hass.data[DOMAIN] = {}
|
|
hass.data[DOMAIN]["modem"] = insteon_modem
|
|
hass.data[DOMAIN][INSTEON_ENTITIES] = {}
|
|
|
|
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, conn.close)
|
|
|
|
insteon_modem.devices.add_device_callback(async_new_insteon_device)
|
|
|
|
if x10_all_units_off_housecode:
|
|
device = insteon_modem.add_x10_device(
|
|
x10_all_units_off_housecode, 20, "allunitsoff"
|
|
)
|
|
if x10_all_lights_on_housecode:
|
|
device = insteon_modem.add_x10_device(
|
|
x10_all_lights_on_housecode, 21, "alllightson"
|
|
)
|
|
if x10_all_lights_off_housecode:
|
|
device = insteon_modem.add_x10_device(
|
|
x10_all_lights_off_housecode, 22, "alllightsoff"
|
|
)
|
|
for device in x10_devices:
|
|
housecode = device.get(CONF_HOUSECODE)
|
|
unitcode = device.get(CONF_UNITCODE)
|
|
x10_type = "onoff"
|
|
steps = device.get(CONF_DIM_STEPS, 22)
|
|
if device.get(CONF_PLATFORM) == "light":
|
|
x10_type = "dimmable"
|
|
elif device.get(CONF_PLATFORM) == "binary_sensor":
|
|
x10_type = "sensor"
|
|
_LOGGER.debug(
|
|
"Adding X10 device to Insteon: %s %d %s", housecode, unitcode, x10_type
|
|
)
|
|
device = insteon_modem.add_x10_device(housecode, unitcode, x10_type)
|
|
if device and hasattr(device.states[0x01], "steps"):
|
|
device.states[0x01].steps = steps
|
|
|
|
hass.async_add_job(_register_services)
|
|
|
|
return True
|
|
|
|
|
|
State = collections.namedtuple("Product", "stateType platform")
|
|
|
|
|
|
class IPDB:
|
|
"""Embodies the INSTEON Product Database static data and access methods."""
|
|
|
|
def __init__(self):
|
|
"""Create the INSTEON Product Database (IPDB)."""
|
|
from insteonplm.states.cover import Cover
|
|
|
|
from insteonplm.states.onOff import (
|
|
OnOffSwitch,
|
|
OnOffSwitch_OutletTop,
|
|
OnOffSwitch_OutletBottom,
|
|
OpenClosedRelay,
|
|
OnOffKeypadA,
|
|
OnOffKeypad,
|
|
)
|
|
|
|
from insteonplm.states.dimmable import (
|
|
DimmableSwitch,
|
|
DimmableSwitch_Fan,
|
|
DimmableRemote,
|
|
DimmableKeypadA,
|
|
)
|
|
|
|
from insteonplm.states.sensor import (
|
|
VariableSensor,
|
|
OnOffSensor,
|
|
SmokeCO2Sensor,
|
|
IoLincSensor,
|
|
LeakSensorDryWet,
|
|
)
|
|
|
|
from insteonplm.states.x10 import (
|
|
X10DimmableSwitch,
|
|
X10OnOffSwitch,
|
|
X10OnOffSensor,
|
|
X10AllUnitsOffSensor,
|
|
X10AllLightsOnSensor,
|
|
X10AllLightsOffSensor,
|
|
)
|
|
|
|
self.states = [
|
|
State(Cover, "cover"),
|
|
State(OnOffSwitch_OutletTop, "switch"),
|
|
State(OnOffSwitch_OutletBottom, "switch"),
|
|
State(OpenClosedRelay, "switch"),
|
|
State(OnOffSwitch, "switch"),
|
|
State(OnOffKeypadA, "switch"),
|
|
State(OnOffKeypad, "switch"),
|
|
State(LeakSensorDryWet, "binary_sensor"),
|
|
State(IoLincSensor, "binary_sensor"),
|
|
State(SmokeCO2Sensor, "sensor"),
|
|
State(OnOffSensor, "binary_sensor"),
|
|
State(VariableSensor, "sensor"),
|
|
State(DimmableSwitch_Fan, "fan"),
|
|
State(DimmableSwitch, "light"),
|
|
State(DimmableRemote, "on_off_events"),
|
|
State(DimmableKeypadA, "light"),
|
|
State(X10DimmableSwitch, "light"),
|
|
State(X10OnOffSwitch, "switch"),
|
|
State(X10OnOffSensor, "binary_sensor"),
|
|
State(X10AllUnitsOffSensor, "binary_sensor"),
|
|
State(X10AllLightsOnSensor, "binary_sensor"),
|
|
State(X10AllLightsOffSensor, "binary_sensor"),
|
|
]
|
|
|
|
def __len__(self):
|
|
"""Return the number of INSTEON state types mapped to HA platforms."""
|
|
return len(self.states)
|
|
|
|
def __iter__(self):
|
|
"""Itterate through the INSTEON state types to HA platforms."""
|
|
for product in self.states:
|
|
yield product
|
|
|
|
def __getitem__(self, key):
|
|
"""Return a Home Assistant platform from an INSTEON state type."""
|
|
for state in self.states:
|
|
if isinstance(key, state.stateType):
|
|
return state
|
|
return None
|
|
|
|
|
|
class InsteonEntity(Entity):
|
|
"""INSTEON abstract base entity."""
|
|
|
|
def __init__(self, device, state_key):
|
|
"""Initialize the INSTEON binary sensor."""
|
|
self._insteon_device_state = device.states[state_key]
|
|
self._insteon_device = device
|
|
self._insteon_device.aldb.add_loaded_callback(self._aldb_loaded)
|
|
|
|
@property
|
|
def should_poll(self):
|
|
"""No polling needed."""
|
|
return False
|
|
|
|
@property
|
|
def address(self):
|
|
"""Return the address of the node."""
|
|
return self._insteon_device.address.human
|
|
|
|
@property
|
|
def group(self):
|
|
"""Return the INSTEON group that the entity responds to."""
|
|
return self._insteon_device_state.group
|
|
|
|
@property
|
|
def unique_id(self) -> str:
|
|
"""Return a unique ID."""
|
|
if self._insteon_device_state.group == 0x01:
|
|
uid = self._insteon_device.id
|
|
else:
|
|
uid = "{:s}_{:d}".format(
|
|
self._insteon_device.id, self._insteon_device_state.group
|
|
)
|
|
return uid
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the node (used for Entity_ID)."""
|
|
# Set a base description
|
|
description = self._insteon_device.description
|
|
if self._insteon_device.description is None:
|
|
description = "Unknown Device"
|
|
|
|
# Get an extension label if there is one
|
|
extension = self._get_label()
|
|
if extension:
|
|
extension = " " + extension
|
|
name = "{:s} {:s}{:s}".format(
|
|
description, self._insteon_device.address.human, extension
|
|
)
|
|
return name
|
|
|
|
@property
|
|
def device_state_attributes(self):
|
|
"""Provide attributes for display on device card."""
|
|
attributes = {"INSTEON Address": self.address, "INSTEON Group": self.group}
|
|
return attributes
|
|
|
|
@callback
|
|
def async_entity_update(self, deviceid, group, val):
|
|
"""Receive notification from transport that new data exists."""
|
|
_LOGGER.debug(
|
|
"Received update for device %s group %d value %s",
|
|
deviceid.human,
|
|
group,
|
|
val,
|
|
)
|
|
self.async_schedule_update_ha_state()
|
|
|
|
async def async_added_to_hass(self):
|
|
"""Register INSTEON update events."""
|
|
_LOGGER.debug(
|
|
"Tracking updates for device %s group %d statename %s",
|
|
self.address,
|
|
self.group,
|
|
self._insteon_device_state.name,
|
|
)
|
|
self._insteon_device_state.register_updates(self.async_entity_update)
|
|
self.hass.data[DOMAIN][INSTEON_ENTITIES][self.entity_id] = self
|
|
load_signal = f"{self.entity_id}_{SIGNAL_LOAD_ALDB}"
|
|
async_dispatcher_connect(self.hass, load_signal, self._load_aldb)
|
|
print_signal = f"{self.entity_id}_{SIGNAL_PRINT_ALDB}"
|
|
async_dispatcher_connect(self.hass, print_signal, self._print_aldb)
|
|
|
|
def _load_aldb(self, reload=False):
|
|
"""Load the device All-Link Database."""
|
|
if reload:
|
|
self._insteon_device.aldb.clear()
|
|
self._insteon_device.read_aldb()
|
|
|
|
def _print_aldb(self):
|
|
"""Print the device ALDB to the log file."""
|
|
print_aldb_to_log(self._insteon_device.aldb)
|
|
|
|
@callback
|
|
def _aldb_loaded(self):
|
|
"""All-Link Database loaded for the device."""
|
|
self._print_aldb()
|
|
|
|
def _get_label(self):
|
|
"""Get the device label for grouped devices."""
|
|
label = ""
|
|
if len(self._insteon_device.states) > 1:
|
|
if self._insteon_device_state.name in STATE_NAME_LABEL_MAP:
|
|
label = STATE_NAME_LABEL_MAP[self._insteon_device_state.name]
|
|
else:
|
|
label = f"Group {self.group:d}"
|
|
return label
|
|
|
|
|
|
def print_aldb_to_log(aldb):
|
|
"""Print the All-Link Database to the log file."""
|
|
from insteonplm.devices import ALDBStatus
|
|
|
|
_LOGGER.info("ALDB load status is %s", aldb.status.name)
|
|
if aldb.status not in [ALDBStatus.LOADED, ALDBStatus.PARTIAL]:
|
|
_LOGGER.warning("Device All-Link database not loaded")
|
|
_LOGGER.warning("Use service insteon.load_aldb first")
|
|
return
|
|
|
|
_LOGGER.info("RecID In Use Mode HWM Group Address Data 1 Data 2 Data 3")
|
|
_LOGGER.info("----- ------ ---- --- ----- -------- ------ ------ ------")
|
|
for mem_addr in aldb:
|
|
rec = aldb[mem_addr]
|
|
# For now we write this to the log
|
|
# Roadmap is to create a configuration panel
|
|
in_use = "Y" if rec.control_flags.is_in_use else "N"
|
|
mode = "C" if rec.control_flags.is_controller else "R"
|
|
hwm = "Y" if rec.control_flags.is_high_water_mark else "N"
|
|
_LOGGER.info(
|
|
" {:04x} {:s} {:s} {:s} {:3d} {:s}"
|
|
" {:3d} {:3d} {:3d}".format(
|
|
rec.mem_addr,
|
|
in_use,
|
|
mode,
|
|
hwm,
|
|
rec.group,
|
|
rec.address.human,
|
|
rec.data1,
|
|
rec.data2,
|
|
rec.data3,
|
|
)
|
|
)
|