core/homeassistant/components/insteon/schemas.py

"""Schemas used by insteon component."""
from binascii import Error as HexError, unhexlify
from typing import Dict

from pyinsteon.address import Address
from pyinsteon.constants import HC_LOOKUP
import voluptuous as vol

from homeassistant.const import (
    CONF_ADDRESS,
    CONF_DEVICE,
    CONF_ENTITY_ID,
    CONF_HOST,
    CONF_PASSWORD,
    CONF_PLATFORM,
    CONF_PORT,
    CONF_USERNAME,
    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,
    PORT_HUB_V1,
    PORT_HUB_V2,
    SRV_ALL_LINK_GROUP,
    SRV_ALL_LINK_MODE,
    SRV_CONTROLLER,
    SRV_HOUSECODE,
    SRV_LOAD_DB_RELOAD,
    SRV_RESPONDER,
    X10_PLATFORMS,
)


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
        schema[CONF_IP_PORT] = PORT_HUB_V1 if hub_version == 1 else PORT_HUB_V2
    return schema


CONF_DEVICE_OVERRIDE_SCHEMA = vol.All(
    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(
            cv.deprecated(CONF_X10_ALL_UNITS_OFF),
            cv.deprecated(CONF_X10_ALL_LIGHTS_ON),
            cv.deprecated(CONF_X10_ALL_LIGHTS_OFF),
            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): 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)}
)


ADD_DEFAULT_LINKS_SCHEMA = vol.Schema({vol.Required(CONF_ENTITY_ID): cv.entity_id})


def normalize_byte_entry_to_int(entry: [int, bytes, str]):
    """Format a hex entry value."""
    if isinstance(entry, int):
        if entry in range(0, 256):
            return entry
        raise ValueError("Must be single byte")
    if isinstance(entry, str):
        if entry[0:2].lower() == "0x":
            entry = entry[2:]
        if len(entry) != 2:
            raise ValueError("Not a valid hex code")
        try:
            entry = unhexlify(entry)
        except HexError as err:
            raise ValueError("Not a valid hex code") from err
    return int.from_bytes(entry, byteorder="big")


def add_device_override(config_data, new_override):
    """Add a new device override."""
    try:
        address = str(Address(new_override[CONF_ADDRESS]))
        cat = normalize_byte_entry_to_int(new_override[CONF_CAT])
        subcat = normalize_byte_entry_to_int(new_override[CONF_SUBCAT])
    except ValueError as err:
        raise ValueError("Incorrect values") from err

    overrides = config_data.get(CONF_OVERRIDE, [])
    curr_override = {}

    # If this address has an override defined, remove it
    for override in overrides:
        if override[CONF_ADDRESS] == address:
            curr_override = override
            break
    if curr_override:
        overrides.remove(curr_override)

    curr_override[CONF_ADDRESS] = address
    curr_override[CONF_CAT] = cat
    curr_override[CONF_SUBCAT] = subcat
    overrides.append(curr_override)
    config_data[CONF_OVERRIDE] = overrides
    return config_data


def add_x10_device(config_data, new_x10):
    """Add a new X10 device to X10 device list."""
    curr_device = {}
    x10_devices = config_data.get(CONF_X10, [])
    for x10_device in x10_devices:
        if (
            x10_device[CONF_HOUSECODE] == new_x10[CONF_HOUSECODE]
            and x10_device[CONF_UNITCODE] == new_x10[CONF_UNITCODE]
        ):
            curr_device = x10_device
            break

    if curr_device:
        x10_devices.remove(curr_device)

    curr_device[CONF_HOUSECODE] = new_x10[CONF_HOUSECODE]
    curr_device[CONF_UNITCODE] = new_x10[CONF_UNITCODE]
    curr_device[CONF_PLATFORM] = new_x10[CONF_PLATFORM]
    curr_device[CONF_DIM_STEPS] = new_x10[CONF_DIM_STEPS]
    x10_devices.append(curr_device)
    config_data[CONF_X10] = x10_devices
    return config_data


