Fix colorlog removing colors and refactor color code (#1671)

esphome/__main__.py

@@ -19,7 +19,7 @@ from esphome.const import (
     CONF_PLATFORMIO_OPTIONS,
 )
 from esphome.core import CORE, EsphomeError, coroutine, coroutine_with_priority
-from esphome.helpers import color, indent
+from esphome.helpers import indent
 from esphome.util import (
     run_external_command,
     run_external_process,
@@ -27,6 +27,7 @@ from esphome.util import (
     list_yaml_files,
     get_serial_ports,
 )
+from esphome.log import color, setup_log, Fore
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -57,7 +58,7 @@ def choose_prompt(options):
                 raise ValueError
             break
         except ValueError:
-            safe_print(color("red", f"Invalid option: '{opt}'"))
+            safe_print(color(Fore.RED, f"Invalid option: '{opt}'"))
     return options[opt - 1][1]
 
 
@@ -263,46 +264,6 @@ def clean_mqtt(config, args):
     )
 
 
-def setup_log(debug=False, quiet=False):
-    if debug:
-        log_level = logging.DEBUG
-        CORE.verbose = True
-    elif quiet:
-        log_level = logging.CRITICAL
-    else:
-        log_level = logging.INFO
-    logging.basicConfig(level=log_level)
-    fmt = "%(levelname)s %(message)s"
-    colorfmt = f"%(log_color)s{fmt}%(reset)s"
-    datefmt = "%H:%M:%S"
-
-    logging.getLogger("urllib3").setLevel(logging.WARNING)
-
-    try:
-        import colorama
-
-        colorama.init(strip=True)
-
-        from colorlog import ColoredFormatter
-
-        logging.getLogger().handlers[0].setFormatter(
-            ColoredFormatter(
-                colorfmt,
-                datefmt=datefmt,
-                reset=True,
-                log_colors={
-                    "DEBUG": "cyan",
-                    "INFO": "green",
-                    "WARNING": "yellow",
-                    "ERROR": "red",
-                    "CRITICAL": "red",
-                },
-            )
-        )
-    except ImportError:
-        pass
-
-
 def command_wizard(args):
     from esphome import wizard
 
@@ -442,30 +403,30 @@ def command_update_all(args):
         click.echo(f"{half_line}{middle_text}{half_line}")
 
     for f in files:
-        print("Updating {}".format(color("cyan", f)))
+        print("Updating {}".format(color(Fore.CYAN, f)))
         print("-" * twidth)
         print()
         rc = run_external_process(
             "esphome", "--dashboard", f, "run", "--no-logs", "--upload-port", "OTA"
         )
         if rc == 0:
-            print_bar("[{}] {}".format(color("bold_green", "SUCCESS"), f))
+            print_bar("[{}] {}".format(color(Fore.BOLD_GREEN, "SUCCESS"), f))
             success[f] = True
         else:
-            print_bar("[{}] {}".format(color("bold_red", "ERROR"), f))
+            print_bar("[{}] {}".format(color(Fore.BOLD_RED, "ERROR"), f))
             success[f] = False
 
         print()
         print()
         print()
 
-    print_bar("[{}]".format(color("bold_white", "SUMMARY")))
+    print_bar("[{}]".format(color(Fore.BOLD_WHITE, "SUMMARY")))
     failed = 0
     for f in files:
         if success[f]:
-            print("  - {}: {}".format(f, color("green", "SUCCESS")))
+            print("  - {}: {}".format(f, color(Fore.GREEN, "SUCCESS")))
         else:
-            print("  - {}: {}".format(f, color("bold_red", "FAILED")))
+            print("  - {}: {}".format(f, color(Fore.BOLD_RED, "FAILED")))
             failed += 1
     return failed
 

esphome/api/client.py

@@ -13,7 +13,8 @@ from esphome import const
 import esphome.api.api_pb2 as pb
 from esphome.const import CONF_PASSWORD, CONF_PORT
 from esphome.core import EsphomeError
-from esphome.helpers import resolve_ip_address, indent, color
+from esphome.helpers import resolve_ip_address, indent
+from esphome.log import color, Fore
 from esphome.util import safe_print
 
 _LOGGER = logging.getLogger(__name__)
@@ -488,7 +489,7 @@ def run_logs(config, address):
         text = msg.message
         if msg.send_failed:
             text = color(
-                "white",
+                Fore.WHITE,
                 "(Message skipped because it was too big to fit in "
                 "TCP buffer - This is only cosmetic)",
             )

esphome/config.py

@@ -19,13 +19,14 @@ from esphome.const import (
     CONF_SUBSTITUTIONS,
 )
 from esphome.core import CORE, EsphomeError  # noqa
-from esphome.helpers import color, indent
+from esphome.helpers import indent
 from esphome.util import safe_print, OrderedDict
 
 from typing import List, Optional, Tuple, Union  # noqa
 from esphome.core import ConfigType  # noqa
 from esphome.yaml_util import is_secret, ESPHomeDataBase, ESPForceValue
 from esphome.voluptuous_schema import ExtraKeysInvalid
+from esphome.log import color, Fore
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -790,7 +791,7 @@ def line_info(config, path, highlight=True):
     if obj:
         mark = obj.start_mark
         source = "[source {}:{}]".format(mark.document, mark.line + 1)
-        return color("cyan", source)
+        return color(Fore.CYAN, source)
     return "None"
 
 
@@ -813,7 +814,9 @@ def dump_dict(config, path, at_root=True):
     if at_root:
         error = config.get_error_for_path(path)
         if error is not None:
-            ret += "\n" + color("bold_red", _format_vol_invalid(error, config)) + "\n"
+            ret += (
+                "\n" + color(Fore.BOLD_RED, _format_vol_invalid(error, config)) + "\n"
+            )
 
     if isinstance(conf, (list, tuple)):
         multiline = True
@@ -826,12 +829,14 @@ def dump_dict(config, path, at_root=True):
             error = config.get_error_for_path(path_)
             if error is not None:
                 ret += (
-                    "\n" + color("bold_red", _format_vol_invalid(error, config)) + "\n"
+                    "\n"
+                    + color(Fore.BOLD_RED, _format_vol_invalid(error, config))
+                    + "\n"
                 )
 
             sep = "- "
             if config.is_in_error_path(path_):
-                sep = color("red", sep)
+                sep = color(Fore.RED, sep)
             msg, _ = dump_dict(config, path_, at_root=False)
             msg = indent(msg)
             inf = line_info(config, path_, highlight=config.is_in_error_path(path_))
@@ -851,12 +856,14 @@ def dump_dict(config, path, at_root=True):
             error = config.get_error_for_path(path_)
             if error is not None:
                 ret += (
-                    "\n" + color("bold_red", _format_vol_invalid(error, config)) + "\n"
+                    "\n"
+                    + color(Fore.BOLD_RED, _format_vol_invalid(error, config))
+                    + "\n"
                 )
 
             st = f"{k}: "
             if config.is_in_error_path(path_):
-                st = color("red", st)
+                st = color(Fore.RED, st)
             msg, m = dump_dict(config, path_, at_root=False)
 
             inf = line_info(config, path_, highlight=config.is_in_error_path(path_))
@@ -878,7 +885,7 @@ def dump_dict(config, path, at_root=True):
         if len(conf) > 80:
             conf = "|-\n" + indent(conf)
         error = config.get_error_for_path(path)
-        col = "bold_red" if error else "white"
+        col = Fore.BOLD_RED if error else Fore.KEEP
         ret += color(col, str(conf))
     elif isinstance(conf, core.Lambda):
         if is_secret(conf):
@@ -886,13 +893,13 @@ def dump_dict(config, path, at_root=True):
 
         conf = "!lambda |-\n" + indent(str(conf.value))
         error = config.get_error_for_path(path)
-        col = "bold_red" if error else "white"
+        col = Fore.BOLD_RED if error else Fore.KEEP
         ret += color(col, conf)
     elif conf is None:
         pass
     else:
         error = config.get_error_for_path(path)
-        col = "bold_red" if error else "white"
+        col = Fore.BOLD_RED if error else Fore.KEEP
         ret += color(col, str(conf))
         multiline = "\n" in ret
 
@@ -934,13 +941,13 @@ def read_config(command_line_substitutions):
         if not CORE.verbose:
             res = strip_default_ids(res)
 
-        safe_print(color("bold_red", "Failed config"))
+        safe_print(color(Fore.BOLD_RED, "Failed config"))
         safe_print("")
         for path, domain in res.output_paths:
             if not res.is_in_error_path(path):
                 continue
 
-            errstr = color("bold_red", f"{domain}:")
+            errstr = color(Fore.BOLD_RED, f"{domain}:")
             errline = line_info(res, path)
             if errline:
                 errstr += " " + errline

esphome/helpers.py

@@ -57,17 +57,6 @@ def cpp_string_escape(string, encoding="utf-8"):
     return '"' + result + '"'
 
 
-def color(the_color, message=""):
-    from colorlog.escape_codes import escape_codes, parse_colors
-
-    if not message:
-        res = parse_colors(the_color)
-    else:
-        res = parse_colors(the_color) + message + escape_codes["reset"]
-
-    return res
-
-
 def run_system_command(*args):
     import subprocess
 

esphome/log.py

@@ -0,0 +1,82 @@
+import logging
+
+from esphome.core import CORE
+
+
+class AnsiFore:
+    KEEP = ""
+    BLACK = "\033[30m"
+    RED = "\033[31m"
+    GREEN = "\033[32m"
+    YELLOW = "\033[33m"
+    BLUE = "\033[34m"
+    MAGENTA = "\033[35m"
+    CYAN = "\033[36m"
+    WHITE = "\033[37m"
+    RESET = "\033[39m"
+
+    BOLD_BLACK = "\033[1;30m"
+    BOLD_RED = "\033[1;31m"
+    BOLD_GREEN = "\033[1;32m"
+    BOLD_YELLOW = "\033[1;33m"
+    BOLD_BLUE = "\033[1;34m"
+    BOLD_MAGENTA = "\033[1;35m"
+    BOLD_CYAN = "\033[1;36m"
+    BOLD_WHITE = "\033[1;37m"
+    BOLD_RESET = "\033[1;39m"
+
+
+class AnsiStyle:
+    BRIGHT = "\033[1m"
+    BOLD = "\033[1m"
+    DIM = "\033[2m"
+    THIN = "\033[2m"
+    NORMAL = "\033[22m"
+    RESET_ALL = "\033[0m"
+
+
+Fore = AnsiFore()
+Style = AnsiStyle()
+
+
+def color(col: str, msg: str, reset: bool = True) -> bool:
+    if col and not col.startswith("\033["):
+        raise ValueError("Color must be value from esphome.log.Fore")
+    s = str(col) + msg
+    if reset and col:
+        s += str(Style.RESET_ALL)
+    return s
+
+
+class ESPHomeLogFormatter(logging.Formatter):
+    def __init__(self):
+        super().__init__(fmt="%(levelname)s %(message)s", datefmt="%H:%M:%S", style="%")
+
+    def format(self, record):
+        formatted = super().format(record)
+        prefix = {
+            "DEBUG": Fore.CYAN,
+            "INFO": Fore.GREEN,
+            "WARNING": Fore.YELLOW,
+            "ERROR": Fore.RED,
+            "CRITICAL": Fore.RED,
+        }.get(record.levelname, "")
+        return f"{prefix}{formatted}{Style.RESET_ALL}"
+
+
+def setup_log(debug=False, quiet=False):
+    import colorama
+
+    if debug:
+        log_level = logging.DEBUG
+        CORE.verbose = True
+    elif quiet:
+        log_level = logging.CRITICAL
+    else:
+        log_level = logging.INFO
+    logging.basicConfig(level=log_level)
+
+    logging.getLogger("urllib3").setLevel(logging.WARNING)
+
+    colorama.init()
+    logging.getLogger().handlers[0].setFormatter(ESPHomeLogFormatter())

esphome/mqtt.py

@@ -22,7 +22,7 @@ from esphome.const import (
     CONF_USERNAME,
 )
 from esphome.core import CORE, EsphomeError
-from esphome.helpers import color
+from esphome.log import color, Fore
 from esphome.util import safe_print
 
 _LOGGER = logging.getLogger(__name__)
@@ -158,7 +158,7 @@ def get_fingerprint(config):
 
     sha1 = hashlib.sha1(cert_der).hexdigest()
 
-    safe_print("SHA1 Fingerprint: " + color("cyan", sha1))
+    safe_print("SHA1 Fingerprint: " + color(Fore.CYAN, sha1))
     safe_print(
         "Copy the string above into mqtt.ssl_fingerprints section of {}"
         "".format(CORE.config_path)

esphome/wizard.py

@@ -6,7 +6,8 @@ import unicodedata
 import voluptuous as vol
 
 import esphome.config_validation as cv
-from esphome.helpers import color, get_bool_env, write_file
+from esphome.helpers import get_bool_env, write_file
+from esphome.log import color, Fore
 
 # pylint: disable=anomalous-backslash-in-string
 from esphome.pins import ESP32_BOARD_PINS, ESP8266_BOARD_PINS
@@ -148,13 +149,13 @@ def wizard(path):
     if not path.endswith(".yaml") and not path.endswith(".yml"):
         safe_print(
             "Please make your configuration file {} have the extension .yaml or .yml"
-            "".format(color("cyan", path))
+            "".format(color(Fore.CYAN, path))
         )
         return 1
     if os.path.exists(path):
         safe_print(
             "Uh oh, it seems like {} already exists, please delete that file first "
-            "or chose another configuration file.".format(color("cyan", path))
+            "or chose another configuration file.".format(color(Fore.CYAN, path))
         )
         return 2
     safe_print("Hi there!")
@@ -171,7 +172,7 @@ def wizard(path):
     safe_print()
     safe_print_step(1, CORE_BIG)
     safe_print(
-        "First up, please choose a " + color("green", "name") + " for your node."
+        "First up, please choose a " + color(Fore.GREEN, "name") + " for your node."
     )
     safe_print(
         "It should be a unique name that can be used to identify the device later."
@@ -179,12 +180,12 @@ def wizard(path):
     sleep(1)
     safe_print(
         "For example, I like calling the node in my living room {}.".format(
-            color("bold_white", "livingroom")
+            color(Fore.BOLD_WHITE, "livingroom")
         )
     )
     safe_print()
     sleep(1)
-    name = input(color("bold_white", "(name): "))
+    name = input(color(Fore.BOLD_WHITE, "(name): "))
 
     while True:
         try:
@@ -193,7 +194,7 @@ def wizard(path):
         except vol.Invalid:
             safe_print(
                 color(
-                    "red",
+                    Fore.RED,
                     f'Oh noes, "{name}" isn\'t a valid name. Names can only '
                     f"include numbers, lower-case letters, underscores and "
                     f"hyphens.",
@@ -202,12 +203,12 @@ def wizard(path):
             name = strip_accents(name).lower().replace(" ", "_")
             name = "".join(c for c in name if c in ALLOWED_NAME_CHARS)
             safe_print(
-                'Shall I use "{}" as the name instead?'.format(color("cyan", name))
+                'Shall I use "{}" as the name instead?'.format(color(Fore.CYAN, name))
             )
             sleep(0.5)
             name = default_input("(name [{}]): ", name)
 
-    safe_print('Great! Your node is now called "{}".'.format(color("cyan", name)))
+    safe_print('Great! Your node is now called "{}".'.format(color(Fore.CYAN, name)))
     sleep(1)
     safe_print_step(2, ESP_BIG)
     safe_print(
@@ -216,16 +217,16 @@ def wizard(path):
     )
     safe_print(
         "Are you using an "
-        + color("green", "ESP32")
+        + color(Fore.GREEN, "ESP32")
         + " or "
-        + color("green", "ESP8266")
+        + color(Fore.GREEN, "ESP8266")
         + " platform? (Choose ESP8266 for Sonoff devices)"
     )
     while True:
         sleep(0.5)
         safe_print()
         safe_print("Please enter either ESP32 or ESP8266.")
-        platform = input(color("bold_white", "(ESP32/ESP8266): "))
+        platform = input(color(Fore.BOLD_WHITE, "(ESP32/ESP8266): "))
         try:
             platform = vol.All(vol.Upper, vol.Any("ESP32", "ESP8266"))(platform)
             break
@@ -235,7 +236,7 @@ def wizard(path):
                 '"{}". Please try again.'.format(platform)
             )
     safe_print(
-        "Thanks! You've chosen {} as your platform.".format(color("cyan", platform))
+        "Thanks! You've chosen {} as your platform.".format(color(Fore.CYAN, platform))
     )
     safe_print()
     sleep(1)
@@ -250,37 +251,39 @@ def wizard(path):
         )
 
     safe_print(
-        "Next, I need to know what " + color("green", "board") + " you're using."
+        "Next, I need to know what " + color(Fore.GREEN, "board") + " you're using."
     )
     sleep(0.5)
-    safe_print("Please go to {} and choose a board.".format(color("green", board_link)))
+    safe_print(
+        "Please go to {} and choose a board.".format(color(Fore.GREEN, board_link))
+    )
     if platform == "ESP32":
-        safe_print("(Type " + color("green", "esp01_1m") + " for Sonoff devices)")
+        safe_print("(Type " + color(Fore.GREEN, "esp01_1m") + " for Sonoff devices)")
     safe_print()
     # Don't sleep because user needs to copy link
     if platform == "ESP32":
-        safe_print('For example "{}".'.format(color("bold_white", "nodemcu-32s")))
+        safe_print('For example "{}".'.format(color(Fore.BOLD_WHITE, "nodemcu-32s")))
         boards = list(ESP32_BOARD_PINS.keys())
     else:
-        safe_print('For example "{}".'.format(color("bold_white", "nodemcuv2")))
+        safe_print('For example "{}".'.format(color(Fore.BOLD_WHITE, "nodemcuv2")))
         boards = list(ESP8266_BOARD_PINS.keys())
     safe_print("Options: {}".format(", ".join(sorted(boards))))
 
     while True:
-        board = input(color("bold_white", "(board): "))
+        board = input(color(Fore.BOLD_WHITE, "(board): "))
         try:
             board = vol.All(vol.Lower, vol.Any(*boards))(board)
             break
         except vol.Invalid:
             safe_print(
-                color("red", f'Sorry, I don\'t think the board "{board}" exists.')
+                color(Fore.RED, f'Sorry, I don\'t think the board "{board}" exists.')
             )
             safe_print()
             sleep(0.25)
             safe_print()
 
     safe_print(
-        "Way to go! You've chosen {} as your board.".format(color("cyan", board))
+        "Way to go! You've chosen {} as your board.".format(color(Fore.CYAN, board))
     )
     safe_print()
     sleep(1)
@@ -291,20 +294,20 @@ def wizard(path):
     sleep(1)
     safe_print(
         "First, what's the "
-        + color("green", "SSID")
+        + color(Fore.GREEN, "SSID")
         + f" (the name) of the WiFi network {name} I should connect to?"
     )
     sleep(1.5)
-    safe_print('For example "{}".'.format(color("bold_white", "Abraham Linksys")))
+    safe_print('For example "{}".'.format(color(Fore.BOLD_WHITE, "Abraham Linksys")))
     while True:
-        ssid = input(color("bold_white", "(ssid): "))
+        ssid = input(color(Fore.BOLD_WHITE, "(ssid): "))
         try:
             ssid = cv.ssid(ssid)
             break
         except vol.Invalid:
             safe_print(
                 color(
-                    "red",
+                    Fore.RED,
                     'Unfortunately, "{}" doesn\'t seem to be a valid SSID. '
                     "Please try again.".format(ssid),
                 )
@@ -314,20 +317,20 @@ def wizard(path):
 
     safe_print(
         'Thank you very much! You\'ve just chosen "{}" as your SSID.'
-        "".format(color("cyan", ssid))
+        "".format(color(Fore.CYAN, ssid))
     )
     safe_print()
     sleep(0.75)
 
     safe_print(
         "Now please state the "
-        + color("green", "password")
+        + color(Fore.GREEN, "password")
         + " of the WiFi network so that I can connect to it (Leave empty for no password)"
     )
     safe_print()
-    safe_print('For example "{}"'.format(color("bold_white", "PASSWORD42")))
+    safe_print('For example "{}"'.format(color(Fore.BOLD_WHITE, "PASSWORD42")))
     sleep(0.5)
-    psk = input(color("bold_white", "(PSK): "))
+    psk = input(color(Fore.BOLD_WHITE, "(PSK): "))
     safe_print(
         "Perfect! WiFi is now set up (you can create static IPs and so on later)."
     )
@@ -340,12 +343,12 @@ def wizard(path):
     )
     safe_print(
         "This can be insecure if you do not trust the WiFi network. Do you want to set "
-        "a " + color("green", "password") + " for connecting to this ESP?"
+        "a " + color(Fore.GREEN, "password") + " for connecting to this ESP?"
     )
     safe_print()
     sleep(0.25)
     safe_print("Press ENTER for no password")
-    password = input(color("bold_white", "(password): "))
+    password = input(color(Fore.BOLD_WHITE, "(password): "))
 
     wizard_write(
         path=path,
@@ -359,8 +362,8 @@ def wizard(path):
 
     safe_print()
     safe_print(
-        color("cyan", "DONE! I've now written a new configuration file to ")
+        color(Fore.CYAN, "DONE! I've now written a new configuration file to ")
-        + color("bold_cyan", path)
+        + color(Fore.BOLD_CYAN, path)
     )
     safe_print()
     safe_print("Next steps:")

requirements.txt

@@ -2,7 +2,6 @@ voluptuous==0.12.1
 PyYAML==5.4.1
 paho-mqtt==1.5.1
 colorama==0.4.4
-colorlog==4.7.2
 tornado==6.1
 protobuf==3.15.7
 tzlocal==2.1