xref: /aosp_15_r20/external/pigweed/pw_build/py/pw_build/python_runner.py (revision 61c4878ac05f98d0ceed94b57d316916de578985)

# Copyright 2022 The Pigweed Authors
#
# 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
#
#     https://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.
"""Script that preprocesses a Python command then runs it.

This script evaluates expressions in the Python command's arguments then invokes
the command.
"""

import argparse
import atexit
import json
import logging
import os
from pathlib import Path
import platform
import shlex
import subprocess
import sys
import time

try:
    from pw_build import gn_resolver
    from pw_build.python_package import load_packages
except (ImportError, ModuleNotFoundError):
    # Load from python_package from this directory if pw_build is not available.
    from python_package import load_packages  # type: ignore
    import gn_resolver  # type: ignore

if sys.platform != 'win32':
    import fcntl  # pylint: disable=import-error

    # TODO: b/227670947 - Support Windows.

_LOG = logging.getLogger(__name__)
_LOCK_ACQUISITION_TIMEOUT = 30 * 60  # 30 minutes in seconds

# TODO(frolv): Remove these aliases once downstream projects are migrated.
GnPaths = gn_resolver.GnPaths
expand_expressions = gn_resolver.expand_expressions


def _parse_args() -> argparse.Namespace:
    """Parses arguments for this script, splitting out the command to run."""

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '--gn-root',
        type=Path,
        required=True,
        help=(
            'Path to the root of the GN tree; '
            'value of rebase_path("//", root_build_dir)'
        ),
    )
    parser.add_argument(
        '--current-path',
        type=Path,
        required=True,
        help='Value of rebase_path(".", root_build_dir)',
    )
    parser.add_argument(
        '--default-toolchain', required=True, help='Value of default_toolchain'
    )
    parser.add_argument(
        '--current-toolchain', required=True, help='Value of current_toolchain'
    )
    parser.add_argument('--module', help='Run this module instead of a script')
    parser.add_argument(
        '--env',
        action='append',
        help='Environment variables to set as NAME=VALUE',
    )
    parser.add_argument(
        '--touch',
        type=Path,
        help='File to touch after the command is run',
    )
    parser.add_argument(
        '--capture-output',
        action='store_true',
        help='Capture subcommand output; display only on error',
    )
    parser.add_argument(
        '--working-directory',
        type=Path,
        help='Change to this working directory before running the subcommand',
    )
    parser.add_argument(
        '--python-dep-list-files',
        nargs='+',
        type=Path,
        help='Paths to text files containing lists of Python package metadata '
        'json files.',
    )
    parser.add_argument(
        '--python-virtualenv-config',
        type=Path,
        help='Path to a virtualenv json config to use for this action.',
    )
    parser.add_argument(
        '--command-launcher', help='Arguments to prepend to Python command'
    )
    parser.add_argument(
        'original_cmd',
        nargs=argparse.REMAINDER,
        help='Python script with arguments to run',
    )
    parser.add_argument(
        '--lockfile',
        type=Path,
        help=(
            'Path to a pip lockfile. Any pip execution will acquire an '
            'exclusive lock on it, any other module a shared lock.'
        ),
    )
    return parser.parse_args()


class LockAcquisitionTimeoutError(Exception):
    """Raised on a timeout."""


def acquire_lock(lockfile: Path, exclusive: bool):
    """Attempts to acquire the lock.

    Args:
      lockfile: pathlib.Path to the lock.
      exclusive: whether this needs to be an exclusive lock.

    Raises:
      LockAcquisitionTimeoutError: If the lock is not acquired after a
        reasonable time.
    """
    if sys.platform == 'win32':
        # No-op on Windows, which doesn't have POSIX file locking.
        # TODO: b/227670947 - Get this working on Windows, too.
        return

    start_time = time.monotonic()
    if exclusive:
        # pylint: disable-next=used-before-assignment
        lock_type = fcntl.LOCK_EX  # type: ignore[name-defined]
    else:
        # pylint: disable-next=used-before-assignment
        lock_type = fcntl.LOCK_SH  # type: ignore[name-defined]
    fd = os.open(lockfile, os.O_RDWR | os.O_CREAT)

    # Make sure we close the file when the process exits. If we manage to
    # acquire the lock below, closing the file will release it.
    def cleanup():
        os.close(fd)

    atexit.register(cleanup)

    backoff = 1
    while time.monotonic() - start_time < _LOCK_ACQUISITION_TIMEOUT:
        try:
            # pylint: disable=used-before-assignment
            fcntl.flock(  # type: ignore[name-defined]
                fd, lock_type | fcntl.LOCK_NB  # type: ignore[name-defined]
            )
            # pylint: enable=used-before-assignment
            return  # Lock acquired!
        except BlockingIOError:
            pass  # Keep waiting.

        time.sleep(backoff * 0.05)
        backoff += 1

    raise LockAcquisitionTimeoutError(
        f"Failed to acquire lock {lockfile} in {_LOCK_ACQUISITION_TIMEOUT}"
    )


class MissingPythonDependency(Exception):
    """An error occurred while processing a Python dependency."""


def _load_virtualenv_config(json_file_path: Path) -> tuple[str, str]:
    with json_file_path.open() as json_fp:
        json_dict = json.load(json_fp)
    return json_dict.get('interpreter'), json_dict.get('path')


