• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python
2# Copyright 2022 The Pigweed Authors
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may not
5# use this file except in compliance with the License. You may obtain a copy of
6# the License at
7#
8#     https://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations under
14# the License.
15"""Watch build config dataclasses."""
16
17from __future__ import annotations
18
19from dataclasses import dataclass, field
20import functools
21import logging
22from pathlib import Path
23import shlex
24from typing import Callable, Mapping, TYPE_CHECKING
25
26from prompt_toolkit.formatted_text import ANSI, StyleAndTextTuples
27from prompt_toolkit.formatted_text.base import OneStyleAndTextTuple
28
29from pw_presubmit.build import write_gn_args_file
30
31if TYPE_CHECKING:
32    from pw_build.project_builder import ProjectBuilder
33    from pw_build.project_builder_prefs import ProjectBuilderPrefs
34
35_LOG = logging.getLogger('pw_build.watch')
36
37
38class UnknownBuildSystem(Exception):
39    """Exception for requesting unsupported build systems."""
40
41
42class UnknownBuildDir(Exception):
43    """Exception for an unknown build dir before command running."""
44
45
46@dataclass
47class BuildCommand:
48    """Store details of a single build step.
49
50    Example usage:
51
52    .. code-block:: python
53
54        from pw_build.build_recipe import BuildCommand, BuildRecipe
55
56        def should_gen_gn(out: Path):
57            return not (out / 'build.ninja').is_file()
58
59        cmd1 = BuildCommand(build_dir='out',
60                            command=['gn', 'gen', '{build_dir}'],
61                            run_if=should_gen_gn)
62
63        cmd2 = BuildCommand(build_dir='out',
64                            build_system_command='ninja',
65                            build_system_extra_args=['-k', '0'],
66                            targets=['default']),
67
68    Args:
69        build_dir: Output directory for this build command. This can be omitted
70            if the BuildCommand is included in the steps of a BuildRecipe.
71        build_system_command: This command should end with ``ninja``, ``make``,
72            or ``bazel``.
73        build_system_extra_args: A list of extra arguments passed to the
74            build_system_command. If running ``bazel test`` include ``test`` as
75            an extra arg here.
76        targets: Optional list of targets to build in the build_dir.
77        command: List of strings to run as a command. These are passed to
78            subprocess.run(). Any instances of the ``'{build_dir}'`` string
79            literal will be replaced at run time with the out directory.
80        run_if: A callable function to run before executing this
81            BuildCommand. The callable takes one Path arg for the build_dir. If
82            the callable returns true this command is executed. All
83            BuildCommands are run by default.
84    """
85
86    build_dir: Path | None = None
87    build_system_command: str | None = None
88    build_system_extra_args: list[str] = field(default_factory=list)
89    targets: list[str] = field(default_factory=list)
90    command: list[str] = field(default_factory=list)
91    run_if: Callable[[Path], bool] = lambda _build_dir: True
92
93    def __post_init__(self) -> None:
94        # Copy self._expanded_args from the command list.
95        self._expanded_args: list[str] = []
96        if self.command:
97            self._expanded_args = self.command
98
99    def should_run(self) -> bool:
100        """Return True if this build command should be run."""
101        if self.build_dir:
102            return self.run_if(self.build_dir)
103        return True
104
105    def _get_starting_build_system_args(self) -> list[str]:
106        """Return flags that appear immediately after the build command."""
107        assert self.build_system_command
108        assert self.build_dir
109        return []
110
111    def _get_build_system_args(self) -> list[str]:
112        assert self.build_system_command
113        assert self.build_dir
114
115        # Both make and ninja use -C for a build directory.
116        if self.make_command() or self.ninja_command():
117            return ['-C', str(self.build_dir), *self.targets]
118
119        if self.bazel_command():
120            # Bazel doesn't use -C for the out directory. Instead we use
121            # --symlink_prefix to save some outputs to the desired
122            # location. This is the same pattern used by pw_presubmit.
123            bazel_args = ['--symlink_prefix', str(self.build_dir / 'bazel-')]
124            if self.bazel_clean_command():
125                # Targets are unrecognized args for bazel clean
126                return bazel_args
127            return bazel_args + [*self.targets]
128
129        raise UnknownBuildSystem(
130            f'\n\nUnknown build system command "{self.build_system_command}" '
131            f'for build directory "{self.build_dir}".\n'
132            'Supported commands: ninja, bazel, make'
133        )
134
135    def _resolve_expanded_args(self) -> list[str]:
136        """Replace instances of '{build_dir}' with the self.build_dir."""
137        resolved_args = []
138        for arg in self._expanded_args:
139            if arg == "{build_dir}":
140                if not self.build_dir:
141                    raise UnknownBuildDir(
142                        '\n\nUnknown "{build_dir}" value for command:\n'
143                        f'  {self._expanded_args}\n'
144                        f'In BuildCommand: {repr(self)}\n\n'
145                        'Check build_dir is set for the above BuildCommand'
146                        'or included as a step to a BuildRecipe.'
147                    )
148                resolved_args.append(str(self.build_dir.resolve()))
149            else:
150                resolved_args.append(arg)
151        return resolved_args
152
153    def make_command(self) -> bool:
154        return (
155            self.build_system_command is not None
156            and self.build_system_command.endswith('make')
157        )
158
159    def ninja_command(self) -> bool:
160        return (
161            self.build_system_command is not None
162            and self.build_system_command.endswith('ninja')
163        )
164
165    def bazel_command(self) -> bool:
166        return (
167            self.build_system_command is not None
168            and self.build_system_command.endswith('bazel')
169        )
170
171    def bazel_build_command(self) -> bool:
172        return self.bazel_command() and 'build' in self.build_system_extra_args
173
174    def bazel_test_command(self) -> bool:
175        return self.bazel_command() and 'test' in self.build_system_extra_args
176
177    def bazel_clean_command(self) -> bool:
178        return self.bazel_command() and 'clean' in self.build_system_extra_args
179
180    def get_args(
181        self,
182        additional_ninja_args: list[str] | None = None,
183        additional_bazel_args: list[str] | None = None,
184        additional_bazel_build_args: list[str] | None = None,
185    ) -> list[str]:
186        """Return all args required to launch this BuildCommand."""
187        # If this is a plain command step, return self._expanded_args as-is.
188        if not self.build_system_command:
189            return self._resolve_expanded_args()
190
191        # Assmemble user-defined extra args.
192        extra_args = []
193        extra_args.extend(self.build_system_extra_args)
194        if additional_ninja_args and self.ninja_command():
195            extra_args.extend(additional_ninja_args)
196
197        if additional_bazel_build_args and self.bazel_build_command():
198            extra_args.extend(additional_bazel_build_args)
199
200        if additional_bazel_args and self.bazel_command():
201            extra_args.extend(additional_bazel_args)
202
203        build_system_target_args = self._get_build_system_args()
204
205        # Construct the build system command args.
206        command = [
207            self.build_system_command,
208            *self._get_starting_build_system_args(),
209            *extra_args,
210            *build_system_target_args,
211        ]
212        return command
213
214    def __str__(self) -> str:
215        return ' '.join(shlex.quote(arg) for arg in self.get_args())
216
217
218@dataclass
219class BuildRecipeStatus:
220    """Stores the status of a build recipe."""
221
222    recipe: BuildRecipe
223    current_step: str = ''
224    percent: float = 0.0
225    error_count: int = 0
226    return_code: int | None = None
227    flag_done: bool = False
228    flag_started: bool = False
229    error_lines: dict[int, list[str]] = field(default_factory=dict)
230
231    def pending(self) -> bool:
232        return self.return_code is None
233
234    def failed(self) -> bool:
235        if self.return_code is not None:
236            return self.return_code != 0
237        return False
238
239    def append_failure_line(self, line: str) -> None:
240        lines = self.error_lines.get(self.error_count, [])
241        lines.append(line)
242        self.error_lines[self.error_count] = lines
243
244    def has_empty_ninja_errors(self) -> bool:
245        for error_lines in self.error_lines.values():
246            # NOTE: There will be at least 2 lines for each ninja failure:
247            # - A starting 'FAILED: target' line
248            # - An ending line with this format:
249            #   'ninja: error: ... cannot make progress due to previous errors'
250
251            # If the total error line count is very short, assume it's an empty
252            # ninja error.
253            if len(error_lines) <= 3:
254                # If there is a failure in the regen step, there will be 3 error
255                # lines: The above two and one more with the regen command.
256                return True
257            # Otherwise, if the line starts with FAILED: build.ninja the failure
258            # is likely in the regen step and there will be extra cmake or gn
259            # error text that was not captured.
260            for line in error_lines:
261                if line.startswith(
262                    '\033[31mFAILED: \033[0mbuild.ninja'
263                ) or line.startswith('FAILED: build.ninja'):
264                    return True
265        return False
266
267    def increment_error_count(self, count: int = 1) -> None:
268        self.error_count += count
269        if self.error_count not in self.error_lines:
270            self.error_lines[self.error_count] = []
271
272    def should_log_failures(self) -> bool:
273        return (
274            self.recipe.project_builder is not None
275            and self.recipe.project_builder.separate_build_file_logging
276            and (not self.recipe.project_builder.send_recipe_logs_to_root)
277        )
278
279    def log_last_failure(self) -> None:
280        """Log the last ninja error if available."""
281        if not self.should_log_failures():
282            return
283
284        logger = self.recipe.error_logger
285        if not logger:
286            return
287
288        _color = self.recipe.project_builder.color  # type: ignore
289
290        lines = self.error_lines.get(self.error_count, [])
291        _LOG.error('')
292        _LOG.error(' ╔════════════════════════════════════')
293        _LOG.error(
294            ' ║  START %s Failure #%d:',
295            _color.cyan(self.recipe.display_name),
296            self.error_count,
297        )
298
299        logger.error('')
300        for line in lines:
301            logger.error(line)
302        logger.error('')
303
304        _LOG.error(
305            ' ║  END %s Failure #%d',
306            _color.cyan(self.recipe.display_name),
307            self.error_count,
308        )
309        _LOG.error(" ╚════════════════════════════════════")
310        _LOG.error('')
311
312    def log_entire_recipe_logfile(self) -> None:
313        """Log the entire build logfile if no ninja errors available."""
314        if not self.should_log_failures():
315            return
316
317        recipe_logfile = self.recipe.logfile
318        if not recipe_logfile:
319            return
320
321        _color = self.recipe.project_builder.color  # type: ignore
322
323        logfile_path = str(recipe_logfile.resolve())
324
325        _LOG.error('')
326        _LOG.error(' ╔════════════════════════════════════')
327        _LOG.error(
328            ' ║  %s Failure; Entire log below:',
329            _color.cyan(self.recipe.display_name),
330        )
331        _LOG.error(' ║  %s %s', _color.yellow('START'), logfile_path)
332
333        logger = self.recipe.error_logger
334        if not logger:
335            return
336
337        logger.error('')
338        for line in recipe_logfile.read_text(
339            encoding='utf-8', errors='ignore'
340        ).splitlines():
341            logger.error(line)
342        logger.error('')
343
344        _LOG.error(' ║  %s %s', _color.yellow('END'), logfile_path)
345        _LOG.error(" ╚════════════════════════════════════")
346        _LOG.error('')
347
348    def status_slug(self, restarting: bool = False) -> OneStyleAndTextTuple:
349        status = ('', '')
350        if not self.recipe.enabled:
351            return ('fg:ansidarkgray', 'Disabled')
352
353        waiting = False
354        if self.done:
355            if self.passed():
356                status = ('fg:ansigreen', 'OK      ')
357            elif self.failed():
358                status = ('fg:ansired', 'FAIL    ')
359        elif self.started:
360            status = ('fg:ansiyellow', 'Building')
361        else:
362            waiting = True
363            status = ('default', 'Waiting ')
364
365        # Only show Aborting if the process is building (or has failures).
366        if restarting and not waiting and not self.passed():
367            status = ('fg:ansiyellow', 'Aborting')
368        return status
369
370    def current_step_formatted(self) -> StyleAndTextTuples:
371        formatted_text: StyleAndTextTuples = []
372        if self.passed():
373            return formatted_text
374
375        if self.current_step:
376            if '\x1b' in self.current_step:
377                formatted_text = ANSI(self.current_step).__pt_formatted_text__()
378            else:
379                formatted_text = [('', self.current_step)]
380
381        return formatted_text
382
383    @property
384    def done(self) -> bool:
385        return self.flag_done
386
387    @property
388    def started(self) -> bool:
389        return self.flag_started
390
391    def mark_done(self) -> None:
392        self.flag_done = True
393
394    def mark_started(self) -> None:
395        self.flag_started = True
396
397    def set_failed(self) -> None:
398        self.flag_done = True
399        self.return_code = -1
400
401    def set_passed(self) -> None:
402        self.flag_done = True
403        self.return_code = 0
404
405    def passed(self) -> bool:
406        if self.done and self.return_code is not None:
407            return self.return_code == 0
408        return False
409
410
411@dataclass
412class BuildRecipe:
413    """Dataclass to store a list of BuildCommands.
414
415    Example usage:
416
417    .. code-block:: python
418
419        from pw_build.build_recipe import BuildCommand, BuildRecipe
420
421        def should_gen_gn(out: Path) -> bool:
422            return not (out / 'build.ninja').is_file()
423
424        recipe = BuildRecipe(
425            build_dir='out',
426            title='Vanilla Ninja Build',
427            steps=[
428                BuildCommand(command=['gn', 'gen', '{build_dir}'],
429                             run_if=should_gen_gn),
430                BuildCommand(build_system_command='ninja',
431                             build_system_extra_args=['-k', '0'],
432                             targets=['default']),
433            ],
434        )
435
436    Args:
437        build_dir: Output directory for this BuildRecipe. On init this out dir
438            is set for all included steps.
439        steps: List of BuildCommands to run.
440        title: Custom title. The build_dir is used if this is ommited.
441        auto_create_build_dir: Auto create the build directory and all necessary
442            parent directories before running any build commands.
443    """
444
445    build_dir: Path
446    steps: list[BuildCommand] = field(default_factory=list)
447    title: str | None = None
448    enabled: bool = True
449    auto_create_build_dir: bool = True
450
451    def __hash__(self):
452        return hash((self.build_dir, self.title, len(self.steps)))
453
454    def __post_init__(self) -> None:
455        # Update all included steps to use this recipe's build_dir.
456        for step in self.steps:
457            if self.build_dir:
458                step.build_dir = self.build_dir
459
460        # Set logging variables
461        self._logger: logging.Logger | None = None
462        self.error_logger: logging.Logger | None = None
463        self._logfile: Path | None = None
464        self._status: BuildRecipeStatus = BuildRecipeStatus(self)
465        self.project_builder: ProjectBuilder | None = None
466
467    def toggle_enabled(self) -> None:
468        self.enabled = not self.enabled
469
470    def set_project_builder(self, project_builder) -> None:
471        self.project_builder = project_builder
472
473    def set_targets(self, new_targets: list[str]) -> None:
474        """Reset all build step targets."""
475        for step in self.steps:
476            step.targets = new_targets
477
478    def set_logger(self, logger: logging.Logger) -> None:
479        self._logger = logger
480
481    def set_error_logger(self, logger: logging.Logger) -> None:
482        self.error_logger = logger
483
484    def set_logfile(self, log_file: Path) -> None:
485        self._logfile = log_file
486
487    def reset_status(self) -> None:
488        self._status = BuildRecipeStatus(self)
489
490    @property
491    def status(self) -> BuildRecipeStatus:
492        return self._status
493
494    @property
495    def log(self) -> logging.Logger:
496        if self._logger:
497            return self._logger
498        return logging.getLogger()
499
500    @property
501    def logfile(self) -> Path | None:
502        return self._logfile
503
504    @property
505    def display_name(self) -> str:
506        if self.title:
507            return self.title
508        return str(self.build_dir)
509
510    def targets(self) -> list[str]:
511        return list(
512            set(target for step in self.steps for target in step.targets)
513        )
514
515    def __str__(self) -> str:
516        message = self.display_name
517        targets = self.targets()
518        if targets:
519            target_list = ' '.join(self.targets())
520            message = f'{message} -- {target_list}'
521        return message
522
523
524def create_build_recipes(prefs: ProjectBuilderPrefs) -> list[BuildRecipe]:
525    """Create a list of BuildRecipes from ProjectBuilderPrefs."""
526    build_recipes: list[BuildRecipe] = []
527
528    if prefs.run_commands:
529        for command_str in prefs.run_commands:
530            build_recipes.append(
531                BuildRecipe(
532                    build_dir=Path.cwd(),
533                    steps=[BuildCommand(command=shlex.split(command_str))],
534                    title=command_str,
535                )
536            )
537
538    for build_dir, targets in prefs.build_directories.items():
539        steps: list[BuildCommand] = []
540        build_path = Path(build_dir)
541        if not targets:
542            targets = []
543
544        for (
545            build_system_command,
546            build_system_extra_args,
547        ) in prefs.build_system_commands(build_dir):
548            steps.append(
549                BuildCommand(
550                    build_system_command=build_system_command,
551                    build_system_extra_args=build_system_extra_args,
552                    targets=targets,
553                )
554            )
555
556        build_recipes.append(
557            BuildRecipe(
558                build_dir=build_path,
559                steps=steps,
560            )
561        )
562
563    return build_recipes
564
565
566def should_gn_gen(out: Path) -> bool:
567    """Returns True if the gn gen command should be run.
568
569    Returns True if ``build.ninja`` or ``args.gn`` files are missing from the
570    build directory.
571    """
572    # gn gen only needs to run if build.ninja or args.gn files are missing.
573    expected_files = [
574        out / 'build.ninja',
575        out / 'args.gn',
576    ]
577    return any(not gen_file.is_file() for gen_file in expected_files)
578
579
580def should_gn_gen_with_args(
581    gn_arg_dict: Mapping[str, bool | str | list | tuple]
582) -> Callable:
583    """Returns a callable which writes an args.gn file prior to checks.
584
585    Args:
586      gn_arg_dict: Dictionary of key value pairs to use as gn args.
587
588    Returns:
589      Callable which takes a single Path argument and returns a bool
590      for True if the gn gen command should be run.
591
592    The returned function will:
593
594    1. Always re-write the ``args.gn`` file.
595    2. Return True if ``build.ninja`` or ``args.gn`` files are missing.
596    """
597
598    def _write_args_and_check(out: Path) -> bool:
599        # Always re-write the args.gn file.
600        write_gn_args_file(out / 'args.gn', **gn_arg_dict)
601
602        return should_gn_gen(out)
603
604    return _write_args_and_check
605
606
607def _should_regenerate_cmake(
608    cmake_generate_command: list[str], out: Path
609) -> bool:
610    """Save the full cmake command to a file.
611
612    Returns True if cmake files should be regenerated.
613    """
614    _should_regenerate = True
615    cmake_command = ' '.join(cmake_generate_command)
616    cmake_command_filepath = out / 'cmake_cfg_command.txt'
617    if (out / 'build.ninja').is_file() and cmake_command_filepath.is_file():
618        if cmake_command == cmake_command_filepath.read_text():
619            _should_regenerate = False
620
621    if _should_regenerate:
622        out.mkdir(parents=True, exist_ok=True)
623        cmake_command_filepath.write_text(cmake_command)
624
625    return _should_regenerate
626
627
628def should_regenerate_cmake(
629    cmake_generate_command: list[str],
630) -> Callable[[Path], bool]:
631    """Return a callable to determine if cmake should be regenerated.
632
633    Args:
634      cmake_generate_command: Full list of args to run cmake.
635
636    The returned function will return True signaling CMake should be re-run if:
637
638    1. The provided CMake command does not match an existing args in the
639       ``cmake_cfg_command.txt`` file in the build dir.
640    2. ``build.ninja`` is missing or ``cmake_cfg_command.txt`` is missing.
641
642    When the function is run it will create the build directory if needed and
643    write the cmake_generate_command args to the ``cmake_cfg_command.txt`` file.
644    """
645    return functools.partial(_should_regenerate_cmake, cmake_generate_command)
646