Colors rebased

bin/colcon

@@ -105,6 +105,7 @@ custom_extension_points.update({
         # there is no point in registering the extension_point extensions here
         # since they can't be queried through pkg_resources without installing
     },
+    'colcon_core.output_style': {},
     'colcon_core.package_augmentation': {
         'python': PythonPackageAugmentation,
     },

colcon_core/__init__.py

@@ -1,4 +1,4 @@
 # Copyright 2016-2020 Dirk Thomas
 # Licensed under the Apache License, Version 2.0
 
-__version__ = '0.17.0'
+__version__ = '0.19.0'

colcon_core/command.py

@@ -61,6 +61,8 @@
 from colcon_core.logging import colcon_logger  # noqa: E402
 from colcon_core.logging import get_numeric_log_level  # noqa: E402
 from colcon_core.logging import set_logger_level_from_env  # noqa: E402
+from colcon_core.output_style import add_output_style_arguments  # noqa: E402
+from colcon_core.output_style import apply_output_style  # noqa: E402
 from colcon_core.plugin_system import get_first_line_doc  # noqa: E402
 from colcon_core.verb import get_verb_extensions  # noqa: E402
 
@@ -93,7 +95,8 @@ def register_command_exit_handler(handler):
 
 def main(
     *, command_name='colcon', argv=None, verb_group_name=None,
-    environment_variable_group_name=None,
+    environment_variable_group_name=None, default_verb=None,
+    default_log_base='log',
 ):
     """
     Execute the main logic of the command.
@@ -114,13 +117,21 @@ def main(
 
     :param str command_name: The name of the command invoked
     :param list argv: The list of arguments
+    :param str verb_group_name: The extension point group name for verbs
+    :param str environment_variable_group_name: The extension point group name
+      for environment variables
+    :param Type default_verb: The verb class type to invoke if no explicit
+      verb was provided on the command line
+    :param str default_log_base: The default logging base path if the command
+      line argument isn't specified and the environment variable is not set
     :returns: The return code
     """
     try:
         return _main(
             command_name=command_name, argv=argv,
             verb_group_name=verb_group_name,
-            environment_variable_group_name=environment_variable_group_name)
+            environment_variable_group_name=environment_variable_group_name,
+            default_verb=default_verb, default_log_base=default_log_base)
     except KeyboardInterrupt:
         return signal.SIGINT
     finally:
