lib/oe/log_colorizer.py: add LogColorizerProxyProgressHandler

This progress handler intercepts log output, stripping any ANSII color
escape codes. Then the stripped output is fed to the underlying progress
handler which will render the progress bar as usual.

Signed-off-by: Chris Laplante <chris.laplante@agilent.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>

1 files changed, 86 insertions, 0 deletions

diff --git a/meta/lib/oe/log_colorizer.py b/meta/lib/oe/log_colorizer.py
new file mode 100644
index 0000000000..10021ef880
--- /dev/null
+++ b/meta/lib/oe/log_colorizer.py
@@ -0,0 +1,86 @@
+# Copyright (C) 2020  Agilent Technologies, Inc.
+# Author: Chris Laplante <chris.laplante@agilent.com>
+#
+# Released under the MIT license (see COPYING.MIT)
+
+from bb.progress import ProgressHandler
+from bb.build import create_progress_handler
+import contextlib
+import re
+import os.path
+
+# from https://stackoverflow.com/a/14693789/221061
+ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
+
+
+class LogColorizerProxyProgressHandler(ProgressHandler):
+    """
+    This is a proxy progress handler. It intercepts log output, stripping any
+    ANSII color escape codes. Then the stripped output is fed to the task's
+    original progress handler which will render the progress bar as usual.
+    """
+    def __init__(self, d, outfile=None, otherargs=None):
+        self._task = d.getVar("BB_RUNTASK")
+        self._color_log = None
+        self._nocolor_log = None
+        self._exit_stack = None
+        self._original_ph = None
+
+        self._suppress_color_output = not not d.getVar("LOG_COLORIZER_SUPPRESS_COLORIZED_OUTPUT")
+
+        self._tempdir = d.getVar("T")
+        logfmt = (d.getVar('BB_LOGFMT') or 'log.{task}.{pid}')
+        self._logbasename = logfmt.format(task=self._task, pid=os.getpid())
+
+        # Setup courtesy symlinks
+        for suffix in [".nocolor", ".color"]:
+            loglink = os.path.join(self._tempdir, 'log.{0}{1}'.format(self._task, suffix))
+            logfn = os.path.join(self._tempdir, self._logbasename + suffix)
+            if loglink:
+                bb.utils.remove(loglink)
+
+                try:
+                    os.symlink(self._logbasename + suffix, loglink)
+                except OSError:
+                    pass
+
+        super().__init__(d, outfile)
+
+    def __enter__(self):
+        with contextlib.ExitStack() as es:
+            self._color_log = es.enter_context(open(os.path.join(self._tempdir, self._logbasename) + ".color", "w"))
+            self._nocolor_log = es.enter_context(open(os.path.join(self._tempdir, self._logbasename) + ".nocolor", "w"))
+
+            # Reconstitute the original progress handler. We will feed stripped output to it so
+            # that the progress bar still shows up for the task.
+            original_ph = self._data.getVarFlag(self._task, "originalprogress")
+            if original_ph:
+                # We don't want task output showing up on the screen twice, so we'll just have
+                # the original progress handler write to /dev/null.
+                # Note the progress handler itself is responsible for closing the devnull handler.
+                devnull = open("/dev/null", "w")
+                self._original_ph = es.enter_context(create_progress_handler(self._task, original_ph, devnull, self._data))
+
+            self._exit_stack = es.pop_all()
+        super().__enter__()
+
+    def __exit__(self, *exc_info):
+        if self._exit_stack:
+            self._exit_stack.__exit__(*exc_info)
+        super().__exit__(*exc_info)
+
+    def write(self, string):
+        # Filter out ANSI escape sequences using the regular expression
+        filtered = ansi_escape.sub('', string)
+
+        if self._color_log:
+            self._color_log.write(string)
+
+        if self._nocolor_log:
+            self._nocolor_log.write(filtered)
+
+        if self._original_ph:
+            # Pass-through to the original progress handler so we get our progress bar
+            self._original_ph.write(filtered)
+
+        super().write(filtered if self._suppress_color_output else string)