core/homeassistant/components/insteon/__init__.py

"""
Support for INSTEON Modems (PLM and Hub).

For more details about this component, please refer to the documentation at
https://home-assistant.io/components/insteon/
"""
import collections
import logging
from typing import Dict

import voluptuous as vol

from homeassistant.core import callback
from homeassistant.const import (CONF_PORT, EVENT_HOMEASSISTANT_STOP,
                                 CONF_PLATFORM,
                                 CONF_ENTITY_ID,
                                 CONF_HOST)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import discovery
from homeassistant.helpers.entity import Entity

REQUIREMENTS = ['insteonplm==0.15.2']

_LOGGER = logging.getLogger(__name__)

DOMAIN = 'insteon'

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_ADDRESS = 'address'
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'

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): cv.entity_id,
    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),
    })


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.get(CONF_ENTITY_ID)
        reload = service.data.get(SRV_LOAD_DB_RELOAD)
        entities = hass.data[DOMAIN].get('entities')
        entity = entities.get(entity_id)
        if entity:
            entity.load_aldb(reload)
        else:
            _LOGGER.error('Entity %s is not an INSTEON device', entity_id)

    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.get(CONF_ENTITY_ID)
        entities = hass.data[DOMAIN].get('entities')
        entity = entities.get(entity_id)
        if entity:
            entity.print_aldb()
        else:
            _LOGGER.error('Entity %s is not an INSTEON device', entity_id)

    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 _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)
        _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]['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)."""
        name = ''
        if self._insteon_device_state.group == 0x01:
            name = self._insteon_device.id
        else:
            name = '{:s}_{:d}'.format(self._insteon_device.id,
                                      self._insteon_device_state.group)
        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]['entities'][self.entity_id] = self

    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 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))