class FakeLock(asyncio.locks.Lock):

    """Fake lock used to mimic an asyncio.Lock but without causing synchronization"""

    def locked(self):
        return False

    async def acquire(self):
        return True

    def release(self):
        pass def expand_glob_package(pattern, all_workspace_packages):
    """Return all packages that match the pattern"""
    return [p for p in all_workspace_packages if fnmatch(p, pattern)] def format_time_delta(delta):
    """Formats a given time delta, in seconds, into a day-hour-minute-second string

    Seconds are limited to one decimal point accuracy. Days, hours, and minutes
    are not printed unless required.

    Examples:
        1.45        => 1.4 seconds
        61.45       => 1 minute and 1.4 seconds
        121.45      => 2 minutes and 1.4 seconds
        3721.45     => 1 hour 2 minutes and 1.4 seconds
        7321.45     => 2 hours 2 minutes and 1.4 seconds
        93821.45    => 1 days, 2 hours 2 minutes and 1.4 seconds

    :param delta: time delta to format, in seconds
    :type delta: float
    :returns: formatted time string
    :rtype: str
    """
    days = "0"
    date_str = str(datetime.timedelta(seconds=delta))
    if ', ' in date_str:
        days, date_str = date_str.split(', ')
    hours, minutes, seconds = date_str.split(':')
    msg = "" if int(days.split(' ')[0]) == 0 else days + " "
    msg += "" if int(hours) == 0 else (hours + " hour{0} ".format('' if int(hours) <= 1 else 's'))
    msg += "" if int(minutes) == 0 else ("{0} minute{1} and ".format(int(minutes), '' if int(minutes) <= 1 else 's'))
    msg += "{0:.1f}".format(float(seconds))
    msg += " seconds"
    return msg def get_cached_recursive_build_depends_in_workspace(package, workspace_packages):
    """Returns cached or calculated recursive build depends for a given package

    If the recursive build depends for this package and this set of workspace
    packages has already been calculated, the cached results are returned.

    :param package: package for which the recursive depends should be calculated
    :type package: :py:class:`catkin_pkg.package.Package`
    :param workspace_packages: packages in the workspace, keyed by name, with
        value being a tuple of package path and package object
    :type workspace_packages: dict(package_name, tuple(package path,
        :py:class:`catkin_pkg.package.Package`))
    :returns: list of package path, package object tuples which are the
        recursive build depends for the given package
    :rtype: list(tuple(package path, :py:class:`catkin_pkg.package.Package`))
    """
    workspace_key = ','.join([pkg.name for pth, pkg in workspace_packages])
    if workspace_key not in __recursive_build_depends_cache:
        __recursive_build_depends_cache[workspace_key] = {}
    cache = __recursive_build_depends_cache[workspace_key]
    if package.name not in cache:
        cache[package.name] = get_recursive_build_depends_in_workspace(package, workspace_packages)
    __recursive_build_depends_cache[workspace_key] = cache
    return __recursive_build_depends_cache[workspace_key][package.name] def get_recursive_run_depends_in_workspace(packages, ordered_packages):
    """Calculates the recursive run depends of a set of packages which are also in the ordered_packages
    but excluding packages which are build depended on by another package in the list

    :param packages: packages for which the recursive depends should be calculated
    :type packages: list(:py:class:`catkin_pkg.package.Package`)
    :param ordered_packages: packages in the workspace, ordered topologically
    :type ordered_packages: list(tuple(package path,
        :py:class:`catkin_pkg.package.Package`))
    :returns: list of package path, package object tuples which are the
        recursive run depends for the given package
    :rtype: list(tuple(package path, :py:class:`catkin_pkg.package.Package`))
    """

    return get_recursive_depends_in_workspace(
        packages,
        ordered_packages,
        include_function=lambda p: p.run_depends,
        exclude_function=lambda p: p.buildtool_depends + p.build_depends
    ) def log(*args, **kwargs):
    """Wrapper for print, allowing for special handling where necessary"""
    global unicode_error_printed
    try:
        print(*args, **kwargs)
    except UnicodeEncodeError:
        # Strip unicode characters from string args
        sanitized_args = [unicode_sanitizer.sub('?', a)
                          if isinstance(a, str)
                          else a
                          for a in args]
        print(*sanitized_args, **kwargs)

        # Warn the user that
        if not unicode_error_printed:
            print('WARNING: Could not encode unicode characters. Please set the'
                  ' PYTHONIOENCODING environment variable to see complete output.'
                  ' (i.e. PYTHONIOENCODING=UTF-8)',
                  file=sys.stderr)
            unicode_error_printed = True def wide_log(msg, **kwargs):
    """Prints a message to the screen, filling the remainder of the screen with spaces

    This is useful for printing lines which will completely overwrite previous
    content printed with a carriage return at the end.

    If the message is wider than the terminal, then no filling is done.

    The wide logging can be disabled with :py:func:`disable_wide_log`, in order
    to prevent queries to the terminal width, which is useful when output is
    not to a terminal, like when being used with Continuous Integration.

    Truncating and right hand side messages are disabled when wide_log is
    disabled as well.

    When a right hand side message is given, it implies truncate is True.

    :param msg: message to be printed
    :type msg: str
    :param rhs: message to print at the right hand side of the screen
    :type rhs: str
    :param truncate: If True, messages wider the then terminal will be truncated
    :type truncate: bool
    """
    try:
        global wide_log_fn
        wide_log_fn(msg, **kwargs)
    except IOError:
        # This happens when someone ctrl-c's during a log message
        pass class ConsoleStatusController(threading.Thread):

    """Status thread for displaying events to the console.


    TODO: switch to interleaved output if only one job is running
    """

    def __init__(
            self,
            label,
            job_labels,
            jobs,
            max_toplevel_jobs,
            available_jobs,
            buildlisted_jobs,
            skiplisted_jobs,
            event_queue,
            show_notifications=False,
            show_stage_events=False,
            show_buffered_stdout=False,
            show_buffered_stderr=True,
            show_live_stdout=False,
            show_live_stderr=False,
            show_compact_io=False,
            show_active_status=True,
            show_summary=True,
            show_full_summary=False,
            show_repro_cmd=True,
            active_status_rate=10.0,
            pre_start_time=None):
        """
        :param label: The label for this task (build, clean, etc)
        :param job_labels: The labels to be used for the jobs (packages, tests, etc)
        :param event_queue: The event queue used by an Executor
        :param show_notifications: Show a libnotify notification when the jobs are finished
        :param show_stage_events: Show events relating to stages in each job
        :param show_buffered_stdout: Show stdout from jobs as they finish
        :param show_buffered_stderr: Show stderr from jobs as they finish
        :param show_live_stdout: Show stdout lines from jobs as they're generated
        :param show_live_stderr: Show stdout lines from jobs as they're generated
        :param show_compact_io: Don't print blank lines from redirected io
        :param show_active_status: Periodically show a status line displaying the active jobs
        :param show_summary: Show numbers of jobs that completed with errors and warnings
        :param show_full_summary: Show lists of jobs in each termination category
        :param show_repro_cmd: Show the commands to reproduce failed stages
        :param active_status_rate: The rate in Hz at which the status line should be printed
        :param pre_start_time: The actual start time to report, if preprocessing was done
        """
        super(ConsoleStatusController, self).__init__()

        self.label = label
        self.job_label = job_labels[0]
        self.jobs_label = job_labels[1]
        self.event_queue = event_queue
        self.max_toplevel_jobs = max_toplevel_jobs

        self.show_notifications = show_notifications
        self.show_stage_events = show_stage_events
        self.show_buffered_stdout = show_buffered_stdout
        self.show_buffered_stderr = show_buffered_stderr
        self.show_live_stdout = show_live_stdout
        self.show_live_stderr = show_live_stderr
        self.show_compact_io = show_compact_io
        self.show_active_status = show_active_status
        self.show_full_summary = show_full_summary
        self.show_summary = show_summary
        self.show_repro_cmd = show_repro_cmd
        self.active_status_rate = active_status_rate
        self.pre_start_time = pre_start_time

        self.keep_running = True

        # Map from jid -> job
        self.jobs = dict([(j.jid, j) for j in jobs])

        self.available_jobs = available_jobs
        self.skiplisted_jobs = skiplisted_jobs
        self.buildlisted_jobs = buildlisted_jobs

        # Compute the max job id length when combined with stage labels
        self.max_jid_length = 1
        if len(self.jobs) > 0:
            self.max_jid_length += max(
                [len(jid) + max([len(s.label) for s in job.stages] or [0])
                 for jid, job
                 in self.jobs.items()]
            )

    def print_exec_summary(self, completed_jobs, warned_jobs, failed_jobs):
        """
        Print verbose execution summary.
        """

        # Calculate the longest jid
        max_jid_len = max([len(jid) for jid in self.available_jobs])

        templates = {
            'successful': clr(" [@!@{gf}Successful@|] @{cf}{jid:<%d}@|" % max_jid_len),
            'warned': clr(" [    @!@{yf}Warned@|] @{cf}{jid:<%d}@|" % max_jid_len),
            'failed': clr(" [    @!@{rf}Failed@|] @{cf}{jid:<%d}@|" % max_jid_len),
            'ignored': clr(" [   @!@{kf}Ignored@|] @{cf}{jid:<%d}@|" % max_jid_len),
            'abandoned': clr(" [ @!@{rf}Abandoned@|] @{cf}{jid:<%d}@|" % max_jid_len),
        }

        # Calculate the maximum _printed_ length for each template
        max_column_len = max([
            len(remove_ansi_escape(t.format(jid=("?" * max_jid_len))))
            for t in templates.values()
        ])

        # Calculate the number of columns
        number_of_columns = int((terminal_width() / max_column_len) or 1)

        # Construct different categories of jobs (jid -> output template)
        successfuls = {}
        warneds = {}
        faileds = {}
        ignoreds = {}
        abandoneds = {}
        non_buildlisted = {}
        skiplisted = {}

        # Give each package an output template to use
        for jid in self.available_jobs:
            if jid in self.skiplisted_jobs:
                skiplisted[jid] = templates['ignored']
            elif jid not in self.jobs:
                ignoreds[jid] = templates['ignored']
            elif len(self.buildlisted_jobs) > 0 and jid not in self.buildlisted_jobs:
                non_buildlisted[jid] = templates['ignored']
            elif jid in completed_jobs:
                if jid in failed_jobs:
                    faileds[jid] = templates['failed']
                elif jid in warned_jobs:
                    warneds[jid] = templates['warned']
                else:
                    successfuls[jid] = templates['successful']
            else:
                abandoneds[jid] = templates['abandoned']

        # Combine successfuls and ignoreds, sort by key
        if len(successfuls) + len(ignoreds) > 0:
            wide_log("")
            wide_log(clr("[{}] Successful {}:").format(self.label, self.jobs_label))
            wide_log("")
            print_items_in_columns(
                sorted(itertools.chain(successfuls.items(), ignoreds.items())),
                number_of_columns)
        else:
            wide_log("")
            wide_log(clr("[{}] No {} succeeded.").format(self.label, self.jobs_label))
            wide_log("")

        # Print out buildlisted jobs
        if len(non_buildlisted) > 0:
            wide_log("")
            wide_log(clr("[{}] Non-buildlisted {}:").format(self.label, self.jobs_label))
            wide_log("")
            print_items_in_columns(sorted(non_buildlisted.items()), number_of_columns)

        # Print out skiplisted jobs
        if len(skiplisted) > 0:
            wide_log("")
            wide_log(clr("[{}] Skiplisted {}:").format(self.label, self.jobs_label))
            wide_log("")
            print_items_in_columns(sorted(skiplisted.items()), number_of_columns)

        # Print out jobs that failed
        if len(faileds) > 0:
            wide_log("")
            wide_log(clr("[{}] Failed {}:").format(self.label, self.jobs_label))
            wide_log("")
            print_items_in_columns(sorted(faileds.items()), number_of_columns)

        # Print out jobs that were abandoned
        if len(abandoneds) > 0:
            wide_log("")
            wide_log(clr("[{}] Abandoned {}:").format(self.label, self.jobs_label))
            wide_log("")
            print_items_in_columns(sorted(abandoneds.items()), number_of_columns)

        wide_log("")

    def print_compact_summary(self, completed_jobs, warned_jobs, failed_jobs):
        """Print a compact build summary."""

        notification_title = ""
        notification_msg = []
        # Print error summary
        if len(completed_jobs) == len(self.jobs) and all(completed_jobs.items()) and len(failed_jobs) == 0:
            notification_title = "{} Succeeded".format(self.label.capitalize())
            notification_msg.append("All {} {} succeeded!".format(len(self.jobs), self.jobs_label))

            wide_log(clr('[{}] Summary: All {} {} succeeded!').format(
                self.label,
                len(self.jobs),
                self.jobs_label))
        else:
            notification_msg.append("{} of {} {} succeeded.".format(
                len([succeeded for jid, succeeded in completed_jobs.items() if succeeded]),
                len(self.jobs), self.jobs_label))
            wide_log(clr('[{}] Summary: {} of {} {} succeeded.').format(
                self.label,
                len([succeeded for jid, succeeded in completed_jobs.items() if succeeded]),
                len(self.jobs),
                self.jobs_label))

        # Display number of ignored jobs (jobs which shouldn't have been built)
        all_ignored_jobs = [j for j in self.available_jobs if j not in self.jobs]
        if len(all_ignored_jobs) == 0:
            wide_log(clr('[{}] Ignored: None.').format(
                self.label))
        else:
            notification_msg.append("{} {} were skipped.".format(len(all_ignored_jobs), self.jobs_label))
            wide_log(clr('[{}] Ignored: {} {} were skipped or are skiplisted.').format(
                self.label,
                len(all_ignored_jobs),
                self.jobs_label))

        # Display number of jobs which produced warnings
        if len(warned_jobs) == 0:
            wide_log(clr('[{}] Warnings: None.').format(
                self.label))
        else:
            notification_title = "{} Succeeded with Warnings".format(self.label.capitalize())
            notification_msg.append("{} {} succeeded with warnings.".format(len(warned_jobs), self.jobs_label))

            wide_log(clr('[{}] Warnings: {} {} succeeded with warnings.').format(
                self.label,
                len(warned_jobs),
                self.jobs_label))

        # Display number of abandoned jobs
        all_abandoned_jobs = [j for j in self.jobs if j not in completed_jobs]
        if len(all_abandoned_jobs) == 0:
            wide_log(clr('[{}] Abandoned: No {} were abandoned.').format(
                self.label,
                self.jobs_label))
        else:
            notification_title = "{} Incomplete".format(self.label.capitalize())
            notification_msg.append("{} {} were abandoned.".format(len(all_abandoned_jobs), self.jobs_label))

            wide_log(clr('[{}] Abandoned: {} {} were abandoned.').format(
                self.label,
                len(all_abandoned_jobs),
                self.jobs_label))

        # Display number of failed jobs
        if len(failed_jobs) == 0:
            wide_log(clr('[{}] Failed: No {} failed.').format(
                self.label,
                self.jobs_label))
        else:
            notification_title = "{} Failed".format(self.label.capitalize())
            notification_msg.append("{} {} failed.".format(len(failed_jobs), self.jobs_label))

            wide_log(clr('[{}] Failed: {} {} failed.').format(
                self.label,
                len(failed_jobs),
                self.jobs_label))

        if self.show_notifications:
            if len(failed_jobs) != 0:
                notify(notification_title, "\n".join(notification_msg), icon_image='catkin_icon_red.png')
            elif len(warned_jobs) != 0:
                notify(notification_title, "\n".join(notification_msg), icon_image='catkin_icon_yellow.png')
            else:
                notify(notification_title, "\n".join(notification_msg))

    def run(self):
        queued_jobs = []
        active_jobs = []
        completed_jobs = {}
        failed_jobs = []
        warned_jobs = []

        cumulative_times = dict()
        start_times = dict()
        active_stages = dict()

        start_time = self.pre_start_time or time.time()
        last_update_time = time.time()

        # If the status rate is too low, just disable it
        if self.active_status_rate < 1E-3:
            self.show_active_status = False
        else:
            update_duration = 1.0 / self.active_status_rate

        # Disable the wide log padding if the status is disabled
        if not self.show_active_status:
            disable_wide_log()

        while True:
            # Check if we should stop
            if not self.keep_running:
                wide_log(clr('[{}] An internal error occurred!').format(self.label))
                return

            # Write a continuously-updated status line
            if self.show_active_status:

                # Try to get an event from the queue (non-blocking)
                try:
                    event = self.event_queue.get(False)
                except Empty:
                    # Determine if the status should be shown based on the desired
                    # status rate
                    elapsed_time = time.time() - last_update_time
                    show_status_now = elapsed_time > update_duration

                    if show_status_now:
                        # Print live status (overwrites last line)
                        status_line = clr('[{} {} s] [{}/{} complete] [{}/{} jobs] [{} queued]').format(
                            self.label,
                            format_time_delta_short(time.time() - start_time),
                            len(completed_jobs),
                            len(self.jobs),
                            job_server.running_jobs(),
                            job_server.max_jobs(),
                            len(queued_jobs) + len(active_jobs) - len(active_stages)
                        )

                        # Show failed jobs
                        if len(failed_jobs) > 0:
                            status_line += clr(' [@!@{rf}{}@| @{rf}failed@|]').format(len(failed_jobs))

                        # Check load / mem
                        if not job_server.load_ok():
                            status_line += clr(' [@!@{rf}High Load@|]')
                        if not job_server.mem_ok():
                            status_line += clr(' [@!@{rf}Low Memory@|]')

                        # Add active jobs
                        if len(active_jobs) == 0:
                            status_line += clr(' @/@!@{kf}Waiting for jobs...@|')
                        else:
                            active_labels = []

                            for j, (s, t, p) in active_stages.items():
                                d = format_time_delta_short(cumulative_times[j] + time.time() - t)
                                if p == '':
                                    active_labels.append(clr('[{}:{} - {}]').format(j, s, d))
                                else:
                                    active_labels.append(clr('[{}:{} ({}%) - {}]').format(j, s, p, d))

                            status_line += ' ' + ' '.join(active_labels)

                        # Print the status line
                        # wide_log(status_line)
                        wide_log(status_line, rhs='', end='\r')
                        sys.stdout.flush()

                        # Store this update time
                        last_update_time = time.time()
                    else:
                        time.sleep(max(0.0, min(update_duration - elapsed_time, 0.01)))

                    # Only continue when no event was received
                    continue
            else:
                # Try to get an event from the queue (blocking)
                try:
                    event = self.event_queue.get(True)
                except Empty:
                    break

            # A `None` event is a signal to terminate
            if event is None:
                break

            # Handle the received events
            eid = event.event_id

            if 'JOB_STATUS' == eid:
                queued_jobs = event.data['queued']
                active_jobs = event.data['active']
                completed_jobs = event.data['completed']

                # Check if all jobs have finished in some way
                if all([len(event.data[t]) == 0 for t in ['pending', 'queued', 'active']]):
                    break

            elif 'STARTED_JOB' == eid:
                cumulative_times[event.data['job_id']] = 0.0
                wide_log(clr('Starting >>> {:<{}}').format(
                    event.data['job_id'],
                    self.max_jid_length))

            elif 'FINISHED_JOB' == eid:
                duration = format_time_delta(cumulative_times[event.data['job_id']])

                if event.data['succeeded']:
                    wide_log(clr('Finished <<< {:<{}} [ {} ]').format(
                        event.data['job_id'],
                        self.max_jid_length,
                        duration))
                else:
                    failed_jobs.append(event.data['job_id'])
                    wide_log(clr('Failed <<< {:<{}} [ {} ]').format(
                        event.data['job_id'],
                        self.max_jid_length,
                        duration))

            elif 'ABANDONED_JOB' == eid:
                # Create a human-readable reason string
                if 'DEP_FAILED' == event.data['reason']:
                    direct = event.data['dep_job_id'] == event.data['direct_dep_job_id']
                    if direct:
                        reason = clr('Depends on failed job {}').format(event.data['dep_job_id'])
                    else:
                        reason = clr('Depends on failed job {} via {}').format(
                            event.data['dep_job_id'],
                            event.data['direct_dep_job_id'])
                elif 'PEER_FAILED' == event.data['reason']:
                    reason = clr('Unrelated job failed')
                elif 'MISSING_DEPS' == event.data['reason']:
                    reason = clr('Depends on unknown jobs: {}').format(
                        ', '.join([clr('@!{}@|').format(jid) for jid in event.data['dep_ids']]))

                wide_log(clr('Abandoned <<< {:<{}} [ {} ]').format(
                    event.data['job_id'],
                    self.max_jid_length,
                    reason))

            elif 'STARTED_STAGE' == eid:
                active_stages[event.data['job_id']] = [event.data['stage_label'], event.time, '']
                start_times[event.data['job_id']] = event.time

                if self.show_stage_events:
                    wide_log(clr('Starting >> {}:{}').format(
                        event.data['job_id'],
                        event.data['stage_label']))

            elif 'STAGE_PROGRESS' == eid:
                active_stages[event.data['job_id']][2] = event.data['percent']

            elif 'SUBPROCESS' == eid:
                if self.show_stage_events:
                    wide_log(clr('Subprocess > {}:{} `{}`').format(
                        event.data['job_id'],
                        event.data['stage_label'],
                        event.data['stage_repro']))

            elif 'FINISHED_STAGE' == eid:
                # Get the stage duration
                duration = event.time - start_times[event.data['job_id']]
                cumulative_times[event.data['job_id']] += duration

                # This is no longer the active stage for this job
                del active_stages[event.data['job_id']]

                header_border = None
                header_border_file = sys.stdout
                header_title = None
                header_title_file = sys.stdout
                lines = []
                footer_title = None
                footer_title_file = sys.stdout
                footer_border = None
                footer_border_file = sys.stdout

                # Generate headers / borders for output
                if event.data['succeeded']:
                    footer_title = clr(
                        'Finished << {}:{}').format(
                            event.data['job_id'],
                            event.data['stage_label'])

                    if len(event.data['stderr']) > 0:
                        # Mark that this job warned about something
                        if event.data['job_id'] not in warned_jobs:
                            warned_jobs.append(event.data['job_id'])

                        # Output contains warnings
                        header_border = clr('@!@{yf}' + '_' * (terminal_width() - 1) + '@|')
                        header_border_file = sys.stderr
                        header_title = clr(
                            'Warnings << {}:{} {}').format(
                                event.data['job_id'],
                                event.data['stage_label'],
                                event.data['logfile_filename'])
                        header_title_file = sys.stderr
                        footer_border = clr('@{yf}' + '.' * (terminal_width() - 1) + '@|')
                        footer_border_file = sys.stderr
                    else:
                        # Normal output, no warnings
                        header_title = clr(
                            'Output << {}:{} {}').format(
                                event.data['job_id'],
                                event.data['stage_label'],
                                event.data['logfile_filename'])

                    # Don't print footer title
                    if not self.show_stage_events:
                        footer_title = None
                else:
                    # Output contains errors
                    header_border = clr('@!@{rf}' + '_' * (terminal_width() - 1) + '@|')
                    header_border_file = sys.stderr
                    header_title = clr(
                        'Errors << {}:{} {}').format(
                            event.data['job_id'],
                            event.data['stage_label'],
                            event.data['logfile_filename'])
                    header_title_file = sys.stderr
                    footer_border = clr('@{rf}' + '.' * (terminal_width() - 1) + '@|')
                    footer_border_file = sys.stderr

                    footer_title = clr(
                        'Failed << {}:{:<{}} [ Exited with code {} ]').format(
                            event.data['job_id'],
                            event.data['stage_label'],
                            max(0, self.max_jid_length - len(event.data['job_id'])),
                            event.data['retcode'])
                    footer_title_file = sys.stderr

                lines_target = sys.stdout
                if self.show_buffered_stdout:
                    if len(event.data['interleaved']) > 0:
                        lines = [
                            line + '\n'
                            for line in event.data['interleaved'].splitlines()
                            if (self.show_compact_io is False or len(line.strip()) > 0)
                        ]
                    else:
                        header_border = None
                        header_title = None
                        footer_border = None
                elif self.show_buffered_stderr:
                    if len(event.data['stderr']) > 0:
                        lines = [
                            line + '\n'
                            for line in event.data['stderr'].splitlines()
                            if (self.show_compact_io is False or len(line.strip()) > 0)
                        ]
                        lines_target = sys.stderr
                    else:
                        header_border = None
                        header_title = None
                        footer_border = None

                if len(lines) > 0:
                    if self.show_repro_cmd:
                        if event.data['repro'] is not None:
                            lines.append(clr('@!@{kf}{}@|\n').format(event.data['repro']))

                    # Print the output
                    if header_border:
                        wide_log(header_border, file=header_border_file)
                    if header_title:
                        wide_log(header_title, file=header_title_file)
                    if len(lines) > 0:
                        wide_log(''.join(lines), end='\n', file=lines_target)
                    if footer_border:
                        wide_log(footer_border, file=footer_border_file)
                    if footer_title:
                        wide_log(footer_title, file=footer_title_file)

            elif 'STDERR' == eid:
                if self.show_live_stderr and len(event.data['data']) > 0:
                    wide_log(self.format_interleaved_lines(event.data), file=sys.stderr)

            elif 'STDOUT' == eid:
                if self.show_live_stdout and len(event.data['data']) > 0:
                    wide_log(self.format_interleaved_lines(event.data))

            elif 'MESSAGE' == eid:
                wide_log(event.data['msg'])

        # Print the full summary
        if self.show_full_summary:
            self.print_exec_summary(completed_jobs, warned_jobs, failed_jobs)

        # Print a compact summary
        if self.show_summary or self.show_full_summary:
            self.print_compact_summary(completed_jobs, warned_jobs, failed_jobs)

        # Print final runtime
        wide_log(clr('[{}] Runtime: {} total.').format(
            self.label,
            format_time_delta(time.time() - start_time)))

    def format_interleaved_lines(self, data):
        if self.max_toplevel_jobs != 1:
            prefix = clr('[{}:{}] ').format(
                data['job_id'],
                data['stage_label'])
        else:
            prefix = ''

        # This is used to clear the status bar that is printed in the current line
        clear_line = '\r{}\r'.format(' ' * terminal_width())
        suffix = clr('@|')
        lines = data['data'].splitlines()
        return clear_line + '\n'.join(prefix + line + suffix for line in lines) async def execute_jobs(
        verb,
        jobs,
        locks,
        event_queue,
        log_path,
        max_toplevel_jobs=None,
        continue_on_failure=False,
        continue_without_deps=False):
    """Process a number of jobs asynchronously.

    :param verb: Current command verb
    :param jobs: A list of topologically-sorted Jobs with no circular dependencies.
    :param locks: Dict containing the locks to acquire
    :param event_queue: A python queue for reporting events.
    :param log_path: The path in which logfiles can be written
    :param max_toplevel_jobs: Max number of top-level jobs
    :param continue_on_failure: Keep running jobs even if one fails.
    :param continue_without_deps: Run jobs even if their dependencies fail.
    """

    # Map of jid -> job
    job_map = dict([(j.jid, j) for j in jobs])
    # Jobs which are not ready to be executed
    pending_jobs = []
    # Jobs which are ready to be executed once workers are available
    queued_jobs = []
    # List of active jobs
    active_jobs = []
    # Set of active job futures
    active_job_fs = set()
    # Dict of completd jobs job_id -> succeeded
    completed_jobs = {}
    # List of jobs whose deps failed
    abandoned_jobs = []

    # Make sure job server has been initialized
    if not job_server.initialized():
        raise RuntimeError('JobServer has not been initialized.')

    # Create a thread pool executor for blocking python stages in the asynchronous jobs
    threadpool = ThreadPoolExecutor(max_workers=job_server.max_jobs())

    # Immediately abandon jobs with bad dependencies
    pending_jobs, new_abandoned_jobs = split(jobs, lambda j: all([d in job_map for d in j.deps]))

    for abandoned_job in new_abandoned_jobs:
        abandoned_jobs.append(abandoned_job)
        event_queue.put(ExecutionEvent(
            'ABANDONED_JOB',
            job_id=abandoned_job.jid,
            reason='MISSING_DEPS',
            dep_ids=[d for d in abandoned_job.deps if d not in job_map]))

    # Initialize list of ready and pending jobs (jobs not ready to be executed)
    queued_jobs, pending_jobs = split(pending_jobs, lambda j: len(j.deps) == 0)

    # Process all jobs asynchronously until there are none left
    while len(active_job_fs) + len(queued_jobs) + len(pending_jobs) > 0:

        # Activate jobs while the jobserver dispenses tokens
        while ((len(queued_jobs) > 0) and
               ((max_toplevel_jobs is None) or (len(active_jobs) < max_toplevel_jobs)) and
               (job_server.try_acquire() is not None)):

            # Pop a job off of the job queue
            job = queued_jobs.pop(0)

            # Label it (for debugging)
            job_server.add_label(job.jid)

            # Notify that the job is being started
            event_queue.put(ExecutionEvent(
                'STARTED_JOB',
                job_id=job.jid))

            # Start the job coroutine
            active_jobs.append(job)
            active_job_fs.add(async_job(verb, job, threadpool, locks, event_queue, log_path))

        # Report running jobs
        event_queue.put(ExecutionEvent(
            'JOB_STATUS',
            pending=[j.jid for j in pending_jobs],
            queued=[j.jid for j in queued_jobs],
            active=[j.jid for j in active_jobs],
            abandoned=[j.jid for j in abandoned_jobs],
            completed=completed_jobs
        ))

        # Process jobs as they complete asynchronously
        done_job_fs, active_job_fs = await asyncio.wait(
            active_job_fs,
            timeout=0.10,
            return_when=FIRST_COMPLETED)

        for done_job_f in done_job_fs:
            # Capture a result once the job has finished
            job_id, succeeded = await done_job_f

            # Release a jobserver token now that this job has succeeded
            job_server.release(job_id)
            active_jobs = [j for j in active_jobs if j.jid != job_id]

            # Generate event with the results of this job
            event_queue.put(ExecutionEvent(
                'FINISHED_JOB',
                job_id=job_id,
                succeeded=succeeded))

            # Add the job to the completed list
            completed_jobs[job_id] = succeeded

            # Handle failure modes
            if not succeeded:
                # Handle different abandoning policies
                if not continue_on_failure:
                    # Abort all pending jobs if any job fails
                    new_abandoned_jobs = queued_jobs + pending_jobs
                    queued_jobs = []
                    pending_jobs = []

                    # Notify that jobs have been abandoned
                    for abandoned_job in new_abandoned_jobs:
                        abandoned_jobs.append(abandoned_job)
                        event_queue.put(ExecutionEvent(
                            'ABANDONED_JOB',
                            job_id=abandoned_job.jid,
                            reason='PEER_FAILED',
                            peer_job_id=job_id))

                elif not continue_without_deps:
                    unhandled_abandoned_job_ids = [job_id]

                    # Abandon jobs which depend on abandoned jobs
                    while len(unhandled_abandoned_job_ids) > 0:
                        # Get the abandoned job
                        abandoned_job_id = unhandled_abandoned_job_ids.pop(0)

                        # Abandon all pending jobs which depend on this job_id
                        unhandled_abandoned_jobs, pending_jobs = split(
                            pending_jobs,
                            lambda j: abandoned_job_id in j.deps)

                        # Handle each new abandoned job
                        for abandoned_job in unhandled_abandoned_jobs:
                            abandoned_jobs.append(abandoned_job)
                            # Notify if any jobs have been abandoned
                            event_queue.put(ExecutionEvent(
                                'ABANDONED_JOB',
                                job_id=abandoned_job.jid,
                                reason='DEP_FAILED',
                                direct_dep_job_id=abandoned_job_id,
                                dep_job_id=job_id))

                        # Add additional job ids to check
                        unhandled_abandoned_job_ids.extend(
                            [j.jid for j in unhandled_abandoned_jobs])

            # Update the list of ready jobs (based on completed job dependencies)
            new_queued_jobs, pending_jobs = split(
                pending_jobs,
                lambda j: j.all_deps_completed(completed_jobs))
            new_queued_jobs.extend(queued_jobs)

            # queued jobs should preserve the original topological sort of jobs
            queued_jobs = [j for j in jobs if j in new_queued_jobs]

            # Notify of newly queued jobs
            for queued_job in new_queued_jobs:
                event_queue.put(ExecutionEvent(
                    'QUEUED_JOB',
                    job_id=queued_job.jid))

    # Report running jobs
    event_queue.put(ExecutionEvent(
        'JOB_STATUS',
        pending=[j.jid for j in pending_jobs],
        queued=[j.jid for j in queued_jobs],
        active=[j.jid for j in active_jobs],
        abandoned=[j.jid for j in abandoned_jobs],
        completed=completed_jobs
    ))

    return all(completed_jobs.values()) and len(abandoned_jobs) == 0 def run_until_complete(coroutine):
    # Get event loop
    loop = get_loop()
    loop.slow_callback_duration = 1.0

    # Run jobs
    return loop.run_until_complete(coroutine) def create_catkin_build_job(context, package, package_path, dependencies, force_cmake, pre_clean, prebuild=False):
    """Job class for building catkin packages"""

    # Package source space path
    pkg_dir = os.path.join(context.source_space_abs, package_path)

    # Package build space path
    build_space = context.package_build_space(package)
    # Package devel space path
    devel_space = context.package_devel_space(package)
    # Package install space path
    install_space = context.package_install_space(package)
    # Package metadata path
    metadata_path = context.package_metadata_path(package)
    # Environment dictionary for the job, which will be built
    # up by the executions in the loadenv stage.
    job_env = dict(os.environ)

    # Create job stages
    stages = []

    # Load environment for job.
    stages.append(FunctionStage(
        'loadenv',
        loadenv,
        locked_resource=None if context.isolate_install else 'installspace',
        job_env=job_env,
        package=package,
        context=context
    ))

    # Create package build space
    stages.append(FunctionStage(
        'mkdir',
        makedirs,
        path=build_space
    ))

    # Create package metadata dir
    stages.append(FunctionStage(
        'mkdir',
        makedirs,
        path=metadata_path
    ))

    # Copy source manifest
    stages.append(FunctionStage(
        'cache-manifest',
        copyfiles,
        source_paths=[os.path.join(context.source_space_abs, package_path, 'package.xml')],
        dest_path=os.path.join(metadata_path, 'package.xml')
    ))

    # Only run CMake if the Makefile doesn't exist or if --force-cmake is given
    # TODO: This would need to be different with `cmake --build`
    makefile_path = os.path.join(build_space, 'Makefile')

    if not os.path.isfile(makefile_path) or force_cmake:

        require_command('cmake', CMAKE_EXEC)

        # CMake command
        stages.append(CommandStage(
            'cmake',
            [
                CMAKE_EXEC,
                pkg_dir,
                '--no-warn-unused-cli',
                '-DCATKIN_DEVEL_PREFIX=' + devel_space,
                '-DCMAKE_INSTALL_PREFIX=' + install_space
            ] + context.cmake_args,
            cwd=build_space,
            logger_factory=CMakeIOBufferProtocol.factory_factory(pkg_dir),
            occupy_job=True
        ))
    else:
        # Check buildsystem command
        stages.append(CommandStage(
            'check',
            [MAKE_EXEC, 'cmake_check_build_system'],
            cwd=build_space,
            logger_factory=CMakeIOBufferProtocol.factory_factory(pkg_dir),
            occupy_job=True
        ))

    # Filter make arguments
    make_args = handle_make_arguments(
        context.make_args +
        context.catkin_make_args)

    # Pre-clean command
    if pre_clean:
        # TODO: Remove target args from `make_args`
        stages.append(CommandStage(
            'preclean',
            [MAKE_EXEC, 'clean'] + make_args,
            cwd=build_space,
        ))

    require_command('make', MAKE_EXEC)

    # Make command
    stages.append(CommandStage(
        'make',
        [MAKE_EXEC] + make_args,
        cwd=build_space,
        logger_factory=CMakeMakeIOBufferProtocol.factory
    ))

    # Symlink command if using a linked develspace
    if context.link_devel:
        stages.append(FunctionStage(
            'symlink',
            link_devel_products,
            locked_resource='symlink-collisions-file',
            package=package,
            package_path=package_path,
            devel_manifest_path=context.package_metadata_path(package),
            source_devel_path=context.package_devel_space(package),
            dest_devel_path=context.devel_space_abs,
            metadata_path=context.metadata_path(),
            prebuild=prebuild
        ))

    # Make install command, if installing
    if context.install:
        stages.append(CommandStage(
            'install',
            [MAKE_EXEC, 'install'],
            cwd=build_space,
            logger_factory=CMakeMakeIOBufferProtocol.factory,
            locked_resource=None if context.isolate_install else 'installspace'
        ))

    return Job(
        jid=package.name,
        deps=dependencies,
        env=job_env,
        stages=stages) def create_catkin_clean_job(
        context,
        package,
        package_path,
        dependencies,
        dry_run,
        clean_build,
        clean_devel,
        clean_install):
    """Generate a Job that cleans a catkin package"""

    stages = []

    # Package build space path
    build_space = context.package_build_space(package)
    # Package metadata path
    metadata_path = context.package_metadata_path(package)
    # Environment dictionary for the job, empty for a clean job
    job_env = {}

    # Remove installed files
    if clean_install:
        installed_files = get_installed_files(context.package_metadata_path(package))
        stages.append(FunctionStage(
            'cleaninstall',
            rmfiles,
            paths=sorted(installed_files),
            remove_empty=True,
            empty_root=context.install_space_abs,
            dry_run=dry_run))

    # Remove products in develspace
    if clean_devel:
        if context.merge_devel:
            # Remove build targets from devel space
            stages.append(CommandStage(
                'clean',
                [MAKE_EXEC, 'clean'],
                cwd=build_space,
            ))
        elif context.link_devel:
            # Remove symlinked products
            stages.append(FunctionStage(
                'unlink',
                unlink_devel_products,
                locked_resource='symlink-collisions-file',
                devel_space_abs=context.devel_space_abs,
                private_devel_path=context.package_private_devel_path(package),
                metadata_path=context.metadata_path(),
                package_metadata_path=context.package_metadata_path(package),
                dry_run=dry_run
            ))

            # Remove devel space
            stages.append(FunctionStage(
                'rmdevel',
                rmfiles,
                paths=[context.package_private_devel_path(package)],
                dry_run=dry_run))
        elif context.isolate_devel:
            # Remove devel space
            stages.append(FunctionStage(
                'rmdevel',
                rmfiles,
                paths=[context.package_devel_space(package)],
                dry_run=dry_run))

    # Remove build space
    if clean_build:
        stages.append(FunctionStage(
            'rmbuild',
            rmfiles,
            paths=[build_space],
            dry_run=dry_run))

    # Remove cached metadata
    if clean_build and clean_devel and clean_install:
        stages.append(FunctionStage(
            'rmmetadata',
            rmfiles,
            paths=[metadata_path],
            dry_run=dry_run))

    return Job(
        jid=package.name,
        deps=dependencies,
        env=job_env,
        stages=stages) def get_prebuild_package(build_space_abs, devel_space_abs, force):
    """This generates a minimal Catkin package used to generate Catkin
    environment setup files in a merged devel space.

    :param build_space_abs: The path to a merged build space
    :param devel_space_abs: The path to a merged devel space
    :param force: Overwrite files if they exist
    :returns: source directory path
    """

    # Get the path to the prebuild package
    prebuild_path = os.path.join(build_space_abs, 'catkin_tools_prebuild')
    if not os.path.exists(prebuild_path):
        mkdir_p(prebuild_path)

    # Create CMakeLists.txt file
    cmakelists_txt_path = os.path.join(prebuild_path, 'CMakeLists.txt')
    if force or not os.path.exists(cmakelists_txt_path):
        with open(cmakelists_txt_path, 'wb') as cmakelists_txt:
            cmakelists_txt.write(SETUP_PREBUILD_CMAKELISTS_TEMPLATE.encode('utf-8'))

    # Create package.xml file
    package_xml_path = os.path.join(prebuild_path, 'package.xml')
    if force or not os.path.exists(package_xml_path):
        with open(package_xml_path, 'wb') as package_xml:
            package_xml.write(SETUP_PREBUILD_PACKAGE_XML_TEMPLATE.encode('utf-8'))

    # Create CATKIN_IGNORE file because this package should not be found by catkin
    # This is only necessary when the build space is inside of the source space
    catkin_ignore_path = os.path.join(prebuild_path, 'CATKIN_IGNORE')
    open(catkin_ignore_path, 'wb').close()

    # Create the build directory for this package
    mkdir_p(os.path.join(build_space_abs, 'catkin_tools_prebuild'))

    return prebuild_path
