<filename>colcon_cache/subverb/lock.py
# Copyright 2016-2018 <NAME>
# Copyright 2021 <NAME>
# Licensed under the Apache License, Version 2.0

from collections import OrderedDict
import os
import os.path
from pathlib import Path

from colcon_cache.subverb import CacheSubverbExtensionPoint
from colcon_core.argument_parser.destination_collector \
    import DestinationCollectorDecorator
from colcon_core.event.job import JobUnselected
from colcon_core.event_handler import add_event_handler_arguments
from colcon_core.executor import add_executor_arguments
from colcon_core.executor import execute_jobs
from colcon_core.executor import Job
from colcon_core.executor import OnError
from colcon_core.package_identification.ignore import IGNORE_MARKER
from colcon_core.package_selection import add_arguments \
    as add_packages_arguments
from colcon_core.package_selection import get_packages
from colcon_core.plugin_system import satisfies_version
from colcon_core.task import add_task_arguments
from colcon_core.task import get_task_extension
from colcon_core.task import TaskContext
from colcon_core.verb import check_and_mark_build_tool
from colcon_core.verb import logger
from colcon_core.verb import update_object


class LockCachePackageArguments:
    """Arguments to cache lock a specific package."""

    def __init__(self, pkg, args, *, additional_destinations=None):
        """
        Construct a LockCachePackageArguments.

        :param pkg: The package descriptor
        :param args: The parsed command line arguments
        :param list additional_destinations: The destinations of additional
          arguments
        """
        super().__init__()
        self.path = os.path.abspath(
            os.path.join(os.getcwd(), str(pkg.path)))
        self.build_base = os.path.abspath(os.path.join(
            os.getcwd(), args.build_base, pkg.name))
        self.ignore_dependencies = args.ignore_dependencies

        # set additional arguments
        for dest in (additional_destinations or []):
            # from the command line
            if hasattr(args, dest):
                update_object(
                    self, dest, getattr(args, dest),
                    pkg.name, 'cache lock', 'command line')
            # from the package metadata
            if dest in pkg.metadata:
                update_object(
                    self, dest, pkg.metadata[dest],
                    pkg.name, 'cache lock', 'package metadata')


class LockCacheSubverb(CacheSubverbExtensionPoint):
    """Lock current cache for packages."""

    def __init__(self):  # noqa: D107
        super().__init__()
        satisfies_version(
            CacheSubverbExtensionPoint.EXTENSION_POINT_VERSION, '^1.0')

    def add_arguments(self, *, parser):  # noqa: D102
        parser.add_argument(
            '--build-base',
            default='build',
            help='The base path for all build directories (default: build)')

        parser.add_argument(
            '--ignore-dependencies',
            action='store_true',
            help='Ignore dependencies when capturing caches (default: false)')

        add_executor_arguments(parser)
        add_event_handler_arguments(parser)
        add_packages_arguments(parser)

        decorated_parser = DestinationCollectorDecorator(parser)
        add_task_arguments(decorated_parser, 'colcon_cache.task.lock')
        self.task_argument_destinations = decorated_parser.get_destinations()

    def main(self, *, context):  # noqa: D102
        check_and_mark_build_tool(context.args.build_base)

        self._create_paths(context.args)

        decorators = get_packages(
            context.args,
            additional_argument_names=self.task_argument_destinations,
            recursive_categories=('run', ))

        jobs, unselected_packages = self._get_jobs(
            context.args, decorators)

        def post_unselected_packages(*, event_queue):
            nonlocal unselected_packages
            names = [pkg.name for pkg in unselected_packages]
            for name in sorted(names):
                event_queue.put(
                    (JobUnselected(name), None))

        on_error = OnError.continue_
        rc = execute_jobs(
            context, jobs, on_error=on_error,
            pre_execution_callback=post_unselected_packages)

        return rc

    def _create_paths(self, args):
        self._create_path(args.build_base)

    def _create_path(self, path):
        path = Path(os.path.abspath(path))
        if not path.exists():
            path.mkdir(parents=True, exist_ok=True)
        ignore_marker = path / IGNORE_MARKER
        if not os.path.lexists(str(ignore_marker)):
            with ignore_marker.open('w'):
                pass

    def _get_jobs(self, args, decorators):
        jobs = OrderedDict()
        unselected_packages = set()
        for decorator in decorators:
            pkg = decorator.descriptor

            if not decorator.selected:
                unselected_packages.add(pkg)
                continue

            extension = get_task_extension(
                'colcon_cache.task.lock', pkg.metadata['vcs_type'])
            if not extension:
                logger.warning(
                    "No task extension to 'cache lock' "
                    "a '{pkg.type}' package"
                    .format_map(locals()))
                continue

            recursive_dependencies = OrderedDict()
            for dep_name in decorator.recursive_dependencies:
                dep_path = Path(args.build_base) / dep_name
                recursive_dependencies[dep_name] = dep_path

            package_args = LockCachePackageArguments(
                pkg, args, additional_destinations=self
                .task_argument_destinations.values())
            ordered_package_args = ', '.join([
                ('%s: %s' % (repr(k), repr(package_args.__dict__[k])))
                for k in sorted(package_args.__dict__.keys())
            ])
            logger.debug(
                "Staging package '{pkg.name}' with the following arguments: "
                '{{{ordered_package_args}}}'.format_map(locals()))
            task_context = TaskContext(
                pkg=pkg, args=package_args,
                dependencies=recursive_dependencies)

            job = Job(
                identifier=pkg.name,
                dependencies=set(recursive_dependencies.keys()),
                task=extension, task_context=task_context)

            jobs[pkg.name] = job
        return jobs, unselected_packages

