pw_console: LogContainer, rendering and scrolling New LogContainer class which: - Holds log lines for the parent LogPane - Tracks the currently selected line - Handles scrolling - Renders the number of log lines that will fit in the LogPane window No-Docs-Update-Reason: Log viewer functionality still WIP. Change-Id: Iba5c290daad44a49650cc537417e53e99d83a29f Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/49024 Pigweed-Auto-Submit: Anthony DiGirolamo <tonymd@google.com> Commit-Queue: Auto-Submit <auto-submit@pigweed.google.com.iam.gserviceaccount.com> Reviewed-by: Joe Ethier <jethier@google.com>
diff --git a/pw_console/py/BUILD.gn b/pw_console/py/BUILD.gn index 10865c2..b53b2a6 100644 --- a/pw_console/py/BUILD.gn +++ b/pw_console/py/BUILD.gn
@@ -25,6 +25,7 @@ "pw_console/help_window.py", "pw_console/helpers.py", "pw_console/key_bindings.py", + "pw_console/log_container.py", "pw_console/log_pane.py", "pw_console/pw_ptpython_repl.py", "pw_console/repl_pane.py",
diff --git a/pw_console/py/pw_console/console_app.py b/pw_console/py/pw_console/console_app.py index d9444f5..fe2e0da 100644 --- a/pw_console/py/pw_console/console_app.py +++ b/pw_console/py/pw_console/console_app.py
@@ -188,10 +188,9 @@ mouse_support=True, ) - def add_log_handler(self, logger_instance: Iterable): + def add_log_handler(self, logger_instance: logging.Logger): """Add the Log pane as a handler for this logger instance.""" - # TODO: Add log pane to addHandler call. - # logger_instance.addHandler(self.log_pane.log_container) + logger_instance.addHandler(self.log_pane.log_container) def _user_code_thread_entry(self): """Entry point for the user code thread.""" @@ -281,7 +280,7 @@ def embed( global_vars=None, local_vars=None, - loggers: Optional[Iterable] = None, + loggers: Optional[Iterable[logging.Logger]] = None, test_mode=False, ) -> None: """Call this to embed pw console at the call point within your program.
diff --git a/pw_console/py/pw_console/helpers.py b/pw_console/py/pw_console/helpers.py index 484009b..87ecf00 100644 --- a/pw_console/py/pw_console/helpers.py +++ b/pw_console/py/pw_console/helpers.py
@@ -17,3 +17,26 @@ def remove_formatting(formatted_text): """Throw away style info from formatted text tuples.""" return ''.join([formatted_tuple[1] for formatted_tuple in formatted_text]) # pylint: disable=not-an-iterable + + +def get_line_height(text_width, screen_width, prefix_width): + """Calculates line height for a string with line wrapping enabled.""" + if text_width == 0: + return 0 + # If text will fit on the screen without wrapping. + if text_width < screen_width: + return 1 + + # Start with height of 1 row. + total_height = 1 + # One screen_width of characters has been displayed. + remaining_width = text_width - screen_width + + # While the remaining character count won't fit on the screen: + while (remaining_width + prefix_width) > screen_width: + remaining_width += prefix_width + remaining_width -= screen_width + total_height += 1 + + # Add one for the last line that is < screen_width + return total_height + 1
diff --git a/pw_console/py/pw_console/log_container.py b/pw_console/py/pw_console/log_container.py new file mode 100644 index 0000000..804d82c --- /dev/null +++ b/pw_console/py/pw_console/log_container.py
@@ -0,0 +1,445 @@ +# Copyright 2021 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. +"""LogLine and LogContainer.""" + +import logging +import re +import sys +import time + +from collections import deque +from dataclasses import dataclass +from datetime import datetime +from typing import List, Dict, Optional + +from prompt_toolkit.data_structures import Point +from prompt_toolkit.formatted_text import ( + ANSI, + to_formatted_text, + fragment_list_width, +) + +import pw_cli.color +from pw_console.helpers import get_line_height + +_LOG = logging.getLogger(__package__) + +_ANSI_SEQUENCE_REGEX = re.compile(r'\x1b[^m]*m') + + +@dataclass +class LogLine: + """Class to hold a single log event.""" + record: logging.LogRecord + formatted_log: str + + def time(self): + """Return a datetime object for the log record.""" + return datetime.fromtimestamp(self.record.created) + + def get_fragments(self) -> List: + """Return this log line as a list of FormattedText tuples.""" + # Manually make a FormattedText tuple, wrap in a list + # return [('class:bottom_toolbar_colored_text', self.record.msg)] + # Use ANSI, returns a list of FormattedText tuples. + + # fragments = [('[SetCursorPosition]', '')] + # fragments += ANSI(self.formatted_log).__pt_formatted_text__() + # return fragments + + # Add a trailing linebreak + return ANSI(self.formatted_log + '\n').__pt_formatted_text__() + + +def create_empty_log_message(): + """Create an empty LogLine instance.""" + return LogLine(record=logging.makeLogRecord({}), formatted_log='') + + +class LogContainer(logging.Handler): + """Class to hold many log events.""" + + # pylint: disable=too-many-instance-attributes + def __init__(self): + # Log storage deque for fast addition and deletion from the beginning + # and end of the iterable. + self.logs: deque = deque() + # Estimate of the logs in memory. + self.byte_size: int = 0 + + # Only allow this many log lines in memory. + self.max_history_size: int = 1000000 + + # Counts of logs per python logger name + self.channel_counts: Dict[str, int] = {} + # Widths of each logger prefix string. For example: the character length + # of the timestamp string. + self.channel_formatted_prefix_widths: Dict[str, int] = {} + # Longest of the above prefix widths. + self.longest_channel_prefix_width = 0 + + # Current log line index state variables: + self.line_index = 0 + self._last_start_index = 0 + self._last_end_index = 0 + self._current_start_index = 0 + self._current_end_index = 0 + + # LogPane prompt_toolkit container render size. + self._window_height = 20 + self._window_width = 80 + + # Max frequency in seconds of prompt_toolkit UI redraws triggered by new + # log lines. + self._ui_update_frequency = 0.3 + self._last_ui_update_time = time.time() + + # Erase existing logs. + self.clear_logs() + # New logs have arrived flag. + self.has_new_logs = True + + # Should new log lines be tailed? + self.follow = True + + # Parent LogPane reference. Updated by calling `set_log_pane()`. + self.log_pane = None + + # Cache of formatted text tuples used in the last UI render. Used after + # rendering by `get_cursor_position()`. + self._line_fragment_cache = None + + super().__init__() + + def set_log_pane(self, log_pane): + """Set the parent LogPane instance.""" + self.log_pane = log_pane + + def set_formatting(self): + """Setup log formatting.""" + # Copy of pw_cli log formatter + colors = pw_cli.color.colors(True) + timestamp_fmt = colors.black_on_white('%(asctime)s') + ' ' + formatter = logging.Formatter( + timestamp_fmt + '%(levelname)s %(message)s', '%Y%m%d %H:%M:%S') + + self.setLevel(logging.DEBUG) + self.setFormatter(formatter) + + def get_current_line(self): + """Return the currently selected log event index.""" + return self.line_index + + def clear_logs(self): + """Erase all stored pane lines.""" + self.logs = deque() + self.byte_size = 0 + self.channel_counts = {} + self.channel_formatted_prefix_widths = {} + self.line_index = 0 + + def wrap_lines_enabled(self): + """Get the parent log pane wrap lines setting.""" + if not self.log_pane: + return False + return self.log_pane.wrap_lines + + def toggle_follow(self): + """Toggle auto line following.""" + self.follow = not self.follow + if self.follow: + self.scroll_to_bottom() + + def get_total_count(self): + """Total size of the logs container.""" + return len(self.logs) + + def get_last_log_line_index(self): + """Last valid index of the logs.""" + # Subtract 1 since self.logs is zero indexed. + return len(self.logs) - 1 + + def get_channel_counts(self): + """Return the seen channel log counts for this conatiner.""" + return ', '.join([ + f'{name}: {count}' for name, count in self.channel_counts.items() + ]) + + def _update_log_prefix_width(self, record, formatted_log): + """Save the formatted prefix width if this is a new logger channel + name.""" + if record.name not in self.channel_formatted_prefix_widths.keys(): + # Delete ANSI escape sequences. + ansi_stripped_log = _ANSI_SEQUENCE_REGEX.sub('', formatted_log) + # Save the width of the formatted portion of the log message. + self.channel_formatted_prefix_widths[ + record.name] = len(ansi_stripped_log) - len(record.msg) + # Set the max width of all known formats so far. + self.longest_channel_prefix_width = max( + self.channel_formatted_prefix_widths.values()) + + def _append_log(self, record): + """Add a new log event.""" + # Format incoming log line. + formatted_log = self.format(record) + # Save this log. + self.logs.append(LogLine(record=record, formatted_log=formatted_log)) + # Increment this logger count + self.channel_counts[record.name] = self.channel_counts.get( + record.name, 0) + 1 + + # Save prefix width of this log line. + self._update_log_prefix_width(record, formatted_log) + + # Update estimated byte_size. + self.byte_size += sys.getsizeof(self.logs[-1]) + # If the total log lines is > max_history_size, delete the oldest line. + if self.get_total_count() > self.max_history_size: + self.byte_size -= sys.getsizeof(self.logs.popleft()) + + # Set the has_new_logs flag. + self.has_new_logs = True + # If follow is on, scroll to the last line. + if self.follow: + self.scroll_to_bottom() + + def _update_prompt_toolkit_ui(self): + """Update Prompt Toolkit UI if a certain amount of time has passed.""" + emit_time = time.time() + # Has enough time passed since last UI redraw? + if emit_time > self._last_ui_update_time + self._ui_update_frequency: + # Update last log time + self._last_ui_update_time = emit_time + + # Trigger Prompt Toolkit UI redraw. + console_app = self.log_pane.application + if hasattr(console_app, 'application'): + # Thread safe way of sending a repaint trigger to the input + # event loop. + console_app.application.invalidate() + + def log_content_changed(self): + """Return True if new log lines have appeared since the last render.""" + return self.has_new_logs + + def get_cursor_position(self) -> Optional[Point]: + """Return the position of the cursor.""" + # This implementation is based on get_cursor_position from + # prompt_toolkit's FormattedTextControl class. + + fragment = "[SetCursorPosition]" + # If no lines were rendered. + if not self._line_fragment_cache: + return Point(0, 0) + # For each line rendered in the last pass: + for row, line in enumerate(self._line_fragment_cache): + column = 0 + # For each style string and raw text tuple in this line: + for style_str, text, *_ in line: + # If [SetCursorPosition] is in the style set the cursor position + # to this row and column. + if fragment in style_str: + return Point(x=column, y=row) + column += len(text) + return Point(0, 0) + + def emit(self, record): + """Process a new log record. + + This defines the logging.Handler emit() fuction which is called by + logging.Handler.handle() We don't implement handle() as it is done in + the parent class with thread safety and filters applied. + """ + self._append_log(record) + self._update_prompt_toolkit_ui() + + def scroll_to_top(self): + """Move selected index to the beginning.""" + # Stop following so cursor doesn't jump back down to the bottom. + self.follow = False + self.line_index = 0 + + def scroll_to_bottom(self): + """Move selected index to the end.""" + # Don't change following state like scroll_to_top. + self.line_index = max(0, self.get_last_log_line_index()) + + def scroll(self, lines): + """Scroll up or down by plus or minus lines. + + This method is only called by user keybindings. + """ + # If the user starts scrolling, stop auto following. + self.follow = False + + # If scrolling to an index below zero, set to zero. + new_line_index = max(0, self.line_index + lines) + # If past the end, set to the last index of self.logs. + if new_line_index >= self.get_total_count(): + new_line_index = self.get_last_log_line_index() + # Set the new selected line index. + self.line_index = new_line_index + + def scroll_to_position(self, mouse_position: Point): + """Set the selected log line to the mouse_position.""" + # If auto following don't move the cursor arbitrarily. That would stop + # following and position the cursor incorrectly. + if self.follow: + return + + cursor_position = self.get_cursor_position() + if cursor_position: + scroll_amount = cursor_position.y - mouse_position.y + self.scroll(-1 * scroll_amount) + + def scroll_up_one_page(self): + """Move the selected log index up by one window height.""" + lines = 1 + if self._window_height > 0: + lines = self._window_height + self.scroll(-1 * lines) + + def scroll_down_one_page(self): + """Move the selected log index down by one window height.""" + lines = 1 + if self._window_height > 0: + lines = self._window_height + self.scroll(lines) + + def scroll_down(self, lines=1): + """Move the selected log index down by one or more lines.""" + self.scroll(lines) + + def scroll_up(self, lines=1): + """Move the selected log index up by one or more lines.""" + self.scroll(-1 * lines) + + def get_log_window_indices(self, + available_width=None, + available_height=None): + """Get start and end index.""" + self._last_start_index = self._current_start_index + self._last_end_index = self._current_end_index + + starting_index = 0 + ending_index = self.line_index + + self._window_width = self.log_pane.current_log_pane_width + self._window_height = self.log_pane.current_log_pane_height + if available_width: + self._window_width = available_width + if available_height: + self._window_height = available_height + + # If render info is available we use the last window height. + if self._window_height > 0: + # Window lines are zero indexed so subtract 1 from the height. + max_window_row_index = self._window_height - 1 + + starting_index = max(0, self.line_index - max_window_row_index) + # Use the current_window_height if line_index is less + ending_index = max(self.line_index, max_window_row_index) + + if ending_index > self.get_last_log_line_index(): + ending_index = self.get_last_log_line_index() + + # Save start and end index. + self._current_start_index = starting_index + self._current_end_index = ending_index + self.has_new_logs = False + + return starting_index, ending_index + + def draw(self) -> List: + """Return log lines as a list of FormattedText tuples.""" + # If we have no logs add one with at least a single space character for + # the cursor to land on. Otherwise the cursor will be left on the line + # above the log pane container. + if self.get_total_count() == 0: + # No style specified. + return [('', ' \n')] + + starting_index, ending_index = self.get_log_window_indices() + + window_width = self._window_width + total_used_lines = 0 + self._line_fragment_cache = deque() + # Since range() is not inclusive use ending_index + 1. + # for i in range(starting_index, ending_index + 1): + # From the ending_index to the starting index in reverse: + for i in range(ending_index, starting_index - 1, -1): + # If we are past the last valid index. + if i > self.get_last_log_line_index(): + break + + line_fragments = self.logs[i].get_fragments() + + # Get the width of this line. + fragment_width = fragment_list_width(line_fragments) + # Get the line height respecting line wrapping. + line_height = 1 + if self.wrap_lines_enabled() and (fragment_width > window_width): + line_height = get_line_height( + fragment_width, window_width, + self.longest_channel_prefix_width) + + # Keep track of how many lines is used + used_lines = 0 + used_lines += line_height + + # Count the number of line breaks included in the log line. + line_breaks = self.logs[i].record.msg.count('\n') + used_lines += line_breaks + + # If this is the selected line apply a style class for highlighting. + if i == self.line_index: + # Set the cursor to this line + line_fragments = [('[SetCursorPosition]', '')] + line_fragments + # Compute the number of trailing empty characters + + # Calculate the number of spaces to add at the end. + empty_characters = window_width - fragment_width + + # If wrapping is enabled the width of the line prefix needs to + # be accounted for. + if self.wrap_lines_enabled() and (fragment_width > + window_width): + total_width = line_height * window_width + content_width = (self.longest_channel_prefix_width * + (line_height - 1) + fragment_width) + empty_characters = total_width - content_width + + if empty_characters > 0: + line_fragments[-1] = ('', ' ' * empty_characters + '\n') + + line_fragments = to_formatted_text( + line_fragments, style='class:selected-log-line') + + self._line_fragment_cache.appendleft(line_fragments) + total_used_lines += used_lines + # If we have used more lines than available, stop adding new ones. + if total_used_lines > self._window_height: + break + + fragments = [] + for line_fragments in self._line_fragment_cache: + # Append all FormattedText tuples for this line. + for fragment in line_fragments: + fragments.append(fragment) + + # Strip off any trailing line breaks + last_fragment = fragments[-1] + fragments[-1] = (last_fragment[0], last_fragment[1].rstrip('\n')) + + return fragments
diff --git a/pw_console/py/pw_console/log_pane.py b/pw_console/py/pw_console/log_pane.py index bf39a39..653f734 100644 --- a/pw_console/py/pw_console/log_pane.py +++ b/pw_console/py/pw_console/log_pane.py
@@ -16,10 +16,13 @@ from functools import partial from typing import Any, List, Optional +from prompt_toolkit.application.current import get_app from prompt_toolkit.filters import ( Condition, has_focus, ) +from prompt_toolkit.formatted_text import to_formatted_text +from prompt_toolkit.key_binding import KeyBindings, KeyPressEvent from prompt_toolkit.layout import ( ConditionalContainer, Dimension, @@ -27,25 +30,28 @@ FloatContainer, FormattedTextControl, HSplit, + ScrollOffsets, + UIContent, VSplit, + VerticalAlign, Window, WindowAlign, - VerticalAlign, ) from prompt_toolkit.layout.dimension import AnyDimension from prompt_toolkit.mouse_events import MouseEvent, MouseEventType +from pw_console.log_container import LogContainer + class LogPaneLineInfoBar(ConditionalContainer): """One line bar for showing current and total log lines.""" @staticmethod - def get_tokens(unused_log_pane): + def get_tokens(log_pane): """Return formatted text tokens for display.""" tokens = ' Line {} / {} '.format( - # TODO: Replace fake counts with the current line (1) and total - # lines (10). - 1, - 10) + log_pane.log_container.get_current_line() + 1, + log_pane.log_container.get_total_count(), + ) return [('', tokens)] def __init__(self, log_pane): @@ -62,11 +68,7 @@ align=WindowAlign.RIGHT), # Only show current/total line info if not auto-following # logs. Similar to tmux behavior. - filter=Condition( - lambda: True - # TODO: replace True with log follow status. - # not log_pane.log_container.follow - )) + filter=Condition(lambda: not log_pane.log_container.follow)) class LogPaneBottomToolbarBar(ConditionalContainer): @@ -186,6 +188,115 @@ ) +class LogContentControl(FormattedTextControl): + """LogPane prompt_toolkit UIControl for displaying LogContainer lines.""" + @staticmethod + def indent_wrapped_pw_log_format_line(log_pane, line_number, wrap_count): + """Indent wrapped lines to match pw_cli timestamp & level formatter.""" + if wrap_count == 0: + return None + + prefix = ' ' * log_pane.log_container.longest_channel_prefix_width + + # If this line matches the selected log line, highlight it. + if line_number == log_pane.log_container.get_cursor_position().y: + return to_formatted_text(prefix, style='class:selected-log-line') + return prefix + + def create_content(self, width: int, height: Optional[int]) -> UIContent: + # Save redered height + if height: + self.log_pane.last_log_content_height += height + return super().create_content(width, height) + + def __init__(self, log_pane, *args, **kwargs): + self.log_pane = log_pane + + # Key bindings. + key_bindings = KeyBindings() + + @key_bindings.add('w') + def _toggle_wrap_lines(_event: KeyPressEvent) -> None: + """Toggle log line wrapping.""" + self.log_pane.toggle_wrap_lines() + + @key_bindings.add('C') + def _clear_history(_event: KeyPressEvent) -> None: + """Toggle log line wrapping.""" + self.log_pane.clear_history() + + @key_bindings.add('g') + def _scroll_to_top(_event: KeyPressEvent) -> None: + """Scroll to top.""" + self.log_pane.log_container.scroll_to_top() + + @key_bindings.add('G') + def _scroll_to_bottom(_event: KeyPressEvent) -> None: + """Scroll to bottom.""" + self.log_pane.log_container.scroll_to_bottom() + + @key_bindings.add('f') + def _toggle_follow(_event: KeyPressEvent) -> None: + """Toggle log line following.""" + self.log_pane.toggle_follow() + + @key_bindings.add('up') + @key_bindings.add('k') + def _up(_event: KeyPressEvent) -> None: + """Select previous log line.""" + self.log_pane.log_container.scroll_up() + + @key_bindings.add('down') + @key_bindings.add('j') + def _down(_event: KeyPressEvent) -> None: + """Select next log line.""" + self.log_pane.log_container.scroll_down() + + @key_bindings.add('pageup') + def _pageup(_event: KeyPressEvent) -> None: + """Scroll the logs up by one page.""" + self.log_pane.log_container.scroll_up_one_page() + + @key_bindings.add('pagedown') + def _pagedown(_event: KeyPressEvent) -> None: + """Scroll the logs down by one page.""" + self.log_pane.log_container.scroll_down_one_page() + + super().__init__(*args, key_bindings=key_bindings, **kwargs) + + def mouse_handler(self, mouse_event: MouseEvent): + """Mouse handler for this control.""" + mouse_position = mouse_event.position + + # Check for pane focus first. + # If not in focus, change forus to the log pane and do nothing else. + if not has_focus(self)(): + if mouse_event.event_type == MouseEventType.MOUSE_UP: + # Focus buffer when clicked. + get_app().layout.focus(self) + # Mouse event handled, return None. + return None + + if mouse_event.event_type == MouseEventType.MOUSE_UP: + # Scroll to the line clicked. + self.log_pane.log_container.scroll_to_position(mouse_position) + # Mouse event handled, return None. + return None + + if mouse_event.event_type == MouseEventType.SCROLL_DOWN: + self.log_pane.log_container.scroll_down() + # Mouse event handled, return None. + return None + + if mouse_event.event_type == MouseEventType.SCROLL_UP: + self.log_pane.log_container.scroll_up() + # Mouse event handled, return None. + return None + + # Mouse event not handled, return NotImplemented. + return NotImplemented + + class LogLineHSplit(HSplit): """PromptToolkit HSplit class with a write_to_screen function that saves the width and height of the container to be rendered. @@ -234,25 +345,40 @@ self.height = height self.width = width + # Create the log container which stores and handles incoming logs. + self.log_container = LogContainer() + self.log_container.set_log_pane(self) + self.log_container.set_formatting() + # Log pane size variables. These are updated just befor rendering the # pane by the LogLineHSplit class. self.current_log_pane_width = 0 self.current_log_pane_height = 0 self.last_log_pane_width = 0 self.last_log_pane_height = 0 + self.last_log_content_height = 0 # Create the bottom toolbar for the whole log pane. self.bottom_toolbar = LogPaneBottomToolbarBar(self) - # TODO: Render logs here - self.log_content_control = FormattedTextControl( - [('', 'Logs appear here')], + self.log_content_control = LogContentControl( + self, # parent LogPane + # FormattedTextControl args: + self.log_container.draw, + # Hide the cursor, use cursorline=True in self.log_display_window to + # indicate currently selected line. + show_cursor=False, focusable=True, + get_cursor_position=self.log_content_control_get_cursor_position, ) self.log_display_window = Window( content=self.log_content_control, + # TODO: ScrollOffsets here causes jumpiness when lines are wrapped. + scroll_offsets=ScrollOffsets(top=0, bottom=0), allow_scroll_beyond_bottom=True, + get_line_prefix=partial( + LogContentControl.indent_wrapped_pw_log_format_line, self), wrap_lines=Condition(lambda: self.wrap_lines), cursorline=False, # Don't make the window taller to fill the parent split container. @@ -310,12 +436,12 @@ def toggle_follow(self): """Toggle following log lines.""" - # TODO: self.log_container.toggle_follow() + self.log_container.toggle_follow() self.redraw_ui() def clear_history(self): """Erase stored log lines.""" - # TODO: self.log_container.clear_logs() + self.log_container.clear_logs() self.redraw_ui() def __pt_container__(self): @@ -326,7 +452,15 @@ def get_all_key_bindings(self) -> List: """Return all keybinds for this pane.""" # TODO: return log content control keybindings - return [] + return [self.log_content_control.get_key_bindings()] def after_render_hook(self): """Run tasks after the last UI render.""" + self.reset_log_content_height() + + def reset_log_content_height(self): + """Reset log line pane content height.""" + self.last_log_content_height = 0 + + def log_content_control_get_cursor_position(self): + return self.log_container.get_cursor_position()