mirror of
https://github.com/esphome/esphome.git
synced 2024-11-14 02:58:11 +01:00
80214640b1
* Filter some debug lines from PlatformIO in output * Lint * Strip trailing newline from esp-idf output * Only create global std::string if on esp32
268 lines
8.1 KiB
Python
268 lines
8.1 KiB
Python
from __future__ import print_function
|
|
|
|
import collections
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
from esphome import const
|
|
from esphome.py_compat import IS_PY2, decode_text, text_type
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class RegistryEntry(object):
|
|
def __init__(self, name, fun, type_id, schema):
|
|
self.name = name
|
|
self.fun = fun
|
|
self.type_id = type_id
|
|
self.raw_schema = schema
|
|
|
|
@property
|
|
def coroutine_fun(self):
|
|
from esphome.core import coroutine
|
|
return coroutine(self.fun)
|
|
|
|
@property
|
|
def schema(self):
|
|
from esphome.config_validation import Schema
|
|
return Schema(self.raw_schema)
|
|
|
|
|
|
class Registry(dict):
|
|
def __init__(self, base_schema=None, type_id_key=None):
|
|
super(Registry, self).__init__()
|
|
self.base_schema = base_schema or {}
|
|
self.type_id_key = type_id_key
|
|
|
|
def register(self, name, type_id, schema):
|
|
def decorator(fun):
|
|
self[name] = RegistryEntry(name, fun, type_id, schema)
|
|
return fun
|
|
|
|
return decorator
|
|
|
|
|
|
class SimpleRegistry(dict):
|
|
def register(self, name, data):
|
|
def decorator(fun):
|
|
self[name] = (fun, data)
|
|
return fun
|
|
|
|
return decorator
|
|
|
|
|
|
def safe_print(message=""):
|
|
from esphome.core import CORE
|
|
|
|
if CORE.dashboard:
|
|
try:
|
|
message = message.replace('\033', '\\033')
|
|
except UnicodeEncodeError:
|
|
pass
|
|
|
|
try:
|
|
print(message)
|
|
return
|
|
except UnicodeEncodeError:
|
|
pass
|
|
|
|
try:
|
|
print(message.encode('utf-8', 'backslashreplace'))
|
|
except UnicodeEncodeError:
|
|
try:
|
|
print(message.encode('ascii', 'backslashreplace'))
|
|
except UnicodeEncodeError:
|
|
print("Cannot print line because of invalid locale!")
|
|
|
|
|
|
def shlex_quote(s):
|
|
if not s:
|
|
return u"''"
|
|
if re.search(r'[^\w@%+=:,./-]', s) is None:
|
|
return s
|
|
|
|
return u"'" + s.replace(u"'", u"'\"'\"'") + u"'"
|
|
|
|
|
|
ANSI_ESCAPE = re.compile(r'\033[@-_][0-?]*[ -/]*[@-~]')
|
|
|
|
|
|
class RedirectText(object):
|
|
def __init__(self, out, filter_lines=None):
|
|
self._out = out
|
|
if filter_lines is None:
|
|
self._filter_pattern = None
|
|
else:
|
|
pattern = r'|'.join(r'(?:' + pattern + r')' for pattern in filter_lines)
|
|
self._filter_pattern = re.compile(pattern)
|
|
self._line_buffer = ''
|
|
|
|
def __getattr__(self, item):
|
|
return getattr(self._out, item)
|
|
|
|
def _write_color_replace(self, s):
|
|
from esphome.core import CORE
|
|
|
|
if CORE.dashboard:
|
|
# With the dashboard, we must create a little hack to make color output
|
|
# work. The shell we create in the dashboard is not a tty, so python removes
|
|
# all color codes from the resulting stream. We just convert them to something
|
|
# we can easily recognize later here.
|
|
s = s.replace('\033', '\\033')
|
|
self._out.write(s)
|
|
|
|
def write(self, s):
|
|
# s is usually a text_type already (self._out is of type TextIOWrapper)
|
|
# However, s is sometimes also a bytes object in python3. Let's make sure it's a
|
|
# text_type
|
|
# If the conversion fails, we will create an exception, which is okay because we won't
|
|
# be able to print it anyway.
|
|
text = decode_text(s)
|
|
assert isinstance(text, text_type)
|
|
|
|
if self._filter_pattern is not None:
|
|
self._line_buffer += text
|
|
lines = self._line_buffer.splitlines(True)
|
|
for line in lines:
|
|
if '\n' not in line and '\r' not in line:
|
|
# Not a complete line, set line buffer
|
|
self._line_buffer = line
|
|
break
|
|
self._line_buffer = ''
|
|
|
|
line_without_ansi = ANSI_ESCAPE.sub('', line)
|
|
line_without_end = line_without_ansi.rstrip()
|
|
if self._filter_pattern.match(line_without_end) is not None:
|
|
# Filter pattern matched, ignore the line
|
|
continue
|
|
|
|
self._write_color_replace(line)
|
|
else:
|
|
self._write_color_replace(text)
|
|
|
|
# write() returns the number of characters written
|
|
# Let's print the number of characters of the original string in order to not confuse
|
|
# any caller.
|
|
return len(s)
|
|
|
|
# pylint: disable=no-self-use
|
|
def isatty(self):
|
|
return True
|
|
|
|
|
|
def run_external_command(func, *cmd, **kwargs):
|
|
def mock_exit(return_code):
|
|
raise SystemExit(return_code)
|
|
|
|
orig_argv = sys.argv
|
|
orig_exit = sys.exit # mock sys.exit
|
|
full_cmd = u' '.join(shlex_quote(x) for x in cmd)
|
|
_LOGGER.info(u"Running: %s", full_cmd)
|
|
|
|
filter_lines = kwargs.get('filter_lines')
|
|
orig_stdout = sys.stdout
|
|
sys.stdout = RedirectText(sys.stdout, filter_lines=filter_lines)
|
|
orig_stderr = sys.stderr
|
|
sys.stderr = RedirectText(sys.stderr, filter_lines=filter_lines)
|
|
|
|
capture_stdout = kwargs.get('capture_stdout', False)
|
|
if capture_stdout:
|
|
cap_stdout = sys.stdout = io.StringIO()
|
|
|
|
try:
|
|
sys.argv = list(cmd)
|
|
sys.exit = mock_exit
|
|
return func() or 0
|
|
except KeyboardInterrupt:
|
|
return 1
|
|
except SystemExit as err:
|
|
return err.args[0]
|
|
except Exception as err: # pylint: disable=broad-except
|
|
_LOGGER.error(u"Running command failed: %s", err)
|
|
_LOGGER.error(u"Please try running %s locally.", full_cmd)
|
|
finally:
|
|
sys.argv = orig_argv
|
|
sys.exit = orig_exit
|
|
|
|
sys.stdout = orig_stdout
|
|
sys.stderr = orig_stderr
|
|
|
|
if capture_stdout:
|
|
# pylint: disable=lost-exception
|
|
return cap_stdout.getvalue()
|
|
|
|
|
|
def run_external_process(*cmd, **kwargs):
|
|
full_cmd = u' '.join(shlex_quote(x) for x in cmd)
|
|
_LOGGER.info(u"Running: %s", full_cmd)
|
|
filter_lines = kwargs.get('filter_lines')
|
|
|
|
capture_stdout = kwargs.get('capture_stdout', False)
|
|
if capture_stdout:
|
|
sub_stdout = io.BytesIO()
|
|
else:
|
|
sub_stdout = RedirectText(sys.stdout, filter_lines=filter_lines)
|
|
|
|
sub_stderr = RedirectText(sys.stderr, filter_lines=filter_lines)
|
|
|
|
try:
|
|
return subprocess.call(cmd,
|
|
stdout=sub_stdout,
|
|
stderr=sub_stderr)
|
|
except Exception as err: # pylint: disable=broad-except
|
|
_LOGGER.error(u"Running command failed: %s", err)
|
|
_LOGGER.error(u"Please try running %s locally.", full_cmd)
|
|
finally:
|
|
if capture_stdout:
|
|
# pylint: disable=lost-exception
|
|
return sub_stdout.getvalue()
|
|
|
|
|
|
def is_dev_esphome_version():
|
|
return 'dev' in const.__version__
|
|
|
|
|
|
# Custom OrderedDict with nicer repr method for debugging
|
|
class OrderedDict(collections.OrderedDict):
|
|
def __repr__(self):
|
|
return dict(self).__repr__()
|
|
|
|
def move_to_end(self, key, last=True):
|
|
if IS_PY2:
|
|
if len(self) == 1:
|
|
return
|
|
if last:
|
|
# When moving to end, just pop and re-add
|
|
val = self.pop(key)
|
|
self[key] = val
|
|
else:
|
|
# When moving to front, use internals here
|
|
# https://stackoverflow.com/a/16664932
|
|
root = self._OrderedDict__root # pylint: disable=no-member
|
|
first = root[1]
|
|
link = self._OrderedDict__map[key] # pylint: disable=no-member
|
|
link_prev, link_next, _ = link
|
|
link_prev[1] = link_next
|
|
link_next[0] = link_prev
|
|
link[0] = root
|
|
link[1] = first
|
|
root[1] = first[0] = link
|
|
else:
|
|
super(OrderedDict, self).move_to_end(key, last=last) # pylint: disable=no-member
|
|
|
|
|
|
def list_yaml_files(folder):
|
|
files = filter_yaml_files([os.path.join(folder, p) for p in os.listdir(folder)])
|
|
files.sort()
|
|
return files
|
|
|
|
|
|
def filter_yaml_files(files):
|
|
files = [f for f in files if os.path.splitext(f)[1] == '.yaml']
|
|
files = [f for f in files if os.path.basename(f) != 'secrets.yaml']
|
|
files = [f for f in files if not os.path.basename(f).startswith('.')]
|
|
return files
|