Initial implementation (#1)

colcon_rerun/__init__.py

@@ -1,4 +1,5 @@
 # Copyright 2022 Scott K Logan
 # Licensed under the Apache License, Version 2.0
 
-__version__ = '0.0.0'
+__version__ = '0.0.1'
+CONFIG_NAME = __name__ + '.yaml'

colcon_rerun/argument_parser/rerun.py

@@ -0,0 +1,45 @@
+# Copyright 2022 Scott K Logan
+# Licensed under the Apache License, Version 2.0
+
+import sys
+
+from colcon_core.argument_parser import ArgumentParserDecorator
+from colcon_core.argument_parser import ArgumentParserDecoratorExtensionPoint
+from colcon_core.logging import colcon_logger
+from colcon_core.plugin_system import satisfies_version
+from colcon_rerun.config import update_config
+
+logger = colcon_logger.getChild(__name__)
+
+
+class ReRunArgumentParserDecorator(ArgumentParserDecoratorExtensionPoint):
+    """Capture arguments for recently executed verbs."""
+
+    # High priority to capture the arguments as they were passed,
+    # before being modified by other parsers
+    PRIORITY = 500
+
+    def __init__(self):  # noqa: D107
+        super().__init__()
+        satisfies_version(
+            ArgumentParserDecoratorExtensionPoint.EXTENSION_POINT_VERSION,
+            '^1.0')
+
+    def decorate_argument_parser(self, *, parser):  # noqa: D102
+        return ReRunArgumentDecorator(parser)
+
+
+class ReRunArgumentDecorator(ArgumentParserDecorator):
+    """Capture arguments for recently executed verbs."""
+
+    def parse_args(self, *args, **kwargs):  # noqa: D102
+        parsed_args = self._parser.parse_args(*args, **kwargs)
+        if getattr(parsed_args, 'verb_name', None) not in (None, 'rerun'):
+            raw_args = kwargs.get('args')
+            if not raw_args:
+                if args:
+                    raw_args = args[0]
+                else:
+                    raw_args = sys.argv[1:]
+            update_config(parsed_args.verb_name, raw_args)
+        return parsed_args

colcon_rerun/config.py

@@ -0,0 +1,49 @@
+# Copyright 2022 Scott K Logan
+# Licensed under the Apache License, Version 2.0
+
+from colcon_core.location import get_config_path
+from colcon_rerun import CONFIG_NAME
+from colcon_rerun.logging import configure_filelock_logger
+import filelock
+import yaml
+
+
+def get_config():
+    """Get the global colcon-rerun configuration."""
+    configure_filelock_logger()
+
+    config_path = get_config_path()
+    config_path.mkdir(parents=True, exist_ok=True)
+    config_file = config_path / CONFIG_NAME
+    lock_file = config_path / '.{}.lock'.format(CONFIG_NAME)
+    try:
+        with filelock.FileLock(lock_file, timeout=5):
+            with config_file.open() as f:
+                return yaml.safe_load(f) or {}
+    except FileNotFoundError:
+        return {}
+
+
+def update_config(verb_name, commands):
+    """Update the global colcon-rerun configuration."""
+    configure_filelock_logger()
+
+    config_path = get_config_path()
+    config_path.mkdir(parents=True, exist_ok=True)
+    config_file = config_path / CONFIG_NAME
+    lock_file = config_path / '.{}.lock'.format(CONFIG_NAME)
+    with filelock.FileLock(lock_file, timeout=5):
+        with config_file.open('a+') as f:
+            f.seek(0)
+            config = yaml.safe_load(f) or {}
+            config.setdefault('full_captures', {})
+            if (
+                config['full_captures'].get(verb_name, []) == commands and
+                config.get('last_verb') == verb_name
+            ):
+                return
+            config['full_captures'][verb_name] = commands
+            config['last_verb'] = verb_name
+            f.seek(0)
+            f.truncate()
+            yaml.dump(config, f, default_style="'")

colcon_rerun/logging.py

@@ -0,0 +1,19 @@
+# Copyright 2022 Scott K Logan
+# Licensed under the Apache License, Version 2.0
+
+import logging
+
+from colcon_core.logging import colcon_logger
+
+
+def _get_effective_log_level():
+    for handler in colcon_logger.handlers:
+        if isinstance(handler, logging.StreamHandler):
+            return handler.level
+    return logging.WARNING
+
+
+def configure_filelock_logger():
+    """Configure the 'filelock' log level based on colcon's log level."""
+    log_level = _get_effective_log_level()
+    logging.getLogger('filelock').setLevel(log_level)

colcon_rerun/verb/rerun.py

@@ -0,0 +1,79 @@
+# Copyright 2022 Scott K Logan
+# Licensed under the Apache License, Version 2.0
+
+from collections import OrderedDict
+import os
+
+from colcon_core.command import add_subparsers
+from colcon_core.command import CommandContext
+from colcon_core.command import create_parser
+from colcon_core.command import verb_main
+from colcon_core.entry_point import EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE
+from colcon_core.logging import colcon_logger
+from colcon_core.plugin_system import satisfies_version
+from colcon_core.verb import get_verb_extensions
+from colcon_core.verb import VerbExtensionPoint
+from colcon_rerun.config import get_config
+
+logger = colcon_logger.getChild(__name__)
+
+
+def _disable_capture():
+    blocklist = os.environ.get(
+        EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE.name, '')
+    if blocklist:
+        blocklist += os.pathsep
+    blocklist += 'colcon_core.argument_parser.rerun'
+    os.environ[EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE.name] = blocklist
+
+
+def _invoke_verb(command_name, verb_name, argv):
+    print(
+        'Re-running previous command: {} {}'.format(
+            command_name, ' '.join(argv)))
+
+    parser = create_parser('colcon_core.environment_variable')
+    verb_extensions = OrderedDict(
+        pair for pair in get_verb_extensions().items()
+        if pair[0] == verb_name)
+    if verb_extensions:
+        add_subparsers(
+            parser, command_name, verb_extensions, attribute='verb_name')
+
+    args = parser.parse_args(args=argv)
+    context = CommandContext(command_name=command_name, args=args)
+
+    return verb_main(context, colcon_logger)
+
+
+class ReRunVerb(VerbExtensionPoint):
+    """Quickly re-run a recently executed verb."""
+
+    def __init__(self):  # noqa: D107
+        super().__init__()
+        satisfies_version(VerbExtensionPoint.EXTENSION_POINT_VERSION, '^1.0')
+
+    def add_arguments(self, *, parser):  # noqa: D102
+        parser.add_argument(
+            'verb_to_run', nargs='?',
+            help='The verb to re-run (default: most recent verb)')
+        parser.add_argument(
+            'additional_args', nargs='*', type=str.lstrip, default=[],
+            help='Additional arguments to pass to the command')
+
+    def main(self, *, context):  # noqa: D102
+        config_content = get_config()
+
+        if not context.args.verb_to_run:
+            context.args.verb_to_run = config_content.get('last_verb')
+            if not context.args.verb_to_run:
+                raise RuntimeError(
+                    'No previously recorded invocation to re-run')
+
+        _disable_capture()
+
+        argv = config_content.get('full_captures', {}).get(
+            context.args.verb_to_run, [context.args.verb_to_run])
+        argv += context.args.additional_args
+        return _invoke_verb(
+            context.command_name, context.args.verb_to_run, argv)

setup.cfg

@@ -26,6 +26,8 @@ keywords = colcon
 [options]
 install_requires =
   colcon-core
+  filelock
+  PyYAML
 packages = find:
 zip_safe = true
 
@@ -56,6 +58,10 @@ filterwarnings =
 junit_suite_name = colcon-rerun
 
 [options.entry_points]
+colcon_core.argument_parser =
+    rerun = colcon_rerun.argument_parser.rerun:ReRunArgumentParserDecorator
+colcon_core.verb =
+    rerun = colcon_rerun.verb.rerun:ReRunVerb
 
 [flake8]
 import-order-style = google

stdeb.cfg

@@ -1,5 +1,5 @@
 [colcon-rerun]
 No-Python2:
-Depends3: python3-colcon-core
+Depends3: python3-colcon-core, python3-filelock, python3-yaml
 Suite: bionic focal jammy stretch buster bullseye
 X-Python3-Version: >= 3.5

test/spell_check.words

@@ -1,9 +1,17 @@
 apache
+blocklist
 colcon
+filelock
 iterdir
+lstrip
+nargs
+noqa
 pathlib
+plugin
 pytest
 scott
 scspell
 setuptools
+subparsers
 thomas
+yaml