def main(  # pylint: disable=too-many-arguments,too-many-branches,too-many-locals
    gn_root: Path,
    current_path: Path,
    original_cmd: list[str],
    default_toolchain: str,
    current_toolchain: str,
    module: str | None,
    env: list[str] | None,
    python_dep_list_files: list[Path],
    python_virtualenv_config: Path | None,
    capture_output: bool,
    touch: Path | None,
    working_directory: Path | None,
    command_launcher: str | None,
    lockfile: Path | None,
) -> int:
    """Script entry point."""

    python_paths_list = []
    if python_dep_list_files:
        py_packages = load_packages(
            python_dep_list_files,
            # If this python_action has no gn python_deps this file will be
            # empty.
            ignore_missing=True,
        )

        for pkg in py_packages:
            top_level_source_dir = pkg.package_dir
            if not top_level_source_dir:
                raise MissingPythonDependency(
                    'Unable to find top level source dir for the Python '
                    f'package "{pkg}"'
                )
            # Don't add this dir to the PYTHONPATH if no __init__.py exists.
            init_py_files = top_level_source_dir.parent.glob('*/__init__.py')
            if not any(init_py_files):
                continue
            python_paths_list.append(
                gn_resolver.abspath(top_level_source_dir.parent)
            )

        # Sort the PYTHONPATH list, it will be in a different order each build.
        python_paths_list = sorted(python_paths_list)

    if not original_cmd or original_cmd[0] != '--':
        _LOG.error('%s requires a command to run', sys.argv[0])
        return 1

    # GN build scripts are executed from the root build directory.
    root_build_dir = gn_resolver.abspath(Path.cwd())

    tool = current_toolchain if current_toolchain != default_toolchain else ''
    paths = gn_resolver.GnPaths(
        root=gn_resolver.abspath(gn_root),
        build=root_build_dir,
        cwd=gn_resolver.abspath(current_path),
        toolchain=tool,
    )

    command = [sys.executable]

    python_interpreter = None
    python_virtualenv = None
    if python_virtualenv_config:
        python_interpreter, python_virtualenv = _load_virtualenv_config(
            python_virtualenv_config
        )

    if python_interpreter is not None:
        command = [str(root_build_dir / python_interpreter)]

    if command_launcher is not None:
        command = shlex.split(command_launcher) + command

    if module is not None:
        command += ['-m', module]

    run_args: dict = dict()
    # Always inherit the environtment by default. If PYTHONPATH or VIRTUALENV is
    # set below then the environment vars must be copied in or subprocess.run
    # will run with only the new updated variables.
    run_args['env'] = os.environ.copy()

    if env is not None:
        environment = os.environ.copy()
        environment.update((k, v) for k, v in (a.split('=', 1) for a in env))
        run_args['env'] = environment

    script_command = original_cmd[0]
    if script_command == '--':
        script_command = original_cmd[1]

    is_pip_command = (
        module == 'pip' or 'pip_install_python_deps.py' in script_command
    )

    existing_env = run_args['env'] if 'env' in run_args else os.environ.copy()
    new_env = {}
    if python_virtualenv:
        new_env['VIRTUAL_ENV'] = str(root_build_dir / python_virtualenv)
        bin_folder = 'Scripts' if platform.system() == 'Windows' else 'bin'
        new_env['PATH'] = os.pathsep.join(
            [
                str(root_build_dir / python_virtualenv / bin_folder),
                existing_env.get('PATH', ''),
            ]
        )

    if python_virtualenv and python_paths_list and not is_pip_command:
        python_path_prepend = os.pathsep.join(
            str(p) for p in set(python_paths_list)
        )

        # Append the existing PYTHONPATH to the new one.
        new_python_path = os.pathsep.join(
            path_str
            for path_str in [
                python_path_prepend,
                existing_env.get('PYTHONPATH', ''),
            ]
            if path_str
        )

        new_env['PYTHONPATH'] = new_python_path

    if 'env' not in run_args:
        run_args['env'] = {}
    run_args['env'].update(new_env)

    if capture_output:
        # Combine stdout and stderr so that error messages are correctly
        # interleaved with the rest of the output.
        run_args['stdout'] = subprocess.PIPE
        run_args['stderr'] = subprocess.STDOUT

    # Build the command to run.
    try:
        for arg in original_cmd[1:]:
            command += gn_resolver.expand_expressions(paths, arg)
    except gn_resolver.ExpressionError as err:
        _LOG.error('%s: %s', sys.argv[0], err)
        return 1

    if working_directory:
        run_args['cwd'] = working_directory

    # TODO: b/235239674 - Deprecate the --lockfile option as part of the Python
    # GN template refactor.
    if lockfile:
        try:
            acquire_lock(lockfile, is_pip_command)
        except LockAcquisitionTimeoutError as exception:
            _LOG.error('%s', exception)
            return 1

    _LOG.debug('RUN %s', ' '.join(shlex.quote(arg) for arg in command))

    completed_process = subprocess.run(command, **run_args)

    if completed_process.returncode != 0:
        _LOG.debug(
            'Command failed; exit code: %d', completed_process.returncode
        )
        if capture_output:
            sys.stdout.buffer.write(completed_process.stdout)
    elif touch:
        # If a stamp file is provided and the command executed successfully,
        # touch the stamp file to indicate a successful run of the command.
        touch = touch.resolve()
        _LOG.debug('TOUCH %s', touch)

        # Create the parent directory in case GN / Ninja hasn't created it.
        touch.parent.mkdir(parents=True, exist_ok=True)
        touch.touch()

    return completed_process.returncode


if __name__ == '__main__':
    sys.exit(main(**vars(_parse_args())))