diff --git a/colcon_notification/event_handler/status.py b/colcon_notification/event_handler/status.py index e4062df..8027e4a 100644 --- a/colcon_notification/event_handler/status.py +++ b/colcon_notification/event_handler/status.py @@ -7,6 +7,8 @@ import sys import time +import colorama + from colcon_core.event.job import JobEnded from colcon_core.event.job import JobProgress from colcon_core.event.job import JobQueued @@ -20,6 +22,19 @@ from colcon_core.subprocess import SIGINT_RESULT +def _colorama_str_len(s): + in_code = False + l = 1 + for c in s: + if c == '\033': + in_code = True + elif c == 'm' and in_code: + in_code = False + elif not in_code: + l += 1 + return l + + class StatusEventHandler(EventHandlerExtensionPoint): """ Continuously update a status line. @@ -42,6 +57,7 @@ class StatusEventHandler(EventHandlerExtensionPoint): def __init__(self): # noqa: D107 super().__init__() + colorama.init() satisfies_version( EventHandlerExtensionPoint.EXTENSION_POINT_VERSION, '^1.0') @@ -136,33 +152,42 @@ def __call__(self, event): # noqa: D102 # runtime in seconds duration_string = format_duration( now - self._start_time, fixed_decimal_points=1) - blocks.append('[{duration_string}]'.format_map(locals())) + blocks.append('[' + colorama.Fore.YELLOW + duration_string + + colorama.Fore.RESET + ']') # number of completed jobs / number of jobs blocks.append( - '[%d/%d complete]' % - (len(self._ended), self._queued_count)) + '[' + colorama.Style.BRIGHT + colorama.Fore.GREEN + + str(len(self._ended)) + colorama.Style.RESET_ALL + '/' + + colorama.Fore.GREEN + str(self._queued_count) + + colorama.Fore.RESET + ' complete]') # number of failed jobs if not zero failed_jobs = [ j for j, d in self._ended.items() if d['rc'] and d['rc'] != SIGINT_RESULT] if failed_jobs: - blocks.append('[%d failed]' % len(failed_jobs)) + blocks.append('[' + colorama.Fore.RED + colorama.Style.BRIGHT + + str(len(failed_jobs)) + colorama.Style.NORMAL + + ' failed' + colorama.Fore.RESET + ']') # number of ongoing jobs if greater one if len(self._running) > 1: - blocks.append('[%d ongoing]' % len(self._running)) + blocks.append('[' + colorama.Fore.GREEN + + str(len(self._running)) + colorama.Fore.RESET + + ' ongoing]') # job identifier, label and time for ongoing jobs for job, d in self._running.items(): - msg = job.task.context.pkg.name + msg = (colorama.Fore.CYAN + job.task.context.pkg.name + + colorama.Fore.RESET) if 'progress' in d: - msg += ':%s' % ' '.join(d['progress']) + msg += (':' + colorama.Fore.BLUE + ' '.join(d['progress']) + + colorama.Fore.RESET) duration_string = format_duration( now - d['start_time'], fixed_decimal_points=1) - blocks.append( - '[{msg} - {duration_string}]'.format_map(locals())) + blocks.append('[' + msg + ' - ' + colorama.Fore.YELLOW + + duration_string + colorama.Fore.RESET + ']') # determine blocks which fit into terminal width max_width = shutil.get_terminal_size().columns @@ -171,13 +196,13 @@ def __call__(self, event): # noqa: D102 # append dots when skipping at least one block if i < len(blocks) - 1: msg += ' ...' - if len(msg) < max_width: + if _colorama_str_len(msg) < max_width: break else: return print(msg, end='\r') - self._last_status_line_length = len(msg) + self._last_status_line_length = _colorama_str_len(msg) elif isinstance(data, EventReactorShutdown): self._clear_last_status_line() diff --git a/setup.cfg b/setup.cfg index 03c9138..443a505 100644 --- a/setup.cfg +++ b/setup.cfg @@ -28,6 +28,7 @@ keywords = colcon python_requires = >=3.6 install_requires = colcon-core>=0.3.7 + colorama notify2; sys_platform == 'linux' pywin32; sys_platform == 'win32' packages = find: diff --git a/stdeb.cfg b/stdeb.cfg index 02a6606..2ffed8d 100644 --- a/stdeb.cfg +++ b/stdeb.cfg @@ -1,6 +1,5 @@ [colcon-notification] No-Python2: -Build-Depends: python3-setuptools (>= 53.1.0) Depends3: python3-colcon-core (>= 0.3.7), python3-notify2 Suite: focal jammy noble bookworm trixie X-Python3-Version: >= 3.6