import os
import pkg_resources
import sys
import time
import traceback
import yaml
import asyncio
from queue import Queue
    from catkin_pkg.package import parse_package
    from catkin_pkg.package import InvalidPackage
    from catkin_pkg.packages import find_packages
    from catkin_pkg.topological_order import topological_order_packages
from catkin_tools.common import FakeLock, expand_glob_package
from catkin_tools.common import format_time_delta
from catkin_tools.common import get_cached_recursive_build_depends_in_workspace
from catkin_tools.common import get_recursive_run_depends_in_workspace
from catkin_tools.common import log
from catkin_tools.common import wide_log
from catkin_tools.execution.controllers import ConsoleStatusController
from catkin_tools.execution.executor import execute_jobs
from catkin_tools.execution.executor import run_until_complete
from catkin_tools.jobs.catkin import create_catkin_build_job
from catkin_tools.jobs.catkin import create_catkin_clean_job
from catkin_tools.jobs.catkin import get_prebuild_package
from .color import clr

        if setup_util_present:
            # Setup util is already there, determine if it needs to be
            # regenerated
            if catkin_built:
                if catkin_present:
                    prebuild_pkg_path, prebuild_pkg = pkg_dict['catkin']
            elif prebuild_built:
                if catkin_present:
                    # TODO: Clean prebuild package
                    ct_prebuild_pkg_path = get_prebuild_package(
                        context.build_space_abs, context.devel_space_abs, force_cmake)
                    ct_prebuild_pkg = parse_package(ct_prebuild_pkg_path)

                    prebuild_jobs['caktin_tools_prebuild'] = create_catkin_clean_job(
                        context,
                        ct_prebuild_pkg,
                        ct_prebuild_pkg_path,
                        dependencies=[],
                        dry_run=False,
                        clean_build=True,
                        clean_devel=True,
                        clean_install=True)

                    # TODO: Build catkin package
                    prebuild_pkg_path, prebuild_pkg = pkg_dict['catkin']
                    prebuild_pkg_deps.append('catkin_tools_prebuild')
            else:
                # How did these get here??
                log("Warning: devel space setup files have an unknown origin.")
        else:
            # Setup util needs to be generated
            if catkin_built or prebuild_built:
                log("Warning: generated devel space setup files have been deleted.")

            if catkin_present:
                # Build catkin package
                prebuild_pkg_path, prebuild_pkg = pkg_dict['catkin']
            else:
                # Generate and buildexplicit prebuild package
                prebuild_pkg_path = get_prebuild_package(context.build_space_abs, context.devel_space_abs, force_cmake)
                prebuild_pkg = parse_package(prebuild_pkg_path)

        if prebuild_pkg is not None:
            # Create the prebuild job
            prebuild_job = create_catkin_build_job(
                context,
                prebuild_pkg,
                prebuild_pkg_path,
                dependencies=prebuild_pkg_deps,
                force_cmake=force_cmake,
                pre_clean=pre_clean,
                prebuild=True)

            # Add the prebuld job
            prebuild_jobs[prebuild_job.jid] = prebuild_job

    # Remove prebuild jobs from normal job list
    for prebuild_jid, prebuild_job in prebuild_jobs.items():
        if prebuild_jid in packages_to_be_built_names:
            packages_to_be_built_names.remove(prebuild_jid)

    # Initial jobs list is just the prebuild jobs
    jobs = [] + list(prebuild_jobs.values())

    # Get all build type plugins
    build_job_creators = {
        ep.name: ep.load()['create_build_job']
        for ep in pkg_resources.iter_entry_points(group='catkin_tools.jobs')
    }

    # It's a problem if there aren't any build types available
    if len(build_job_creators) == 0:
        sys.exit('Error: No build types available. Please check your catkin_tools installation.')

    # Construct jobs
    for pkg_path, pkg in all_packages:
        if pkg.name not in packages_to_be_built_names:
            continue

        # Get actual build deps
        deps = [
            p.name for _, p
            in get_cached_recursive_build_depends_in_workspace(pkg, packages_to_be_built)
            if p.name not in prebuild_jobs
        ]
        # All jobs depend on the prebuild jobs if they're defined
        if not no_deps:
            for j in prebuild_jobs.values():
                deps.append(j.jid)

        # Determine the job parameters
        build_job_kwargs = dict(
            context=context,
            package=pkg,
            package_path=pkg_path,
            dependencies=deps,
            force_cmake=force_cmake,
            pre_clean=pre_clean)

        # Create the job based on the build type
        build_type = pkg.get_build_type()

        if build_type in build_job_creators:
            jobs.append(build_job_creators[build_type](**build_job_kwargs))
        else:
            wide_log(clr(
                "[build] @!@{yf}Warning:@| Skipping package `{}` because it "
                "has an unsupported package build type: `{}`"
            ).format(pkg.name, build_type))

            wide_log(clr("[build] Note: Available build types:"))
            for bt_name in build_job_creators.keys():
                wide_log(clr("[build]  - `{}`".format(bt_name)))

    # Queue for communicating status
    event_queue = Queue()

    try:
        # Spin up status output thread

        status_thread = ConsoleStatusController(