"""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 _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] 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).""" 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.""" _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, ) )