evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reorganize insteon code (#31183)

Browse source 
* Reorganize code

* Code review
This commit is contained in:
Tom Harris 2020-01-30 04:47:44 -05:00 • committed by GitHub
parent ea666248ce
commit 7ff30fe29d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 740 additions and 624 deletions
Download patch file Download diff file Expand all files Collapse all files

649
homeassistant/components/insteon/__init__.py
View file

@ -1,278 +1,43 @@
"""Support for INSTEON Modems (PLM and Hub)."""
import collections
import logging
from typing import Dict
import insteonplm
from insteonplm.devices import ALDBStatus
from insteonplm.states.cover import Cover
from insteonplm.states.dimmable import (
DimmableKeypadA,
DimmableRemote,
DimmableSwitch,
DimmableSwitch_Fan,
)
from insteonplm.states.onOff import (
OnOffKeypad,
OnOffKeypadA,
OnOffSwitch,
OnOffSwitch_OutletBottom,
OnOffSwitch_OutletTop,
OpenClosedRelay,
)
from insteonplm.states.sensor import (
IoLincSensor,
LeakSensorDryWet,
OnOffSensor,
SmokeCO2Sensor,
VariableSensor,
)
from insteonplm.states.x10 import (
X10AllLightsOffSensor,
X10AllLightsOnSensor,
X10AllUnitsOffSensor,
X10DimmableSwitch,
X10OnOffSensor,
X10OnOffSwitch,
)
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
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect, dispatcher_send
from homeassistant.helpers.entity import Entity
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_FIRMWARE,
CONF_HOUSECODE,
CONF_HUB_PASSWORD,
CONF_HUB_USERNAME,
CONF_HUB_VERSION,
CONF_IP_PORT,
CONF_OVERRIDE,
CONF_PRODUCT_KEY,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
CONF_X10_ALL_LIGHTS_OFF,
CONF_X10_ALL_LIGHTS_ON,
CONF_X10_ALL_UNITS_OFF,
DOMAIN,
INSTEON_ENTITIES,
)
from .schemas import CONFIG_SCHEMA # noqa F440
from .utils import async_register_services, register_new_device_callback
_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."""
ipdb = IPDB()
insteon_modem = None
conf = config[DOMAIN]
@ -288,163 +53,6 @@ async def async_setup(hass, config):
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(
@ -464,6 +72,14 @@ async def async_setup(hass, config):
insteon_modem = conn.protocol
hass.data[DOMAIN] = {}
hass.data[DOMAIN]["modem"] = insteon_modem
hass.data[DOMAIN][INSTEON_ENTITIES] = set()
register_new_device_callback(hass, config, insteon_modem)
async_register_services(hass, config, insteon_modem)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, conn.close)
for device_override in overrides:
#
# Override the device default capabilities for a specific address
@ -477,14 +93,6 @@ async def async_setup(hass, config):
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"
@ -513,199 +121,4 @@ async def async_setup(hass, config):
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)."""
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 = f" {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."""
_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,
)
)

2
homeassistant/components/insteon/binary_sensor.py
View file

@ -3,7 +3,7 @@ import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

106
homeassistant/components/insteon/const.py Normal file
View file

@ -0,0 +1,106 @@
"""Constants used by insteon component."""
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"
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",
}

2
homeassistant/components/insteon/cover.py
View file

@ -10,7 +10,7 @@ from homeassistant.components.cover import (
CoverDevice,
)
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

2
homeassistant/components/insteon/fan.py
View file

@ -11,7 +11,7 @@ from homeassistant.components.fan import (
)
from homeassistant.const import STATE_OFF
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

123
homeassistant/components/insteon/insteon_entity.py Normal file
View file

@ -0,0 +1,123 @@
"""Insteon base entity."""
import logging
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from .const import (
DOMAIN,
INSTEON_ENTITIES,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
STATE_NAME_LABEL_MAP,
)
from .utils import print_aldb_to_log
_LOGGER = logging.getLogger(__name__)
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 = f"{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 = f" {extension}"
name = f"{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].add(self.entity_id)
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

82
homeassistant/components/insteon/ipdb.py Normal file
View file

