1
0
mirror of https://github.com/esphome/esphome.git synced 2025-09-16 10:12:21 +01:00

Cleanup dashboard JS (#491)

* Cleanup dashboard JS

* Add vscode

* Save start_mark/end_mark

* Updates

* Updates

* Remove need for cv.nameable

It's a bit hacky but removes so much bloat from integrations

* Add enum helper

* Document APIs, and Improvements

* Fixes

* Fixes

* Update PULL_REQUEST_TEMPLATE.md

* Updates

* Updates

* Updates
This commit is contained in:
Otto Winter
2019-04-22 21:56:30 +02:00
committed by GitHub
parent 6682c43dfa
commit 8e75980ebd
359 changed files with 4395 additions and 4223 deletions

View File

@@ -6,20 +6,23 @@ import logging
import re
import os.path
# pylint: disable=unused-import, wrong-import-order
from contextlib import contextmanager
import voluptuous as vol
from esphome import core, core_config, yaml_util
from esphome.components import substitutions
from esphome.components.substitutions import CONF_SUBSTITUTIONS
from esphome.const import CONF_ESPHOME, CONF_PLATFORM, ESP_PLATFORMS
from esphome.core import CORE, EsphomeError
from esphome.core import CORE, EsphomeError # noqa
from esphome.helpers import color, indent
from esphome.py_compat import text_type
from esphome.util import safe_print, OrderedDict
# pylint: disable=unused-import, wrong-import-order
from typing import List, Optional, Tuple, Union # noqa
from esphome.core import ConfigType # noqa
from esphome.yaml_util import is_secret
from esphome.yaml_util import is_secret, ESPHomeDataBase
from esphome.voluptuous_schema import ExtraKeysInvalid
_LOGGER = logging.getLogger(__name__)
@@ -162,61 +165,83 @@ def iter_components(config):
ConfigPath = List[Union[str, int]]
def _path_begins_with_(path, other): # type: (ConfigPath, ConfigPath) -> bool
def _path_begins_with(path, other): # type: (ConfigPath, ConfigPath) -> bool
if len(path) < len(other):
return False
return path[:len(other)] == other
def _path_begins_with(path, other): # type: (ConfigPath, ConfigPath) -> bool
ret = _path_begins_with_(path, other)
return ret
class Config(OrderedDict):
def __init__(self):
super(Config, self).__init__()
self.errors = [] # type: List[Tuple[basestring, ConfigPath]]
self.domains = [] # type: List[Tuple[ConfigPath, basestring]]
# A list of voluptuous errors
self.errors = [] # type: List[vol.Invalid]
# A list of paths that should be fully outputted
# The values will be the paths to all "domain", for example (['logger'], 'logger')
# or (['sensor', 'ultrasonic'], 'sensor.ultrasonic')
self.output_paths = [] # type: List[Tuple[ConfigPath, unicode]]
def add_error(self, message, path):
def add_error(self, error):
# type: (vol.Invalid) -> None
if isinstance(error, vol.MultipleInvalid):
for err in error.errors:
self.add_error(err)
return
self.errors.append(error)
@contextmanager
def catch_error(self, path=None):
path = path or []
try:
yield
except vol.Invalid as e:
e.prepend(path)
self.add_error(e)
def add_str_error(self, message, path):
# type: (basestring, ConfigPath) -> None
if not isinstance(message, text_type):
message = text_type(message)
self.errors.append((message, path))
self.add_error(vol.Invalid(message, path))
def add_domain(self, path, name):
# type: (ConfigPath, basestring) -> None
self.domains.append((path, name))
def add_output_path(self, path, domain):
# type: (ConfigPath, unicode) -> None
self.output_paths.append((path, domain))
def remove_domain(self, path, name):
self.domains.remove((path, name))
def lookup_domain(self, path):
# type: (ConfigPath) -> Optional[basestring]
best_len = 0
best_domain = None
for d_path, domain in self.domains:
if len(d_path) < best_len:
continue
if _path_begins_with(path, d_path):
best_len = len(d_path)
best_domain = domain
return best_domain
def remove_output_path(self, path, domain):
# type: (ConfigPath, unicode) -> None
self.output_paths.remove((path, domain))
def is_in_error_path(self, path):
for _, p in self.errors:
if _path_begins_with(p, path):
# type: (ConfigPath) -> bool
for err in self.errors:
if _path_begins_with(err.path, path):
return True
return False
def set_by_path(self, path, value):
conf = self
for key in path[:-1]:
conf = conf[key]
conf[path[-1]] = value
def get_error_for_path(self, path):
for msg, p in self.errors:
if self.nested_item_path(p) == path:
return msg
# type: (ConfigPath) -> Optional[vol.Invalid]
for err in self.errors:
if self.get_deepest_path(err.path) == path:
return err
return None
def nested_item(self, path):
def get_deepest_value_for_path(self, path):
# type: (ConfigPath) -> ConfigType
data = self
for item_index in path:
try:
data = data[item_index]
except (KeyError, IndexError, TypeError):
return data
return data
def get_nested_item(self, path):
# type: (ConfigPath) -> ConfigType
data = self
for item_index in path:
try:
@@ -225,7 +250,9 @@ class Config(OrderedDict):
return {}
return data
def nested_item_path(self, path):
def get_deepest_path(self, path):
# type: (ConfigPath) -> ConfigPath
"""Return the path that is the deepest reachable by following path."""
data = self
part = []
for item_index in path:
@@ -261,9 +288,13 @@ def do_id_pass(result): # type: (Config) -> None
searching_ids = [] # type: List[Tuple[core.ID, ConfigPath]]
for id, path in iter_ids(result):
if id.is_declaration:
if id.id is not None and any(v[0].id == id.id for v in declare_ids):
result.add_error(u"ID {} redefined!".format(id.id), path)
continue
if id.id is not None:
# Look for duplicate definitions
match = next((v for v in declare_ids if v[0].id == id.id), None)
if match is not None:
opath = u'->'.join(text_type(v) for v in match[1])
result.add_str_error(u"ID {} redefined! Check {}".format(id.id, opath), path)
continue
declare_ids.append((id, path))
else:
searching_ids.append((id, path))
@@ -278,14 +309,22 @@ def do_id_pass(result): # type: (Config) -> None
match = next((v[0] for v in declare_ids if v[0].id == id.id), None)
if match is None:
# No declared ID with this name
result.add_error("Couldn't find ID '{}'".format(id.id), path)
import difflib
error = ("Couldn't find ID '{}'. Please check you have defined "
"an ID with that name in your configuration.".format(id.id))
# Find candidates
matches = difflib.get_close_matches(id.id, [v[0].id for v in declare_ids])
if matches:
matches_s = ', '.join('"{}"'.format(x) for x in matches)
error += " These IDs look similar: {}.".format(matches_s)
result.add_str_error(error, path)
continue
if not isinstance(match.type, MockObjClass) or not isinstance(id.type, MockObjClass):
continue
if not match.type.inherits_from(id.type):
result.add_error("ID '{}' of type {} doesn't inherit from {}. Please double check "
"your ID is pointing to the correct value"
"".format(id.id, match.type, id.type), path)
result.add_str_error("ID '{}' of type {} doesn't inherit from {}. Please "
"double check your ID is pointing to the correct value"
"".format(id.id, match.type, id.type), path)
if id.id is None and id.type is not None:
for v in declare_ids:
@@ -296,204 +335,185 @@ def do_id_pass(result): # type: (Config) -> None
id.id = v[0].id
break
else:
result.add_error("Couldn't resolve ID for type '{}'".format(id.type), path)
result.add_str_error("Couldn't resolve ID for type '{}'".format(id.type), path)
def validate_config(config):
result = Config()
def _comp_error(ex, path):
# type: (vol.Invalid, List[basestring]) -> None
if isinstance(ex, vol.MultipleInvalid):
errors = ex.errors
else:
errors = [ex]
# 1. Load substitutions
if CONF_SUBSTITUTIONS in config:
result[CONF_SUBSTITUTIONS] = config[CONF_SUBSTITUTIONS]
result.add_output_path([CONF_SUBSTITUTIONS], CONF_SUBSTITUTIONS)
try:
substitutions.do_substitution_pass(config)
except vol.Invalid as err:
result.add_error(err)
return result
for e in errors:
path_ = path + e.path
domain = result.lookup_domain(path_) or ''
result.add_error(_format_vol_invalid(e, config, path, domain), path_)
skip_paths = list() # type: List[ConfigPath]
# Step 1: Load everything
result.add_domain([CONF_ESPHOME], CONF_ESPHOME)
# 2. Load partial core config
result[CONF_ESPHOME] = config[CONF_ESPHOME]
config_queue = collections.deque()
result.add_output_path([CONF_ESPHOME], CONF_ESPHOME)
try:
core_config.preload_core_config(config)
except vol.Invalid as err:
result.add_error(err)
return result
# Remove temporary esphome config path again, it will be reloaded later
result.remove_output_path([CONF_ESPHOME], CONF_ESPHOME)
# 3. Load components.
# Load components (also AUTO_LOAD) and set output paths of result
# Queue of items to load, FIFO
load_queue = collections.deque()
for domain, conf in config.items():
config_queue.append((domain, conf))
load_queue.append((domain, conf))
while config_queue:
domain, conf = config_queue.popleft()
domain = str(domain)
if domain == CONF_ESPHOME or domain.startswith(u'.'):
skip_paths.append([domain])
# List of items to enter next stage
check_queue = [] # type: List[Tuple[ConfigPath, str, ConfigType, ComponentManifest]]
# This step handles:
# - Adding output path
# - Auto Load
# - Loading configs into result
while load_queue:
domain, conf = load_queue.popleft()
domain = text_type(domain)
if domain.startswith(u'.'):
# Ignore top-level keys starting with a dot
continue
result.add_domain([domain], domain)
result.add_output_path([domain], domain)
result[domain] = conf
if conf is None:
result[domain] = conf = {}
component = get_component(domain)
path = [domain]
if component is None:
result.add_error(u"Component not found: {}".format(domain), [domain])
skip_paths.append([domain])
continue
if component.is_multi_conf and not isinstance(conf, list):
result[domain] = conf = [conf]
success = True
for dependency in component.dependencies:
if dependency not in config:
result.add_error(u"Component {} requires component {}".format(domain, dependency),
[domain])
success = False
if not success:
skip_paths.append([domain])
continue
success = True
for conflict in component.conflicts_with:
if conflict in config:
result.add_error(u"Component {} cannot be used together with component {}"
u"".format(domain, conflict), [domain])
success = False
if not success:
skip_paths.append([domain])
result.add_str_error(u"Component not found: {}".format(domain), path)
continue
# Process AUTO_LOAD
for load in component.auto_load:
if load not in config:
conf = core.AutoLoad()
config[load] = conf
config_queue.append((load, conf))
if CORE.esp_platform not in component.esp_platforms:
result.add_error(u"Component {} doesn't support {}.".format(domain, CORE.esp_platform),
[domain])
skip_paths.append([domain])
continue
load_conf = core.AutoLoad()
config[load] = load_conf
load_queue.append((load, load_conf))
if not component.is_platform_component:
if component.config_schema is None and not isinstance(conf, core.AutoLoad):
result.add_error(u"Component {} cannot be loaded via YAML (no CONFIG_SCHEMA)."
u"".format(domain), [domain])
skip_paths.append([domain])
check_queue.append(([domain], domain, conf, component))
continue
result.remove_domain([domain], domain)
# This is a platform component, proceed to reading platform entries
# Remove this is as an output path
result.remove_output_path([domain], domain)
# Ensure conf is a list
if not isinstance(conf, list) and conf:
result[domain] = conf = [conf]
for i, p_config in enumerate(conf):
path = [domain, i]
# Construct temporary unknown output path
p_domain = u'{}.unknown'.format(domain)
result.add_output_path(path, p_domain)
result[domain][i] = p_config
if not isinstance(p_config, dict):
result.add_error(u"Platform schemas must have 'platform:' key", [domain, i])
skip_paths.append([domain, i])
result.add_str_error(u"Platform schemas must be key-value pairs.", path)
continue
p_name = p_config.get('platform')
if p_name is None:
result.add_error(u"No platform specified for {}".format(domain), [domain, i])
skip_paths.append([domain, i])
result.add_str_error(u"No platform specified! See 'platform' key.", path)
continue
# Remove temp output path and construct new one
result.remove_output_path(path, p_domain)
p_domain = u'{}.{}'.format(domain, p_name)
result.add_domain([domain, i], p_domain)
result.add_output_path(path, p_domain)
# Try Load platform
platform = get_platform(domain, p_name)
if platform is None:
result.add_error(u"Platform not found: '{}'".format(p_domain), [domain, i])
skip_paths.append([domain, i])
continue
success = True
for dependency in platform.dependencies:
if dependency not in config:
result.add_error(u"Platform {} requires component {}"
u"".format(p_domain, dependency), [domain, i])
success = False
if not success:
skip_paths.append([domain, i])
continue
success = True
for conflict in platform.conflicts_with:
if conflict in config:
result.add_error(u"Platform {} cannot be used together with component {}"
u"".format(p_domain, conflict), [domain, i])
success = False
if not success:
skip_paths.append([domain, i])
result.add_str_error(u"Platform not found: '{}'".format(p_domain), path)
continue
# Process AUTO_LOAD
for load in platform.auto_load:
if load not in config:
conf = core.AutoLoad()
config[load] = conf
config_queue.append((load, conf))
load_conf = core.AutoLoad()
config[load] = load_conf
load_queue.append((load, load_conf))
if CORE.esp_platform not in platform.esp_platforms:
result.add_error(u"Platform {} doesn't support {}."
u"".format(p_domain, CORE.esp_platform), [domain, i])
skip_paths.append([domain, i])
continue
check_queue.append((path, p_domain, p_config, platform))
if platform.config_schema is None:
result.add_error(u"Platform {} cannot be loaded via YAML (no PLATFORM_SCHEMA)."
u"".format(p_domain), [domain, i])
skip_paths.append([domain])
# 4. Validate component metadata, including
# - Transformation (nullable, multi conf)
# - Dependencies
# - Conflicts
# - Supported ESP Platform
# Step 2: Validate configuration
try:
result[CONF_ESPHOME] = core_config.CONFIG_SCHEMA(result[CONF_ESPHOME])
except vol.Invalid as ex:
_comp_error(ex, [CONF_ESPHOME])
# List of items to proceed to next stage
validate_queue = [] # type: List[Tuple[ConfigPath, ConfigType, ComponentManifest]]
for path, domain, conf, comp in check_queue:
if conf is None:
result[domain] = conf = {}
for domain, conf in result.items():
domain = str(domain)
if [domain] in skip_paths:
continue
component = get_component(domain)
if not component.is_platform_component:
if component.config_schema is None:
continue
if component.is_multi_conf:
for i, conf_ in enumerate(conf):
try:
validated = component.config_schema(conf_)
result[domain][i] = validated
except vol.Invalid as ex:
_comp_error(ex, [domain, i])
else:
try:
validated = component.config_schema(conf)
result[domain] = validated
except vol.Invalid as ex:
_comp_error(ex, [domain])
continue
success = True
for dependency in comp.dependencies:
if dependency not in config:
result.add_str_error(u"Component {} requires component {}"
u"".format(domain, dependency), path)
success = False
if not success:
continue
for i, p_config in enumerate(conf):
if [domain, i] in skip_paths:
continue
p_name = p_config['platform']
platform = get_platform(domain, p_name)
success = True
for conflict in comp.conflicts_with:
if conflict in config:
result.add_str_error(u"Component {} cannot be used together with component {}"
u"".format(domain, conflict), path)
success = False
if not success:
continue
if platform.config_schema is not None:
if CORE.esp_platform not in comp.esp_platforms:
result.add_str_error(u"Component {} doesn't support {}.".format(domain,
CORE.esp_platform),
path)
continue
if not comp.is_platform_component and comp.config_schema is None and \
not isinstance(conf, core.AutoLoad):
result.add_str_error(u"Component {} cannot be loaded via YAML "
u"(no CONFIG_SCHEMA).".format(domain), path)
continue
if comp.is_multi_conf:
if not isinstance(conf, list):
result[domain] = conf = [conf]
for i, part_conf in enumerate(conf):
validate_queue.append((path + [i], part_conf, comp))
continue
validate_queue.append((path, conf, comp))
# 5. Validate configuration schema
for path, conf, comp in validate_queue:
if comp.config_schema is None:
continue
with result.catch_error(path):
if comp.is_platform:
# Remove 'platform' key for validation
input_conf = OrderedDict(p_config)
input_conf = OrderedDict(conf)
platform_val = input_conf.pop('platform')
try:
p_validated = platform.config_schema(input_conf)
except vol.Invalid as ex:
_comp_error(ex, [domain, i])
continue
if not isinstance(p_validated, OrderedDict):
p_validated = OrderedDict(p_validated)
p_validated['platform'] = platform_val
p_validated.move_to_end('platform', last=False)
result[domain][i] = p_validated
validated = comp.config_schema(input_conf)
# Ensure result is OrderedDict so we can call move_to_end
if not isinstance(validated, OrderedDict):
validated = OrderedDict(validated)
validated['platform'] = platform_val
validated.move_to_end('platform', last=False)
result.set_by_path(path, validated)
else:
validated = comp.config_schema(conf)
result.set_by_path(path, validated)
# 6. If no validation errors, check IDs
if not result.errors:
# Only parse IDs if no validation error. Otherwise
# user gets confusing messages
@@ -511,9 +531,6 @@ def _nested_getitem(data, path):
def humanize_error(config, validation_error):
offending_item_summary = _nested_getitem(config, validation_error.path)
if isinstance(offending_item_summary, dict):
offending_item_summary = None
validation_error = text_type(validation_error)
m = re.match(r'^(.*?)\s*(?:for dictionary value )?@ data\[.*$', validation_error)
if m is not None:
@@ -521,19 +538,26 @@ def humanize_error(config, validation_error):
validation_error = validation_error.strip()
if not validation_error.endswith(u'.'):
validation_error += u'.'
if offending_item_summary is None or is_secret(offending_item_summary):
return validation_error
return u"{} Got '{}'".format(validation_error, offending_item_summary)
return validation_error
def _format_vol_invalid(ex, config, path, domain):
# type: (vol.Invalid, ConfigType, ConfigPath, basestring) -> unicode
def _get_parent_name(path, config):
if not path:
return '<root>'
for domain_path, domain in config.output_paths:
if _path_begins_with(path, domain_path):
if len(path) > len(domain_path):
# Sub-item
break
return domain
return path[-1]
def _format_vol_invalid(ex, config):
# type: (vol.Invalid, Config) -> unicode
message = u''
try:
paren = ex.path[-2]
except IndexError:
paren = domain
paren = _get_parent_name(ex.path[:-1], config)
if isinstance(ex, ExtraKeysInvalid):
if ex.candidates:
@@ -547,20 +571,26 @@ def _format_vol_invalid(ex, config, path, domain):
elif u'required key not provided' in ex.error_message:
message += u"'{}' is a required option for [{}].".format(ex.path[-1], paren)
else:
message += humanize_error(_nested_getitem(config, path), ex)
message += humanize_error(config, ex)
return message
def load_config():
class InvalidYAMLError(EsphomeError):
def __init__(self, path, base_exc):
message = u"Invalid YAML at {}. Please see YAML syntax reference or use an " \
u"online YAML syntax validator. ({})".format(path, base_exc)
super(InvalidYAMLError, self).__init__(message)
self.path = path
self.base_exc = base_exc
def _load_config():
try:
config = yaml_util.load_yaml(CORE.config_path)
except OSError:
raise EsphomeError(u"Invalid YAML at {}. Please see YAML syntax reference or use an online "
u"YAML syntax validator".format(CORE.config_path))
except EsphomeError as e:
raise InvalidYAMLError(CORE.config_path, e)
CORE.raw_config = config
config = substitutions.do_substitution_pass(config)
core_config.preload_core_config(config)
try:
result = validate_config(config)
@@ -573,13 +603,21 @@ def load_config():
return result
def load_config():
try:
return _load_config()
except vol.Invalid as err:
raise EsphomeError("Error while parsing config: {}".format(err))
def line_info(obj, highlight=True):
"""Display line config source."""
if not highlight:
return None
if hasattr(obj, '__config_file__'):
return color('cyan', "[source {}:{}]"
.format(obj.__config_file__, obj.__line__ or '?'))
if isinstance(obj, ESPHomeDataBase) and obj.esp_range is not None:
mark = obj.esp_range.start_mark
source = u"[source {}:{}]".format(mark.document, mark.line + 1)
return color('cyan', source)
return None
@@ -595,14 +633,14 @@ def _print_on_next_line(obj):
def dump_dict(config, path, at_root=True):
# type: (Config, ConfigPath, bool) -> Tuple[unicode, bool]
conf = config.nested_item(path)
conf = config.get_nested_item(path)
ret = u''
multiline = False
if at_root:
error = config.get_error_for_path(path)
if error is not None:
ret += u'\n' + color('bold_red', error) + u'\n'
ret += u'\n' + color('bold_red', _format_vol_invalid(error, config)) + u'\n'
if isinstance(conf, (list, tuple)):
multiline = True
@@ -614,14 +652,14 @@ def dump_dict(config, path, at_root=True):
path_ = path + [i]
error = config.get_error_for_path(path_)
if error is not None:
ret += u'\n' + color('bold_red', error) + u'\n'
ret += u'\n' + color('bold_red', _format_vol_invalid(error, config)) + u'\n'
sep = u'- '
if config.is_in_error_path(path_):
sep = color('red', sep)
msg, _ = dump_dict(config, path_, at_root=False)
msg = indent(msg)
inf = line_info(config.nested_item(path_), highlight=config.is_in_error_path(path_))
inf = line_info(config.get_nested_item(path_), highlight=config.is_in_error_path(path_))
if inf is not None:
msg = inf + u'\n' + msg
elif msg:
@@ -637,14 +675,14 @@ def dump_dict(config, path, at_root=True):
path_ = path + [k]
error = config.get_error_for_path(path_)
if error is not None:
ret += u'\n' + color('bold_red', error) + u'\n'
ret += u'\n' + color('bold_red', _format_vol_invalid(error, config)) + u'\n'
st = u'{}: '.format(k)
if config.is_in_error_path(path_):
st = color('red', st)
msg, m = dump_dict(config, path_, at_root=False)
inf = line_info(config.nested_item(path_), highlight=config.is_in_error_path(path_))
inf = line_info(config.get_nested_item(path_), highlight=config.is_in_error_path(path_))
if m:
msg = u'\n' + indent(msg)
@@ -717,12 +755,12 @@ def read_config(verbose):
safe_print(color('bold_red', u"Failed config"))
safe_print('')
for path, domain in res.domains:
for path, domain in res.output_paths:
if not res.is_in_error_path(path):
continue
safe_print(color('bold_red', u'{}:'.format(domain)) + u' ' +
(line_info(res.nested_item(path)) or u''))
(line_info(res.get_nested_item(path)) or u''))
safe_print(indent(dump_dict(res, path)[0]))
return None
return OrderedDict(res)