schemas.py

Go to the documentation of this file.

"""Schemas for the blueprint integration."""
 
from typing import Any
 
import voluptuous as vol
 
from homeassistant.const import (
    CONF_DEFAULT,
    CONF_DESCRIPTION,
    CONF_DOMAIN,
    CONF_ICON,
    CONF_NAME,
    CONF_PATH,
    CONF_SELECTOR,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv, selector
 
from .const import (
    CONF_AUTHOR,
    CONF_BLUEPRINT,
    CONF_COLLAPSED,
    CONF_HOMEASSISTANT,
    CONF_INPUT,
    CONF_MIN_VERSION,
    CONF_SOURCE_URL,
    CONF_USE_BLUEPRINT,
)
 
 
def version_validator(value: Any) -> str:
    """Validate a Home Assistant version."""
    if not isinstance(value, str):
        raise vol.Invalid("Version needs to be a string")
 
    parts = value.split(".")
 
    if len(parts) != 3:
        raise vol.Invalid("Version needs to be formatted as {major}.{minor}.{patch}")
 
    try:
        [int(p) for p in parts]
    except ValueError:
        raise vol.Invalid(
            "Major, minor and patch version needs to be an integer"
        ) from None
 
    return value
 
 
def unique_input_validator(inputs: Any) -> Any:
    """Validate the inputs don't have duplicate keys under different sections."""
    all_inputs = set()
    for key, value in inputs.items():
        if value and CONF_INPUT in value:
            for key in value[CONF_INPUT]:
                if key in all_inputs:
                    raise vol.Invalid(f"Duplicate use of input key {key} in blueprint.")
                all_inputs.add(key)
        else:
            if key in all_inputs:
                raise vol.Invalid(f"Duplicate use of input key {key} in blueprint.")
            all_inputs.add(key)
 
    return inputs
 
 
@callback
def is_blueprint_config(config: Any) -> bool:
    """Return if it is a blueprint config."""
    return isinstance(config, dict) and CONF_BLUEPRINT in config
 
 
@callback
def is_blueprint_instance_config(config: Any) -> bool:
    """Return if it is a blueprint instance config."""
    return isinstance(config, dict) and CONF_USE_BLUEPRINT in config
 
 
BLUEPRINT_INPUT_SCHEMA = vol.Schema(
    {
        vol.Optional(CONF_NAME): str,
        vol.Optional(CONF_DESCRIPTION): str,
        vol.Optional(CONF_DEFAULT): cv.match_all,
        vol.Optional(CONF_SELECTOR): selector.validate_selector,
    }
)
 
BLUEPRINT_INPUT_SECTION_SCHEMA = vol.Schema(
    {
        vol.Optional(CONF_NAME): str,
        vol.Optional(CONF_ICON): str,
        vol.Optional(CONF_DESCRIPTION): str,
        vol.Optional(CONF_COLLAPSED): bool,
        vol.Required(CONF_INPUT, default=dict): {
            str: vol.Any(
                None,
                BLUEPRINT_INPUT_SCHEMA,
            )
        },
    }
)
 
BLUEPRINT_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_BLUEPRINT): vol.Schema(
            {
                vol.Required(CONF_NAME): str,
                vol.Optional(CONF_DESCRIPTION): str,
                vol.Required(CONF_DOMAIN): str,
                vol.Optional(CONF_SOURCE_URL): cv.url,
                vol.Optional(CONF_AUTHOR): str,
                vol.Optional(CONF_HOMEASSISTANT): {
                    vol.Optional(CONF_MIN_VERSION): version_validator
                },
                vol.Optional(CONF_INPUT, default=dict): vol.All(
                    {
                        str: vol.Any(
                            None,
                            BLUEPRINT_INPUT_SCHEMA,
                            BLUEPRINT_INPUT_SECTION_SCHEMA,
                        )
                    },
                    unique_input_validator,
                ),
            }
        ),
    },
    extra=vol.ALLOW_EXTRA,
)
 
 
def validate_yaml_suffix(value: str) -> str:
    """Validate value has a YAML suffix."""
    if not value.endswith(".yaml"):
        raise vol.Invalid("Path needs to end in .yaml")
    return value
 
 
BLUEPRINT_INSTANCE_FIELDS = vol.Schema(
    {
        vol.Required(CONF_USE_BLUEPRINT): vol.Schema(
            {
                vol.Required(CONF_PATH): vol.All(cv.path, validate_yaml_suffix),
                vol.Required(CONF_INPUT, default=dict): {str: cv.match_all},
            }
        )
    },
    extra=vol.ALLOW_EXTRA,
)