@@ -132,6 +143,7 @@ def main(
 
 def _main(
     *, command_name, argv, verb_group_name, environment_variable_group_name,
+    default_verb, default_log_base,
 ):
     # default log level, for searchability: COLCON_LOG_LEVEL
     colcon_logger.setLevel(logging.WARNING)
@@ -151,6 +163,13 @@ def _main(
 
     parser = create_parser(environment_variable_group_name)
 
+    if default_verb is not None:
+        default_verb_instance = default_verb()
+        parser.set_defaults(
+            verb_parser=parser, verb_extension=default_verb_instance,
+            main=default_verb_instance.main)
+        add_parser_arguments(parser, default_verb_instance)
+
     verb_extensions = get_verb_extensions(group_name=verb_group_name)
 
     # add subparsers for all verb extensions but without arguments for now
@@ -163,7 +182,7 @@ def _main(
         known_args, _ = parser.parse_known_args(args=argv)
 
     # add the arguments for the requested verb
-    if known_args.verb_name:
+    if known_args.verb_name is not None:
         add_parser_arguments(known_args.verb_parser, known_args.verb_extension)
 
     args = parser.parse_args(args=argv)
@@ -175,18 +194,24 @@ def _main(
 
     colcon_logger.debug(f'Parsed command line arguments: {args}')
 
-    # error: no verb provided
-    if args.verb_name is None:
+    apply_output_style(args)
+
+    # verify that one of the verbs set the 'main' attribute to be invoked later
+    if getattr(args, 'main', None) is None:
         print(parser.format_usage())
         return 'Error: No verb provided'
 
     # set default locations for log files, for searchability: COLCON_LOG_PATH
     now = datetime.datetime.now()
     now_str = str(now)[:-7].replace(' ', '_').replace(':', '-')
+    if args.verb_name is None:
+        subdirectory = now_str
+    else:
+        subdirectory = f'{args.verb_name}_{now_str}'
     set_default_log_path(
         base_path=args.log_base,
         env_var=f'{command_name}_LOG_PATH'.upper(),
-        subdirectory=f'{args.verb_name}_{now_str}')
+        subdirectory=subdirectory, default=default_log_base)
 
     # add a file handler writing all levels if logging isn't disabled
     log_path = get_log_path()
@@ -240,9 +265,13 @@ def _parse_optional(self, arg_string):
             # the option.  As of that PR (which is in Python 3.13, and
             # backported to Python 3.12), it returns a 4-tuple.  Check for
             # either here.
+            # A similar regression later occurred when _parse_optional()
+            # started returning a list of tuples, brought in by
+            # https://github.com/python/cpython/pull/124631 .
             if result in (
                 (None, arg_string, None),
                 (None, arg_string, None, None),
+                [(None, arg_string, None, None)],
             ):
                 # in the case there the arg is classified as an unknown 'O'
                 # override that and classify it as an 'A'
@@ -264,6 +293,7 @@ def _parse_optional(self, arg_string):
     parser = decorate_argument_parser(parser)
 
     add_log_level_argument(parser)
+    add_output_style_arguments(parser)
 
     return parser
 
@@ -428,7 +458,9 @@ def create_subparser(parser, cmd_name, verb_extensions, *, attribute):
         title=f'{cmd_name} verbs',
         description='\n'.join(verbs) or None,
         dest=attribute,
-        help=f'call `{cmd_name} VERB -h` for specific help' if verbs else None,
+        help=(
+            f'call `{cmd_name} VERB -h` for specific help' if
+            verbs else argparse.SUPPRESS),
     )
     return subparser
 

colcon_core/event_handler/console_start_end.py

@@ -4,11 +4,14 @@
 import sys
 import time
 
+import colorama
+
 from colcon_core.event.job import JobEnded
 from colcon_core.event.job import JobStarted
 from colcon_core.event.test import TestFailure
 from colcon_core.event_handler import EventHandlerExtensionPoint
 from colcon_core.event_handler import format_duration
+from colcon_core.output_style import Style
 from colcon_core.plugin_system import satisfies_version
 from colcon_core.subprocess import SIGINT_RESULT
 
@@ -25,6 +28,7 @@ class ConsoleStartEndEventHandler(EventHandlerExtensionPoint):
 
     def __init__(self):  # noqa: D107
         super().__init__()
+        colorama.init()
         satisfies_version(
             EventHandlerExtensionPoint.EXTENSION_POINT_VERSION, '^1.0')
         self._start_times = {}
@@ -34,31 +38,47 @@ def __call__(self, event):  # noqa: D102
         data = event[0]
 
         if isinstance(data, JobStarted):
+            job_id = Style.PackageOrJobName(data.identifier)
             self._start_times[data.identifier] = time.monotonic()
-            print(f'Starting >>> {data.identifier}', flush=True)
+            msg = 'Starting ' + colorama.Fore.GREEN + \
+                colorama.Style.BRIGHT + Style.Pictogram('>>>') + colorama.Fore.CYAN + \
+                f' {job_id}' + colorama.Style.RESET_ALL
+            print(Style.SectionStart(msg), flush=True)
 
         elif isinstance(data, TestFailure):
             job = event[1]
             self._with_test_failures.add(job)
 
         elif isinstance(data, JobEnded):
+            job_id = Style.PackageOrJobName(data.identifier)
             duration = \
                 time.monotonic() - self._start_times[data.identifier]
-            duration_string = format_duration(duration)
+            duration_string = Style.Measurement(format_duration(duration))
             if not data.rc:
-                msg = f'Finished <<< {data.identifier} [{duration_string}]'
+                msg = colorama.Style.BRIGHT + colorama.Fore.BLACK + \
+                      'Finished '  + colorama.Fore.GREEN + Style.Pictogram('<<<') + \
+                      colorama.Style.RESET_ALL + colorama.Fore.CYAN + \
+                      f' {job_id}' + colorama.Fore.RESET + \
+                      ' [' + colorama.Fore.YELLOW + \
+                      f'{duration_string}' + colorama.Fore.RESET + ']'
                 job = event[1]
                 if job in self._with_test_failures:
-                    msg += '\t[ with test failures ]'
+                    msg += Style.Warning('\t[ with test failures ]')
                 writable = sys.stdout
 
             elif data.rc == SIGINT_RESULT:
-                msg = f'Aborted  <<< {data.identifier} [{duration_string}]'
+                msg = colorama.Style.BRIGHT + colorama.Fore.RED + \
+                      Style.Warning('Aborted') + '  ' + colorama.Style.NORMAL + \
+                      Style.Pictogram('<<<') + colorama.Fore.CYAN + \
+                      f' {job_id} [{duration_string}]' + colorama.Fore.RESET
                 writable = sys.stdout
-
             else:
-                msg = f'Failed   <<< {data.identifier} ' \
-                    f'[{duration_string}, exited with code {data.rc}]'
+                msg = Style.Critical('Failed') + '    ' + \
+                      Style.Pictogram('<<<') + colorama.Fore.CYAN + f' {job_id} ' + \
+                      colorama.Fore.CYAN + f' {job_id} [{duration_string}]' + \
+                      colorama.Fore.RESET + ' [' + colorama.Fore.RED + \
+                      Style.Error(f'exited with code {data.rc}') + \
+                      colorama.Fore.RESET + ']'
                 writable = sys.stderr
 
-            print(msg, file=writable, flush=True)
+            print(Style.SectionEnd(msg), file=writable, flush=True)

colcon_core/executor/__init__.py

@@ -141,7 +141,7 @@ class ExecutorExtensionPoint:
     """
 
     """The version of the executor extension interface."""
-    EXTENSION_POINT_VERSION = '1.0'
+    EXTENSION_POINT_VERSION = '1.1'
 
     """The default priority of executor extensions."""
     PRIORITY = 100
@@ -191,6 +191,14 @@ def execute(
         """
         raise NotImplementedError()
 
+    def put_event_into_queue(self, event):
+        """
+        Post a message event into the event queue.
+
+        :param event: The event
+        """
+        self._event_controller.get_queue().put((event, self))
+
     def _flush(self):
         if self._event_controller is None:
             return

colcon_core/location.py

@@ -211,10 +211,13 @@ def create_log_path(verb_name):
     ignore_marker.touch()
 
     # create latest symlinks
-    _create_symlink(path, path.parent / f'latest_{verb_name}')
-    _create_symlink(
-        path.parent / f'latest_{verb_name}',
-        path.parent / 'latest')
+    if verb_name is None:
+        _create_symlink(path, path.parent / 'latest')
+    else:
+        _create_symlink(path, path.parent / f'latest_{verb_name}')
+        _create_symlink(
+            path.parent / f'latest_{verb_name}',
+            path.parent / 'latest')
 
 
 def _reset_log_path_creation_global():