third_party/pigweed/repo/pw_build/py/pw_build/python_runner.py

   1 # Copyright 2020 The Pigweed Authors
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License"); you may not
   4 # use this file except in compliance with the License. You may obtain a copy of
   5 # the License at
   6 #
   7 #     https://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 # License for the specific language governing permissions and limitations under
  13 # the License.
  14 """Script that preprocesses a Python command then runs it.
  15
  16 This script evaluates expressions in the Python command's arguments then invokes
  17 the command.
  18 """
  19
  20 import argparse
  21 from dataclasses import dataclass
  22 import enum
  23 import logging
  24 import os
  25 from pathlib import Path
  26 import re
  27 import shlex
  28 import subprocess
  29 import sys
  30 from typing import Callable, Dict, Iterable, Iterator, List, NamedTuple
  31 from typing import Optional, Tuple
  32
  33 _LOG = logging.getLogger(__name__)
  34
  35
  36 def _parse_args() -> argparse.Namespace:
  37     """Parses arguments for this script, splitting out the command to run."""
  38
  39     parser = argparse.ArgumentParser(description=__doc__)
  40     parser.add_argument('--gn-root',
  41                         type=Path,
  42                         required=True,
  43                         help=('Path to the root of the GN tree; '
  44                               'value of rebase_path("//")'))
  45     parser.add_argument('--current-path',
  46                         type=Path,
  47                         required=True,
  48                         help='Value of rebase_path(".")')
  49     parser.add_argument('--default-toolchain',
  50                         required=True,
  51                         help='Value of default_toolchain')
  52     parser.add_argument('--current-toolchain',
  53                         required=True,
  54                         help='Value of current_toolchain')
  55     parser.add_argument('--directory',
  56                         type=Path,
  57                         help='Execute the command from this directory')
  58     parser.add_argument('--module', help='Run this module instead of a script')
  59     parser.add_argument('--env',
  60                         action='append',
  61                         help='Environment variables to set as NAME=VALUE')
  62     parser.add_argument(
  63         '--touch',
  64         type=Path,
  65         help='File to touch after the command is run',
  66     )
  67     parser.add_argument(
  68         '--capture-output',
  69         action='store_true',
  70         help='Capture subcommand output; display only on error',
  71     )
  72     parser.add_argument(
  73         'original_cmd',
  74         nargs=argparse.REMAINDER,
  75         help='Python script with arguments to run',
  76     )
  77     return parser.parse_args()
  78
  79
  80 class GnPaths(NamedTuple):
  81     """The set of paths needed to resolve GN paths to filesystem paths."""
  82     root: Path
  83     build: Path
  84     cwd: Path
  85
  86     # Toolchain label or '' if using the default toolchain
  87     toolchain: str
  88
  89     def resolve(self, gn_path: str) -> Path:
  90         """Resolves a GN path to a filesystem path."""
  91         if gn_path.startswith('//'):
  92             return self.root.joinpath(gn_path.lstrip('/')).resolve()
  93
  94         return self.cwd.joinpath(gn_path).resolve()
  95
  96     def resolve_paths(self, gn_paths: str, sep: str = ';') -> str:
  97         """Resolves GN paths to filesystem paths in a delimited string."""
  98         return sep.join(
  99             str(self.resolve(path)) for path in gn_paths.split(sep))
 100
 101
 102 @dataclass(frozen=True)
 103 class Label:
 104     """Represents a GN label."""
 105     name: str
 106     dir: Path
 107     relative_dir: Path
 108     toolchain: Optional['Label']
 109     out_dir: Path
 110     gen_dir: Path
 111
 112     def __init__(self, paths: GnPaths, label: str):
 113         # Use this lambda to set attributes on this frozen dataclass.
 114         set_attr = lambda attr, val: object.__setattr__(self, attr, val)
 115
 116         # Handle explicitly-specified toolchains
 117         if label.endswith(')'):
 118             label, toolchain = label[:-1].rsplit('(', 1)
 119         else:
 120             # Prevent infinite recursion for toolchains
 121             toolchain = paths.toolchain if paths.toolchain != label else ''
 122
 123         set_attr('toolchain', Label(paths, toolchain) if toolchain else None)
 124
 125         # Split off the :target, if provided, or use the last part of the path.
 126         try:
 127             directory, name = label.rsplit(':', 1)
 128         except ValueError:
 129             directory, name = label, label.rsplit('/', 1)[-1]
 130
 131         set_attr('name', name)
 132
 133         # Resolve the directory to an absolute path
 134         set_attr('dir', paths.resolve(directory))
 135         set_attr('relative_dir', self.dir.relative_to(paths.root.resolve()))
 136
 137         set_attr(
 138             'out_dir',
 139             paths.build / self.toolchain_name() / 'obj' / self.relative_dir)
 140         set_attr(
 141             'gen_dir',
 142             paths.build / self.toolchain_name() / 'gen' / self.relative_dir)
 143
 144     def gn_label(self) -> str:
 145         label = f'//{self.relative_dir.as_posix()}:{self.name}'
 146         return f'{label}({self.toolchain!r})' if self.toolchain else label
 147
 148     def toolchain_name(self) -> str:
 149         return self.toolchain.name if self.toolchain else ''
 150
 151     def __repr__(self) -> str:
 152         return self.gn_label()
 153
 154
 155 class _Artifact(NamedTuple):
 156     path: Path
 157     variables: Dict[str, str]
 158
 159
 160 # Matches a non-phony build statement.
 161 _GN_NINJA_BUILD_STATEMENT = re.compile(r'^build (.+):[ \n](?!phony\b)')
 162
 163
 164 def _parse_build_artifacts(build_dir: Path, fd) -> Iterator[_Artifact]:
 165     """Partially parses the build statements in a Ninja file."""
 166     lines = iter(fd)
 167
 168     def next_line():
 169         try:
 170             return next(lines)
 171         except StopIteration:
 172             return None
 173
 174     # Serves as the parse state (only two states)
 175     artifact: Optional[_Artifact] = None
 176
 177     line = next_line()
 178
 179     while line is not None:
 180         if artifact:
 181             if line.startswith('  '):  # build variable statements are indented
 182                 key, value = (a.strip() for a in line.split('=', 1))
 183                 artifact.variables[key] = value
 184                 line = next_line()
 185             else:
 186                 yield artifact
 187                 artifact = None
 188         else:
 189             match = _GN_NINJA_BUILD_STATEMENT.match(line)
 190             if match:
 191                 artifact = _Artifact(build_dir / match.group(1), {})
 192
 193             line = next_line()
 194
 195     if artifact:
 196         yield artifact
 197
 198
 199 def _search_target_ninja(ninja_file: Path, paths: GnPaths,
 200                          target: Label) -> Tuple[Optional[Path], List[Path]]:
 201     """Parses the main output file and object files from <target>.ninja."""
 202
 203     artifact: Optional[Path] = None
 204     objects: List[Path] = []
 205
 206     _LOG.debug('Parsing target Ninja file %s for %s', ninja_file, target)
 207
 208     with ninja_file.open() as fd:
 209         for path, variables in _parse_build_artifacts(paths.build, fd):
 210             # Older GN used .stamp files when there is no build artifact.
 211             if path.suffix == '.stamp':
 212                 continue
 213
 214             if variables:
 215                 assert not artifact, f'Multiple artifacts for {target}!'
 216                 artifact = path
 217             else:
 218                 objects.append(path)
 219
 220     return artifact, objects
 221
 222
 223 def _search_toolchain_ninja(ninja_file: Path, paths: GnPaths,
 224                             target: Label) -> Optional[Path]:
 225     """Searches the toolchain.ninja file for outputs from the provided target.
 226
 227     Files created by an action appear in toolchain.ninja instead of in their own
 228     <target>.ninja. If the specified target has a single output file in
 229     toolchain.ninja, this function returns its path.
 230     """
 231
 232     _LOG.debug('Searching toolchain Ninja file %s for %s', ninja_file, target)
 233
 234     # Older versions of GN used a .stamp file to signal completion of a target.
 235     stamp_dir = target.out_dir.relative_to(paths.build).as_posix()
 236     stamp_tool = f'{target.toolchain_name()}_stamp'
 237     stamp_statement = f'build {stamp_dir}/{target.name}.stamp: {stamp_tool} '
 238
 239     # Newer GN uses a phony Ninja target to signal completion of a target.
 240     phony_dir = Path(target.toolchain_name(), 'phony',
 241                      target.relative_dir).as_posix()
 242     phony_statement = f'build {phony_dir}/{target.name}: phony '
 243
 244     with ninja_file.open() as fd:
 245         for line in fd:
 246             for statement in (phony_statement, stamp_statement):
 247                 if line.startswith(statement):
 248                     output_files = line[len(statement):].strip().split()
 249                     if len(output_files) == 1:
 250                         return paths.build / output_files[0]
 251
 252                     break
 253
 254     return None
 255
 256
 257 def _search_ninja_files(
 258         paths: GnPaths,
 259         target: Label) -> Tuple[bool, Optional[Path], List[Path]]:
 260     ninja_file = target.out_dir / f'{target.name}.ninja'
 261     if ninja_file.exists():
 262         return (True, *_search_target_ninja(ninja_file, paths, target))
 263
 264     ninja_file = paths.build / target.toolchain_name() / 'toolchain.ninja'
 265     if ninja_file.exists():
 266         return True, _search_toolchain_ninja(ninja_file, paths, target), []
 267
 268     return False, None, []
 269
 270
 271 @dataclass(frozen=True)
 272 class TargetInfo:
 273     """Provides information about a target parsed from a .ninja file."""
 274
 275     label: Label
 276     generated: bool  # True if the Ninja files for this target were generated.
 277     artifact: Optional[Path]
 278     object_files: Tuple[Path]
 279
 280     def __init__(self, paths: GnPaths, target: str):
 281         object.__setattr__(self, 'label', Label(paths, target))
 282
 283         generated, artifact, objects = _search_ninja_files(paths, self.label)
 284
 285         object.__setattr__(self, 'generated', generated)
 286         object.__setattr__(self, 'artifact', artifact)
 287         object.__setattr__(self, 'object_files', tuple(objects))
 288
 289     def __repr__(self) -> str:
 290         return repr(self.label)
 291
 292
 293 class ExpressionError(Exception):
 294     """An error occurred while parsing an expression."""
 295
 296
 297 class _ArgAction(enum.Enum):
 298     APPEND = 0
 299     OMIT = 1
 300     EMIT_NEW = 2
 301
 302
 303 class _Expression:
 304     def __init__(self, match: re.Match, ending: int):
 305         self._match = match
 306         self._ending = ending
 307
 308     @property
 309     def string(self):
 310         return self._match.string
 311
 312     @property
 313     def end(self) -> int:
 314         return self._ending + len(_ENDING)
 315
 316     def contents(self) -> str:
 317         return self.string[self._match.end():self._ending]
 318
 319     def expression(self) -> str:
 320         return self.string[self._match.start():self.end]
 321
 322
 323 _Actions = Iterator[Tuple[_ArgAction, str]]
 324
 325
 326 def _target_file(paths: GnPaths, expr: _Expression) -> _Actions:
 327     target = TargetInfo(paths, expr.contents())
 328
 329     if not target.generated:
 330         raise ExpressionError(f'Target {target} has not been generated by GN!')
 331
 332     if target.artifact is None:
 333         raise ExpressionError(f'Target {target} has no output file!')
 334
 335     yield _ArgAction.APPEND, str(target.artifact)
 336
 337
 338 def _target_file_if_exists(paths: GnPaths, expr: _Expression) -> _Actions:
 339     target = TargetInfo(paths, expr.contents())
 340
 341     if target.generated:
 342         if target.artifact is None:
 343             raise ExpressionError(f'Target {target} has no output file!')
 344
 345         if Path(target.artifact).exists():
 346             yield _ArgAction.APPEND, str(target.artifact)
 347             return
 348
 349     yield _ArgAction.OMIT, ''
 350
 351
 352 def _target_objects(paths: GnPaths, expr: _Expression) -> _Actions:
 353     if expr.expression() != expr.string:
 354         raise ExpressionError(
 355             f'The expression "{expr.expression()}" in "{expr.string}" may '
 356             'expand to multiple arguments, so it cannot be used alongside '
 357             'other text or expressions')
 358
 359     target = TargetInfo(paths, expr.contents())
 360     if not target.generated:
 361         raise ExpressionError(f'Target {target} has not been generated by GN!')
 362
 363     for obj in target.object_files:
 364         yield _ArgAction.EMIT_NEW, str(obj)
 365
 366
 367 _FUNCTIONS: Dict['str', Callable[[GnPaths, _Expression], _Actions]] = {
 368     'TARGET_FILE': _target_file,
 369     'TARGET_FILE_IF_EXISTS': _target_file_if_exists,
 370     'TARGET_OBJECTS': _target_objects,
 371 }
 372
 373 _START_EXPRESSION = re.compile(fr'<({"|".join(_FUNCTIONS)})\(')
 374 _ENDING = ')>'
 375
 376
 377 def _expand_arguments(paths: GnPaths, string: str) -> _Actions:
 378     pos = 0
 379
 380     for match in _START_EXPRESSION.finditer(string):
 381         if pos != match.start():
 382             yield _ArgAction.APPEND, string[pos:match.start()]
 383
 384         ending = string.find(_ENDING, match.end())
 385         if ending == -1:
 386             raise ExpressionError(f'Parse error: no terminating "{_ENDING}" '
 387                                   f'was found for "{string[match.start():]}"')
 388
 389         expression = _Expression(match, ending)
 390         yield from _FUNCTIONS[match.group(1)](paths, expression)
 391
 392         pos = expression.end
 393
 394     if pos < len(string):
 395         yield _ArgAction.APPEND, string[pos:]
 396
 397
 398 def expand_expressions(paths: GnPaths, arg: str) -> Iterable[str]:
 399     """Expands <FUNCTION(...)> expressions; yields zero or more arguments."""
 400     if arg == '':
 401         return ['']
 402
 403     expanded_args: List[List[str]] = [[]]
 404
 405     for action, piece in _expand_arguments(paths, arg):
 406         if action is _ArgAction.OMIT:
 407             return []
 408
 409         expanded_args[-1].append(piece)
 410         if action is _ArgAction.EMIT_NEW:
 411             expanded_args.append([])
 412
 413     return (''.join(arg) for arg in expanded_args if arg)
 414
 415
 416 def main(
 417     gn_root: Path,
 418     current_path: Path,
 419     directory: Optional[Path],
 420     original_cmd: List[str],
 421     default_toolchain: str,
 422     current_toolchain: str,
 423     module: Optional[str],
 424     env: Optional[List[str]],
 425     capture_output: bool,
 426     touch: Optional[Path],
 427 ) -> int:
 428     """Script entry point."""
 429
 430     if not original_cmd or original_cmd[0] != '--':
 431         _LOG.error('%s requires a command to run', sys.argv[0])
 432         return 1
 433
 434     # GN build scripts are executed from the root build directory.
 435     root_build_dir = Path.cwd().resolve()
 436
 437     tool = current_toolchain if current_toolchain != default_toolchain else ''
 438     paths = GnPaths(root=gn_root.resolve(),
 439                     build=root_build_dir,
 440                     cwd=current_path.resolve(),
 441                     toolchain=tool)
 442
 443     command = [sys.executable]
 444
 445     if module is not None:
 446         command += ['-m', module]
 447
 448     run_args: dict = dict(cwd=directory)
 449
 450     if env is not None:
 451         environment = os.environ.copy()
 452         environment.update((k, v) for k, v in (a.split('=', 1) for a in env))
 453         run_args['env'] = environment
 454
 455     if capture_output:
 456         # Combine stdout and stderr so that error messages are correctly
 457         # interleaved with the rest of the output.
 458         run_args['stdout'] = subprocess.PIPE
 459         run_args['stderr'] = subprocess.STDOUT
 460
 461     try:
 462         for arg in original_cmd[1:]:
 463             command += expand_expressions(paths, arg)
 464     except ExpressionError as err:
 465         _LOG.error('%s: %s', sys.argv[0], err)
 466         return 1
 467
 468     _LOG.debug('RUN %s', ' '.join(shlex.quote(arg) for arg in command))
 469
 470     completed_process = subprocess.run(command, **run_args)
 471
 472     if completed_process.returncode != 0:
 473         _LOG.debug('Command failed; exit code: %d',
 474                    completed_process.returncode)
 475         if capture_output:
 476             sys.stdout.buffer.write(completed_process.stdout)
 477     elif touch:
 478         # If a stamp file is provided and the command executed successfully,
 479         # touch the stamp file to indicate a successful run of the command.
 480         _LOG.debug('TOUCH %s', touch)
 481         touch.touch()
 482
 483     return completed_process.returncode
 484
 485
 486 if __name__ == '__main__':
 487     sys.exit(main(**vars(_parse_args())))