platformio-core/platformio/commands/run/command.py

# Copyright (c) 2014-present PlatformIO <contact@platformio.org>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from multiprocessing import cpu_count
from os import getcwd
from os.path import isfile, join
from time import time

import click
from tabulate import tabulate

from platformio import exception, fs, util
from platformio.commands.device import device_monitor as cmd_device_monitor
from platformio.commands.run.helpers import (clean_build_dir,
                                             handle_legacy_libdeps)
from platformio.commands.run.processor import EnvironmentProcessor
from platformio.commands.test.processor import CTX_META_TEST_IS_RUNNING
from platformio.project.config import ProjectConfig
from platformio.project.helpers import (find_project_dir_above,
                                        get_project_build_dir)

# pylint: disable=too-many-arguments,too-many-locals,too-many-branches

try:
    DEFAULT_JOB_NUMS = cpu_count()
except NotImplementedError:
    DEFAULT_JOB_NUMS = 1


@click.command("run", short_help="Process project environments")
@click.option("-e", "--environment", multiple=True)
@click.option("-t", "--target", multiple=True)
@click.option("--upload-port")
@click.option("-d",
              "--project-dir",
              default=getcwd,
              type=click.Path(exists=True,
                              file_okay=True,
                              dir_okay=True,
                              writable=True,
                              resolve_path=True))
@click.option("-c",
              "--project-conf",
              type=click.Path(exists=True,
                              file_okay=True,
                              dir_okay=False,
                              readable=True,
                              resolve_path=True))
@click.option("-j",
              "--jobs",
              type=int,
              default=DEFAULT_JOB_NUMS,
              help=("Allow N jobs at once. "
                    "Default is a number of CPUs in a system (N=%d)" %
                    DEFAULT_JOB_NUMS))
@click.option("-s", "--silent", is_flag=True)
@click.option("-v", "--verbose", is_flag=True)
@click.option("--disable-auto-clean", is_flag=True)
@click.pass_context
def cli(ctx, environment, target, upload_port, project_dir, project_conf, jobs,
        silent, verbose, disable_auto_clean):
    # find project directory on upper level
    if isfile(project_dir):
        project_dir = find_project_dir_above(project_dir)

    is_test_running = CTX_META_TEST_IS_RUNNING in ctx.meta

    with fs.cd(project_dir):
        config = ProjectConfig.get_instance(
            project_conf or join(project_dir, "platformio.ini"))
        config.validate(environment)

        # clean obsolete build dir
        if not disable_auto_clean:
            try:
                clean_build_dir(get_project_build_dir(), config)
            except:  # pylint: disable=bare-except
                click.secho(
                    "Can not remove temporary directory `%s`. Please remove "
                    "it manually to avoid build issues" %
                    get_project_build_dir(force=True),
                    fg="yellow")

        handle_legacy_libdeps(project_dir, config)

        default_envs = config.default_envs()
        results = []
        for env in config.envs():
            skipenv = any([
                environment and env not in environment, not environment
                and default_envs and env not in default_envs
            ])
            if skipenv:
                results.append({"env": env})
                continue

            # print empty line between multi environment project
            if not silent and any(
                    r.get("succeeded") is not None for r in results):
                click.echo()

            results.append(
                process_env(ctx, env, config, environment, target, upload_port,
                            silent, verbose, jobs, is_test_running))

        command_failed = any(r.get("succeeded") is False for r in results)

        if (not is_test_running and (command_failed or not silent)
                and len(results) > 1):
            print_processing_summary(results)

        if command_failed:
            raise exception.ReturnErrorCode(1)
        return True


def process_env(ctx, name, config, environments, targets, upload_port, silent,
                verbose, jobs, is_test_running):
    if not is_test_running and not silent:
        print_processing_header(name, config, verbose)

    ep = EnvironmentProcessor(ctx, name, config, targets, upload_port, silent,
                              verbose, jobs)
    result = {"env": name, "elapsed": time(), "succeeded": ep.process()}
    result['elapsed'] = time() - result['elapsed']

    # print footer on error or when is not unit testing
    if not is_test_running and (not silent or not result['succeeded']):
        print_processing_footer(result)

    if (result['succeeded'] and "monitor" in ep.get_build_targets()
            and "nobuild" not in ep.get_build_targets()):
        ctx.invoke(cmd_device_monitor,
                   environment=environments[0] if environments else None)

    return result


def print_processing_header(env, config, verbose=False):
    env_dump = []
    for k, v in config.items(env=env):
        if verbose or k in ("platform", "framework", "board"):
            env_dump.append("%s: %s" %
                            (k, ", ".join(v) if isinstance(v, list) else v))
    click.echo("Processing %s (%s)" %
               (click.style(env, fg="cyan", bold=True), "; ".join(env_dump)))
    terminal_width, _ = click.get_terminal_size()
    click.secho("-" * terminal_width, bold=True)


def print_processing_footer(result):
    is_failed = not result.get("succeeded")
    util.print_labeled_bar(
        "[%s] Took %.2f seconds" %
        ((click.style("FAILED", fg="red", bold=True) if is_failed else
          click.style("SUCCESS", fg="green", bold=True)), result['elapsed']),
        is_error=is_failed)


def print_processing_summary(results):
    tabular_data = []
    succeeded_nums = 0
    failed_nums = 0
    elapsed = 0

    for result in results:
        elapsed += result.get("elapsed", 0)
        if result.get("succeeded") is False:
            failed_nums += 1
            status_str = click.style("FAILED", fg="red")
        elif result.get("succeeded") is None:
            status_str = "IGNORED"
        else:
            succeeded_nums += 1
            status_str = click.style("SUCCESS", fg="green")

        tabular_data.append(
            (click.style(result['env'], fg="cyan"), status_str,
             util.humanize_elapsed_time(result.get("elapsed"))))

    click.echo()
    click.echo(tabulate(tabular_data,
                        headers=[
                            click.style(s, bold=True)
                            for s in ("Environment", "Status", "Time")
                        ]),
               err=failed_nums)

    util.print_labeled_bar(
        "%s%d succeeded in %s" %
        ("%d failed, " % failed_nums if failed_nums else "", succeeded_nums,
         util.humanize_elapsed_time(elapsed)),
        is_error=failed_nums,
        fg="red" if failed_nums else "green")