def build_device_override_schema(
    address=vol.UNDEFINED,
    cat=vol.UNDEFINED,
    subcat=vol.UNDEFINED,
    firmware=vol.UNDEFINED,
):
    """Build the device override schema for config flow."""
    return vol.Schema(
        {
            vol.Required(CONF_ADDRESS, default=address): str,
            vol.Optional(CONF_CAT, default=cat): str,
            vol.Optional(CONF_SUBCAT, default=subcat): str,
        }
    )


def build_x10_schema(
    housecode=vol.UNDEFINED,
    unitcode=vol.UNDEFINED,
    platform=vol.UNDEFINED,
    dim_steps=22,
):
    """Build the X10 schema for config flow."""
    return vol.Schema(
        {
            vol.Required(CONF_HOUSECODE, default=housecode): vol.In(HC_LOOKUP.keys()),
            vol.Required(CONF_UNITCODE, default=unitcode): vol.In(range(1, 17)),
            vol.Required(CONF_PLATFORM, default=platform): vol.In(X10_PLATFORMS),
            vol.Optional(CONF_DIM_STEPS, default=dim_steps): vol.In(range(1, 255)),
        }
    )


def build_plm_schema(device=vol.UNDEFINED):
    """Build the PLM schema for config flow."""
    return vol.Schema({vol.Required(CONF_DEVICE, default=device): str})


def build_hub_schema(
    hub_version,
    host=vol.UNDEFINED,
    port=PORT_HUB_V2,
    username=vol.UNDEFINED,
    password=vol.UNDEFINED,
):
    """Build the Hub v2 schema for config flow."""
    schema = {
        vol.Required(CONF_HOST, default=host): str,
        vol.Required(CONF_PORT, default=port): int,
    }
    if hub_version == 2:
        schema[vol.Required(CONF_USERNAME, default=username)] = str
        schema[vol.Required(CONF_PASSWORD, default=password)] = str
    return vol.Schema(schema)


def build_remove_override_schema(data):
    """Build the schema to remove device overrides in config flow options."""
    selection = []
    for override in data:
        selection.append(override[CONF_ADDRESS])
    return vol.Schema({vol.Required(CONF_ADDRESS): vol.In(selection)})


def build_remove_x10_schema(data):
    """Build the schema to remove an X10 device in config flow options."""
    selection = []
    for device in data:
        housecode = device[CONF_HOUSECODE].upper()
        unitcode = device[CONF_UNITCODE]
        selection.append(f"Housecode: {housecode}, Unitcode: {unitcode}")
    return vol.Schema({vol.Required(CONF_DEVICE): vol.In(selection)})


def convert_yaml_to_config_flow(yaml_config):
    """Convert the YAML based configuration to a config flow configuration."""
    config = {}
    if yaml_config.get(CONF_HOST):
        hub_version = yaml_config.get(CONF_HUB_VERSION, 2)
        default_port = PORT_HUB_V2 if hub_version == 2 else PORT_HUB_V1
        config[CONF_HOST] = yaml_config.get(CONF_HOST)
        config[CONF_PORT] = yaml_config.get(CONF_PORT, default_port)
        config[CONF_HUB_VERSION] = hub_version
        if hub_version == 2:
            config[CONF_USERNAME] = yaml_config[CONF_USERNAME]
            config[CONF_PASSWORD] = yaml_config[CONF_PASSWORD]
    else:
        config[CONF_DEVICE] = yaml_config[CONF_PORT]

    options = {}
    for old_override in yaml_config.get(CONF_OVERRIDE, []):
        override = {}
        override[CONF_ADDRESS] = str(Address(old_override[CONF_ADDRESS]))
        override[CONF_CAT] = normalize_byte_entry_to_int(old_override[CONF_CAT])
        override[CONF_SUBCAT] = normalize_byte_entry_to_int(old_override[CONF_SUBCAT])
        options = add_device_override(options, override)

    for x10_device in yaml_config.get(CONF_X10, []):
        options = add_x10_device(options, x10_device)

    return config, options