<filename>colcon_cache/subverb/lock.py
# Copyright 2016-2018 <NAME>
# Copyright 2021 <NAME>
# Licensed under the Apache License, Version 2.0

from collections import OrderedDict
import os
import os.path
from pathlib import Path

from colcon_cache.subverb import CacheSubverbExtensionPoint
from colcon_core.argument_parser.destination_collector \
    import DestinationCollectorDecorator
from colcon_core.event.job import JobUnselected
from colcon_core.event_handler import add_event_handler_arguments
from colcon_core.executor import add_executor_arguments
from colcon_core.executor import execute_jobs
from colcon_core.executor import Job
from colcon_core.executor import OnError
from colcon_core.package_identification.ignore import IGNORE_MARKER
from colcon_core.package_selection import add_arguments \
    as add_packages_arguments
from colcon_core.package_selection import get_packages
from colcon_core.plugin_system import satisfies_version
from colcon_core.task import add_task_arguments
from colcon_core.task import get_task_extension
from colcon_core.task import TaskContext
from colcon_core.verb import check_and_mark_build_tool
from colcon_core.verb import logger
from colcon_core.verb import update_object


class LockCachePackageArguments:
    """Arguments to cache lock a specific package."""

    def __init__(self, pkg, args, *, additional_destinations=None):
        """
        Construct a LockCachePackageArguments.

        :param pkg: The package descriptor
        :param args: The parsed command line arguments
        :param list additional_destinations: The destinations of additional
          arguments
        """
        super().__init__()
        self.path = os.path.abspath(
            os.path.join(os.getcwd(), str(pkg.path)))
        self.build_base = os.path.abspath(os.path.join(
            os.getcwd(), args.build_base, pkg.name))
        self.ignore_dependencies = args.ignore_dependencies

        # set additional arguments
        for dest in (additional_destinations or []):
            # from the command line
            if hasattr(args, dest):
                update_object(
                    self, dest, getattr(args, dest),
                    pkg.name, 'cache lock', 'command line')
            # from the package metadata
            if dest in pkg.metadata:
                update_object(
                    self, dest, pkg.metadata[dest],
                    pkg.name, 'cache lock', 'package metadata')


