diff options
Diffstat (limited to 'pw_watch/py/pw_watch/watch.py')
| -rwxr-xr-x | pw_watch/py/pw_watch/watch.py | 324 |
1 files changed, 245 insertions, 79 deletions
diff --git a/pw_watch/py/pw_watch/watch.py b/pw_watch/py/pw_watch/watch.py index 59e384ac5..fe495efba 100755 --- a/pw_watch/py/pw_watch/watch.py +++ b/pw_watch/py/pw_watch/watch.py @@ -38,26 +38,38 @@ Usage examples: import argparse from dataclasses import dataclass import errno +from itertools import zip_longest import logging import os from pathlib import Path +import re import shlex import subprocess import sys import threading from threading import Thread -from typing import (Iterable, List, NamedTuple, NoReturn, Optional, Sequence, - Tuple) +from typing import ( + Iterable, + List, + NamedTuple, + NoReturn, + Optional, + Sequence, + Tuple, +) import httpwatcher # type: ignore from watchdog.events import FileSystemEventHandler # type: ignore[import] from watchdog.observers import Observer # type: ignore[import] -import pw_cli.log +from prompt_toolkit.formatted_text.base import OneStyleAndTextTuple +from prompt_toolkit.formatted_text import StyleAndTextTuples + import pw_cli.branding import pw_cli.color import pw_cli.env +import pw_cli.log import pw_cli.plugins import pw_console.python_logging @@ -65,7 +77,8 @@ from pw_watch.watch_app import WatchApp from pw_watch.debounce import DebouncedFunction, Debouncer _COLOR = pw_cli.color.colors() -_LOG = logging.getLogger(__package__) +_LOG = logging.getLogger('pw_watch') +_NINJA_LOG = logging.getLogger('pw_watch_ninja_output') _ERRNO_INOTIFY_LIMIT_REACHED = 28 # Suppress events under 'fsevents', generated by watchdog on every file @@ -97,6 +110,8 @@ _FAIL_MESSAGE = """ ░ ░ ░ ░ ░ """ +_FULLSCREEN_STATUS_COLUMN_WIDTH = 10 + # TODO(keir): Figure out a better strategy for exiting. The problem with the # watcher is that doing a "clean exit" is slow. However, by directly exiting, @@ -157,8 +172,10 @@ def git_ignored(file: Path) -> bool: class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): """Process filesystem events and launch builds if necessary.""" + # pylint: disable=too-many-instance-attributes + NINJA_BUILD_STEP = re.compile( + r'^\[(?P<step>[0-9]+)/(?P<total_steps>[0-9]+)\] (?P<action>.*)$') - # pylint: disable=R0902 # too many instance attributes def __init__( self, build_commands: Sequence[BuildCommand], @@ -167,18 +184,26 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): charset: WatchCharset = _ASCII_CHARSET, restart: bool = True, jobs: int = None, - watch_app: bool = False, + fullscreen: bool = False, + banners: bool = True, ): super().__init__() + self.banners = banners + self.status_message: Optional[OneStyleAndTextTuple] = None + self.result_message: Optional[StyleAndTextTuples] = None self.current_stdout = '' + self.current_build_step = '' + self.current_build_percent = 0.0 + self.current_build_errors = 0 self.patterns = patterns self.ignore_patterns = ignore_patterns self.build_commands = build_commands self.charset: WatchCharset = charset self.restart_on_changes = restart - self.watch_app_enabled = watch_app + self.fullscreen_enabled = fullscreen + self.watch_app: Optional[WatchApp] = None self._current_build: subprocess.Popen self._extra_ninja_args = [] if jobs is None else [f'-j{jobs}'] @@ -190,16 +215,16 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): self.matching_path: Optional[Path] = None self.builds_succeeded: List[bool] = [] - if not self.watch_app_enabled: + if not self.fullscreen_enabled: self.wait_for_keypress_thread = threading.Thread( None, self._wait_for_enter) self.wait_for_keypress_thread.start() def rebuild(self): - """ Manual rebuild command triggered from watch app.""" + """ Rebuild command triggered from watch app.""" self._current_build.terminate() self._current_build.wait() - self.debouncer.press('Manual build requested...') + self.debouncer.press('Manual build requested') def _wait_for_enter(self) -> NoReturn: try: @@ -245,8 +270,18 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): if self.matching_path is None: self.matching_path = matching_path - self.debouncer.press( - f'File change detected: {os.path.relpath(matching_path)}') + log_message = f'File change detected: {os.path.relpath(matching_path)}' + if self.restart_on_changes: + if self.fullscreen_enabled and self.watch_app: + self.watch_app.rebuild_on_filechange() + self.debouncer.press(f'{log_message} Triggering build...') + else: + _LOG.info('%s ; not rebuilding', log_message) + + def _clear_screen(self) -> None: + if not self.fullscreen_enabled: + print('\033c', end='') # TODO(pwbug/38): Not Windows compatible. + sys.stdout.flush() # Implementation of DebouncedFunction.run() # @@ -257,19 +292,24 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): """Run all the builds in serial and capture pass/fail for each.""" # Clear the screen and show a banner indicating the build is starting. - if not self.watch_app_enabled: - print('\033c', end='') # TODO(pwbug/38): Not Windows compatible. - # TODO: fix the banner function to add the color on each line - for line in pw_cli.branding.banner().splitlines(): - _LOG.info(line) - _LOG.info( - _COLOR.green( - ' Watching for changes. Ctrl-C to exit; enter to rebuild')) + self._clear_screen() + + if self.fullscreen_enabled: + self.create_result_message() + _LOG.info( + _COLOR.green( + 'Watching for changes. Ctrl-d to exit; enter to rebuild')) + else: + for line in pw_cli.branding.banner().splitlines(): + _LOG.info(line) + _LOG.info( + _COLOR.green( + ' Watching for changes. Ctrl-C to exit; enter to rebuild') + ) _LOG.info('') _LOG.info('Change detected: %s', self.matching_path) - if not self.watch_app_enabled: - print('\033c', end='') # TODO(pwbug/38): Not Windows compatible. - sys.stdout.flush() + + self._clear_screen() self.builds_succeeded = [] num_builds = len(self.build_commands) @@ -278,11 +318,12 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): env = os.environ.copy() # Force colors in Pigweed subcommands run through the watcher. env['PW_USE_COLOR'] = '1' + # Force Ninja to output ANSI colors + env['CLICOLOR_FORCE'] = '1' for i, cmd in enumerate(self.build_commands, 1): index = f'[{i}/{num_builds}]' self.builds_succeeded.append(self._run_build(index, cmd, env)) - if self.builds_succeeded[-1]: level = logging.INFO tag = '(OK)' @@ -291,6 +332,33 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): tag = '(FAIL)' _LOG.log(level, '%s Finished build: %s %s', index, cmd, tag) + self.create_result_message() + + def create_result_message(self): + if not self.fullscreen_enabled: + return + + self.result_message = [] + first_building_target_found = False + for (succeeded, command) in zip_longest(self.builds_succeeded, + self.build_commands): + if succeeded: + self.result_message.append( + ('class:theme-fg-green', + 'OK'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH))) + elif succeeded is None and not first_building_target_found: + first_building_target_found = True + self.result_message.append( + ('class:theme-fg-yellow', + 'Building'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH))) + elif first_building_target_found: + self.result_message.append( + ('', ''.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH))) + else: + self.result_message.append( + ('class:theme-fg-red', + 'Failed'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH))) + self.result_message.append(('', f' {command}\n')) def _run_build(self, index: str, cmd: BuildCommand, env: dict) -> bool: # Make sure there is a build.ninja file for Ninja to use. @@ -315,7 +383,11 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): def _execute_command(self, command: list, env: dict) -> bool: """Runs a command with a blank before/after for visual separation.""" - if self.watch_app_enabled: + self.current_build_errors = 0 + self.status_message = ( + 'class:theme-fg-yellow', + 'Building'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH)) + if self.fullscreen_enabled: return self._execute_command_watch_app(command, env) print() self._current_build = subprocess.Popen(command, env=env) @@ -325,6 +397,8 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): def _execute_command_watch_app(self, command: list, env: dict) -> bool: """Runs a command with and outputs the logs.""" + if not self.watch_app: + return False self.current_stdout = '' returncode = None with subprocess.Popen(command, @@ -333,15 +407,50 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): stderr=subprocess.STDOUT, errors='replace') as proc: self._current_build = proc - self.current_stdout += 'START\n' + + # Empty line at the start. + _NINJA_LOG.info('') while returncode is None: - if proc.stdout: - output = proc.stdout.readline() - if output: - self.current_stdout += output + if not proc.stdout: + continue + + output = proc.stdout.readline() + self.current_stdout += output + + line_match_result = self.NINJA_BUILD_STEP.match(output) + if line_match_result: + matches = line_match_result.groupdict() + self.current_build_step = line_match_result.group(0) + self.current_build_percent = float( + int(matches.get('step', 0)) / + int(matches.get('total_steps', 1))) + + elif output.startswith(WatchApp.NINJA_FAILURE_TEXT): + _NINJA_LOG.critical(output.strip()) + self.current_build_errors += 1 + + else: + # Mypy output mixes character encoding in its colored output + # due to it's use of the curses module retrieving the 'sgr0' + # (or exit_attribute_mode) capability from the host + # machine's terminfo database. + # + # This can result in this sequence ending up in STDOUT as + # b'\x1b(B\x1b[m'. (B tells terminals to interpret text as + # USASCII encoding but will appear in prompt_toolkit as a B + # character. + # + # The following replace calls will strip out those + # instances. + _NINJA_LOG.info( + output.replace('\x1b(B\x1b[m', + '').replace('\x1b[1m', '').strip()) + self.watch_app.redraw_ui() returncode = proc.poll() - self.current_stdout += 'CHECK DONE\n' + # Empty line at the end. + _NINJA_LOG.info('') + return returncode == 0 # Implementation of DebouncedFunction.cancel() @@ -357,43 +466,59 @@ class PigweedBuildWatcher(FileSystemEventHandler, DebouncedFunction): def on_complete(self, cancelled: bool = False) -> None: # First, use the standard logging facilities to report build status. if cancelled: + self.status_message = ( + '', 'Cancelled'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH)) _LOG.error('Finished; build was interrupted') elif all(self.builds_succeeded): + self.status_message = ( + 'class:theme-fg-green', + 'Succeeded'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH)) _LOG.info('Finished; all successful') else: + self.status_message = ( + 'class:theme-fg-red', + 'Failed'.rjust(_FULLSCREEN_STATUS_COLUMN_WIDTH)) _LOG.info('Finished; some builds failed') - # Then, show a more distinct colored banner. - if not cancelled: - # Write out build summary table so you can tell which builds passed - # and which builds failed. - _LOG.info('') - _LOG.info(' .------------------------------------') - _LOG.info(' |') - for (succeeded, cmd) in zip(self.builds_succeeded, - self.build_commands): - slug = (self.charset.slug_ok - if succeeded else self.charset.slug_fail) - _LOG.info(' | %s %s', slug, cmd) - _LOG.info(' |') - _LOG.info(" '------------------------------------") - else: - # Build was interrupted. - _LOG.info('') - _LOG.info(' .------------------------------------') - _LOG.info(' |') - _LOG.info(' | %s- interrupted', self.charset.slug_fail) - _LOG.info(' |') - _LOG.info(" '------------------------------------") - - # Show a large color banner so it is obvious what the overall result is. - if all(self.builds_succeeded) and not cancelled: - for line in _PASS_MESSAGE.splitlines(): - _LOG.info(_COLOR.green(line)) + # Show individual build results for fullscreen app + if self.fullscreen_enabled: + self.create_result_message() + # For non-fullscreen pw watch else: - for line in _FAIL_MESSAGE.splitlines(): - _LOG.info(_COLOR.red(line)) - + # Show a more distinct colored banner. + if not cancelled: + # Write out build summary table so you can tell which builds + # passed and which builds failed. + _LOG.info('') + _LOG.info(' .------------------------------------') + _LOG.info(' |') + for (succeeded, cmd) in zip(self.builds_succeeded, + self.build_commands): + slug = (self.charset.slug_ok + if succeeded else self.charset.slug_fail) + _LOG.info(' | %s %s', slug, cmd) + _LOG.info(' |') + _LOG.info(" '------------------------------------") + else: + # Build was interrupted. + _LOG.info('') + _LOG.info(' .------------------------------------') + _LOG.info(' |') + _LOG.info(' | %s- interrupted', self.charset.slug_fail) + _LOG.info(' |') + _LOG.info(" '------------------------------------") + + # Show a large color banner for the overall result. + if self.banners: + if all(self.builds_succeeded) and not cancelled: + for line in _PASS_MESSAGE.splitlines(): + _LOG.info(_COLOR.green(line)) + else: + for line in _FAIL_MESSAGE.splitlines(): + _LOG.info(_COLOR.red(line)) + + if self.watch_app: + self.watch_app.redraw_ui() self.matching_path = None # Implementation of DebouncedFunction.on_keyboard_interrupt() @@ -496,11 +621,18 @@ def add_parser_arguments(parser: argparse.ArgumentParser) -> None: '--jobs', type=int, help="Number of cores to use; defaults to Ninja's default") - parser.add_argument('--watch-app', - dest='watch_app', + parser.add_argument('-f', + '--fullscreen', action='store_true', default=False, - help='Start the watch app console.') + help='Use a fullscreen interface.') + parser.add_argument('--debug-logging', + action='store_true', + help='Enable debug logging.') + parser.add_argument('--no-banners', + dest='banners', + action='store_false', + help='Hide pass/fail banners.') def _exit(code: int) -> NoReturn: @@ -648,12 +780,21 @@ def get_common_excludes() -> List[Path]: return exclude_list -# pylint: disable=R0913, R0914 # too many arguments and local variables -def watch_setup(default_build_targets: List[str], build_directories: List[str], - patterns: str, ignore_patterns_string: str, - exclude_list: List[Path], restart: bool, jobs: Optional[int], - serve_docs: bool, serve_docs_port: int, serve_docs_path: Path, - watch_app: bool): +def watch_setup( + default_build_targets: List[str], + build_directories: List[str], + patterns: str, + ignore_patterns_string: str, + exclude_list: List[Path], + restart: bool, + jobs: Optional[int], + serve_docs: bool, + serve_docs_port: int, + serve_docs_path: Path, + fullscreen: bool, + banners: bool, + # pylint: disable=too-many-arguments +) -> Tuple[str, PigweedBuildWatcher, List[Path]]: """Watches files and runs Ninja commands when they change.""" _LOG.info('Starting Pigweed build watcher') @@ -666,6 +807,9 @@ def watch_setup(default_build_targets: List[str], build_directories: List[str], # Preset exclude list for pigweed directory. exclude_list += get_common_excludes() + # Add build directories to the exclude list. + exclude_list.extend( + Path(build_dir[0]).resolve() for build_dir in build_directories) build_commands = [ BuildCommand(Path(build_dir[0]), tuple(build_dir[1:])) @@ -729,12 +873,12 @@ def watch_setup(default_build_targets: List[str], build_directories: List[str], charset=charset, restart=restart, jobs=jobs, - watch_app=watch_app, + fullscreen=fullscreen, + banners=banners, ) return path_to_log, event_handler, exclude_list -# pylint: disable=R0914 # too many local variables def watch(path_to_log: Path, event_handler: PigweedBuildWatcher, exclude_list: List[Path]): """Watches files and runs Ninja commands when they change.""" @@ -786,10 +930,25 @@ def main() -> None: formatter_class=argparse.RawDescriptionHelpFormatter) add_parser_arguments(parser) args = parser.parse_args() - path_to_log, event_handler, exclude_list = watch_setup(**vars(args)) - if args.watch_app: - watch_logfile = pw_console.python_logging.create_temp_log_file() + path_to_log, event_handler, exclude_list = watch_setup( + default_build_targets=args.default_build_targets, + build_directories=args.build_directories, + patterns=args.patterns, + ignore_patterns_string=args.ignore_patterns_string, + exclude_list=args.exclude_list, + restart=args.restart, + jobs=args.jobs, + serve_docs=args.serve_docs, + serve_docs_port=args.serve_docs_port, + serve_docs_path=args.serve_docs_path, + fullscreen=args.fullscreen, + banners=args.banners, + ) + + if args.fullscreen: + watch_logfile = (pw_console.python_logging.create_temp_log_file( + prefix=__package__)) pw_cli.log.install( level=logging.DEBUG, use_color=True, @@ -803,12 +962,19 @@ def main() -> None: args=(path_to_log, event_handler, exclude_list), daemon=True) watch_thread.start() - watch_app = WatchApp(startup_args=vars(args), - event_handler=event_handler, - loggers=[_LOG]) + watch_app = WatchApp(event_handler=event_handler, + debug_logging=args.debug_logging, + log_file_name=watch_logfile) + + event_handler.watch_app = watch_app watch_app.run() else: - watch(path_to_log, event_handler, exclude_list) + pw_cli.log.install( + level=logging.DEBUG if args.debug_logging else logging.INFO, + use_color=True, + hide_timestamp=False, + ) + watch(Path(path_to_log), event_handler, exclude_list) if __name__ == '__main__': |