@ -0,0 +1,82 @@
"""Insteon product database."""
import collections
from insteonplm.states.cover import Cover
from insteonplm.states.dimmable import (
DimmableKeypadA,
DimmableRemote,
DimmableSwitch,
DimmableSwitch_Fan,
)
from insteonplm.states.onOff import (
OnOffKeypad,
OnOffKeypadA,
OnOffSwitch,
OnOffSwitch_OutletBottom,
OnOffSwitch_OutletTop,
OpenClosedRelay,
)
from insteonplm.states.sensor import (
IoLincSensor,
LeakSensorDryWet,
OnOffSensor,
SmokeCO2Sensor,
VariableSensor,
)
from insteonplm.states.x10 import (
X10AllLightsOffSensor,
X10AllLightsOnSensor,
X10AllUnitsOffSensor,
X10DimmableSwitch,
X10OnOffSensor,
X10OnOffSwitch,
)
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)."""
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

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

@ -3,7 +3,7 @@ import logging
from homeassistant.components.light import ATTR_BRIGHTNESS, SUPPORT_BRIGHTNESS, Light
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

153
homeassistant/components/insteon/schemas.py Normal file
View file

@ -0,0 +1,153 @@
"""Schemas used by insteon component."""
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,
)
import homeassistant.helpers.config_validation as cv
from .const import (
CONF_CAT,
CONF_DIM_STEPS,
CONF_FIRMWARE,
CONF_HOUSECODE,
CONF_HUB_PASSWORD,
CONF_HUB_USERNAME,
CONF_HUB_VERSION,
CONF_IP_PORT,
CONF_OVERRIDE,
CONF_PLM_HUB_MSG,
CONF_PRODUCT_KEY,
CONF_SUBCAT,
CONF_UNITCODE,
CONF_X10,
CONF_X10_ALL_LIGHTS_OFF,
CONF_X10_ALL_LIGHTS_ON,
CONF_X10_ALL_UNITS_OFF,
DOMAIN,
HOUSECODES,
SRV_ALL_LINK_GROUP,
SRV_ALL_LINK_MODE,
SRV_CONTROLLER,
SRV_HOUSECODE,
SRV_LOAD_DB_RELOAD,
SRV_RESPONDER,
)
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)}
)

2
homeassistant/components/insteon/sensor.py
View file

@ -3,7 +3,7 @@ import logging
from homeassistant.helpers.entity import Entity
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

2
homeassistant/components/insteon/switch.py
View file

@ -3,7 +3,7 @@ import logging
from homeassistant.components.switch import SwitchDevice
from . import InsteonEntity
from .insteon_entity import InsteonEntity
_LOGGER = logging.getLogger(__name__)

239
homeassistant/components/insteon/utils.py Normal file
View file

@ -0,0 +1,239 @@
"""Utilities used by insteon component."""
import logging
from insteonplm.devices import ALDBStatus
from homeassistant.const import CONF_ADDRESS, CONF_ENTITY_ID, ENTITY_MATCH_ALL
from homeassistant.core import callback
from homeassistant.helpers import discovery
from homeassistant.helpers.dispatcher import dispatcher_send
from .const import (
BUTTON_PRESSED_STATE_NAME,
DOMAIN,
EVENT_BUTTON_OFF,
EVENT_BUTTON_ON,
EVENT_CONF_BUTTON,
INSTEON_ENTITIES,
SIGNAL_LOAD_ALDB,
SIGNAL_PRINT_ALDB,
SRV_ADD_ALL_LINK,
SRV_ALL_LINK_GROUP,
SRV_ALL_LINK_MODE,
SRV_CONTROLLER,
SRV_DEL_ALL_LINK,
SRV_HOUSECODE,
SRV_LOAD_ALDB,
SRV_LOAD_DB_RELOAD,
SRV_PRINT_ALDB,
SRV_PRINT_IM_ALDB,
SRV_SCENE_OFF,
SRV_SCENE_ON,
SRV_X10_ALL_LIGHTS_OFF,
SRV_X10_ALL_LIGHTS_ON,
SRV_X10_ALL_UNITS_OFF,
)
from .ipdb import IPDB
from .schemas import (
ADD_ALL_LINK_SCHEMA,
DEL_ALL_LINK_SCHEMA,
LOAD_ALDB_SCHEMA,
PRINT_ALDB_SCHEMA,
TRIGGER_SCENE_SCHEMA,
X10_HOUSECODE_SCHEMA,
)
_LOGGER = logging.getLogger(__name__)
def register_new_device_callback(hass, config, insteon_modem):
"""Register callback for new Insteon device."""
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)
@callback
def async_new_insteon_device(device):
"""Detect device from transport to be delegated to platform."""
ipdb = IPDB()
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,
)
)
insteon_modem.devices.add_device_callback(async_new_insteon_device)
@callback
def async_register_services(hass, config, insteon_modem):
"""Register services used by insteon component."""
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][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)
hass.services.async_register(
DOMAIN, SRV_ADD_ALL_LINK, add_all_link, schema=ADD_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_DEL_ALL_LINK, del_all_link, schema=DEL_ALL_LINK_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_LOAD_ALDB, load_aldb, schema=LOAD_ALDB_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_PRINT_ALDB, print_aldb, schema=PRINT_ALDB_SCHEMA
)
hass.services.async_register(DOMAIN, SRV_PRINT_IM_ALDB, print_im_aldb, schema=None)
hass.services.async_register(
DOMAIN, SRV_X10_ALL_UNITS_OFF, x10_all_units_off, schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN, SRV_X10_ALL_LIGHTS_OFF, x10_all_lights_off, schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN, SRV_X10_ALL_LIGHTS_ON, x10_all_lights_on, schema=X10_HOUSECODE_SCHEMA,
)
hass.services.async_register(
DOMAIN, SRV_SCENE_ON, scene_on, schema=TRIGGER_SCENE_SCHEMA
)
hass.services.async_register(
DOMAIN, SRV_SCENE_OFF, scene_off, schema=TRIGGER_SCENE_SCHEMA
)
_LOGGER.debug("Insteon Services registered")
def print_aldb_to_log(aldb):
"""Print the All-Link Database to the log file."""
_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,
)
)