class LockCacheSubverb(CacheSubverbExtensionPoint):
    """Lock current cache for packages."""

    def __init__(self):  # noqa: D107
        super().__init__()
        satisfies_version(
            CacheSubverbExtensionPoint.EXTENSION_POINT_VERSION, '^1.0')

    def add_arguments(self, *, parser):  # noqa: D102
        parser.add_argument(
            '--build-base',
            default='build',
            help='The base path for all build directories (default: build)')

        parser.add_argument(
            '--ignore-dependencies',
            action='store_true',
            help='Ignore dependencies when capturing caches (default: false)')

        add_executor_arguments(parser)
        add_event_handler_arguments(parser)
        add_packages_arguments(parser)

        decorated_parser = DestinationCollectorDecorator(parser)
        add_task_arguments(decorated_parser, 'colcon_cache.task.lock')
        self.task_argument_destinations = decorated_parser.get_destinations()

    def main(self, *, context):  # noqa: D102
        check_and_mark_build_tool(context.args.build_base)

        self._create_paths(context.args)

        decorators = get_packages(
            context.args,
            additional_argument_names=self.task_argument_destinations,
            recursive_categories=('run', ))

        jobs, unselected_packages = self._get_jobs(
            context.args, decorators)

        def post_unselected_packages(*, event_queue):
            nonlocal unselected_packages
            names = [pkg.name for pkg in unselected_packages]
            for name in sorted(names):
                event_queue.put(
                    (JobUnselected(name), None))

        on_error = OnError.continue_
        rc = execute_jobs(
            context, jobs, on_error=on_error,
            pre_execution_callback=post_unselected_packages)

        return rc

    def _create_paths(self, args):
        self._create_path(args.build_base)

    def _create_path(self, path):
        path = Path(os.path.abspath(path))
        if not path.exists():
            path.mkdir(parents=True, exist_ok=True)
        ignore_marker = path / IGNORE_MARKER
        if not os.path.lexists(str(ignore_marker)):
            with ignore_marker.open('w'):
                pass

    def _get_jobs(self, args, decorators):
        jobs = OrderedDict()
        unselected_packages = set()
        for decorator in decorators:
            pkg = decorator.descriptor

            if not decorator.selected:
                unselected_packages.add(pkg)
                continue

            extension = get_task_extension(
                'colcon_cache.task.lock', pkg.metadata['vcs_type'])
            if not extension:
                logger.warning(
                    "No task extension to 'cache lock' "
                    "a '{pkg.type}' package"
                    .format_map(locals()))
                continue

            recursive_dependencies = OrderedDict()
            for dep_name in decorator.recursive_dependencies:
                dep_path = Path(args.build_base) / dep_name
                recursive_dependencies[dep_name] = dep_path

            package_args = LockCachePackageArguments(
                pkg, args, additional_destinations=self
                .task_argument_destinations.values())
            ordered_package_args = ', '.join([
                ('%s: %s' % (repr(k), repr(package_args.__dict__[k])))
                for k in sorted(package_args.__dict__.keys())
            ])
            logger.debug(
                "Staging package '{pkg.name}' with the following arguments: "
                '{{{ordered_package_args}}}'.format_map(locals()))
            task_context = TaskContext(
                pkg=pkg, args=package_args,
                dependencies=recursive_dependencies)

            job = Job(
                identifier=pkg.name,
                dependencies=set(recursive_dependencies.keys()),
                task=extension, task_context=task_context)

            jobs[pkg.name] = job
        return jobs, unselected_packages

# Copyright 2016-2018 <NAME> # Copyright 2021 <NAME> # Licensed under the Apache License, Version 2.0 Arguments to cache lock a specific package. Construct a LockCachePackageArguments.          :param pkg: The package descriptor         :param args: The parsed command line arguments         :param list additional_destinations: The destinations of additional           arguments # set additional arguments # from the command line # from the package metadata Lock current cache for packages. # noqa: D107 # noqa: D102 # noqa: D102