Switch _pyrepl consoles to rendered screens

Teach the reader and terminal backends to refresh from RenderedScreen objects.
This isolates the redraw-planning refactor before layout and styling changes land.

Author: pablogsal

Lib/_pyrepl/commands.py

@@ -32,6 +32,7 @@
 #  finishing
 # [completion]
 
+from .render import RenderedScreen
 from .trace import trace
 
 # types
@@ -131,17 +132,20 @@ def do(self) -> None:
 class clear_screen(Command):
     def do(self) -> None:
         r = self.reader
+        trace("command.clear_screen")
         r.console.clear()
         r.dirty = True
 
 
 class refresh(Command):
     def do(self) -> None:
+        trace("command.refresh")
         self.reader.dirty = True
 
 
 class repaint(Command):
     def do(self) -> None:
+        trace("command.repaint")
         self.reader.dirty = True
         self.reader.console.repaint()
 
@@ -243,7 +247,8 @@ def do(self) -> None:
         r.pos = p
         # r.posxy = 0, 0  # XXX this is invalid
         r.dirty = True
-        r.console.screen = []
+        trace("command.suspend sync_rendered_screen")
+        r.console.sync_rendered_screen(RenderedScreen.empty(), r.console.posxy)
 
 
 class up(MotionCommand):
@@ -478,8 +483,11 @@ def do(self) -> None:
 
         # We need to copy over the state so that it's consistent between
         # console and reader, and console does not overwrite/append stuff
-        self.reader.console.screen = self.reader.screen.copy()
-        self.reader.console.posxy = self.reader.cxy
+        trace("command.show_history sync_rendered_screen")
+        self.reader.console.sync_rendered_screen(
+            self.reader.rendered_screen,
+            self.reader.cxy,
+        )
 
 
 class paste_mode(Command):

Lib/_pyrepl/completing_reader.py

@@ -24,6 +24,7 @@
 
 import re
 from . import commands, console, reader
+from .render import RenderLine, RenderedScreen
 from .reader import Reader
 
 
@@ -257,19 +258,24 @@ def after_command(self, cmd: Command) -> None:
         if not isinstance(cmd, (complete, self_insert)):
             self.cmpltn_reset()
 
-    def calc_screen(self) -> list[str]:
-        screen = super().calc_screen()
+    def calc_screen(self) -> RenderedScreen:
+        rendered_screen = super().calc_screen()
         if self.cmpltn_menu_visible:
             # We display the completions menu below the current prompt
             ly = self.lxy[1] + 1
-            screen[ly:ly] = self.cmpltn_menu
+            render_lines = list(rendered_screen.lines)
+            render_lines[ly:ly] = [
+                RenderLine.from_rendered_text(line) for line in self.cmpltn_menu
+            ]
+            rendered_screen = RenderedScreen(tuple(render_lines), self.cxy)
+            self.rendered_screen = rendered_screen
             # If we're not in the middle of multiline edit, don't append to screeninfo
             # since that screws up the position calculation in pos2xy function.
             # This is a hack to prevent the cursor jumping
             # into the completions menu when pressing left or down arrow.
             if self.pos != len(self.buffer):
                 self.screeninfo[ly:ly] = [(0, [])]*len(self.cmpltn_menu)
-        return screen
+        return rendered_screen
 
     def finish(self) -> None:
         super().finish()

Lib/_pyrepl/console.py

@@ -19,17 +19,20 @@
 
 from __future__ import annotations
 
+import os
 import _colorize
 
 from abc import ABC, abstractmethod
 import ast
 import code
 import linecache
-from dataclasses import dataclass, field
-import os.path
+from dataclasses import dataclass
 import re
 import sys
 
+from .render import RenderedScreen
+from .trace import trace
+
 
 TYPE_CHECKING = False
 
@@ -47,10 +50,17 @@ class Event:
 
 @dataclass
 class Console(ABC):
-    posxy: tuple[int, int]
-    screen: list[str] = field(default_factory=list)
+    posxy: tuple[int, int] = (0, 0)
     height: int = 25
     width: int = 80
+    _redraw_debug_palette: tuple[str, ...] = (
+        "\x1b[41m",
+        "\x1b[42m",
+        "\x1b[43m",
+        "\x1b[44m",
+        "\x1b[45m",
+        "\x1b[46m",
+    )
 
     def __init__(
         self,
@@ -71,8 +81,58 @@ def __init__(
         else:
             self.output_fd = f_out.fileno()
 
+        self.posxy = (0, 0)
+        self.height = 25
+        self.width = 80
+        self._rendered_screen = RenderedScreen.empty()
+        self._redraw_visual_cycle = 0
+
+    @property
+    def screen(self) -> list[str]:
+        return list(self._rendered_screen.screen_lines)
+
+    def sync_rendered_screen(
+        self,
+        rendered_screen: RenderedScreen,
+        posxy: tuple[int, int] | None = None,
+    ) -> None:
+        if posxy is None:
+            posxy = rendered_screen.cursor
+        self.posxy = posxy
+        self._rendered_screen = rendered_screen
+        trace(
+            "console.sync_rendered_screen lines={lines} cursor={cursor}",
+            lines=len(rendered_screen.lines),
+            cursor=posxy,
+        )
+
+    def invalidate_render_state(self) -> None:
+        self._rendered_screen = RenderedScreen.empty()
+        trace("console.invalidate_render_state")
+
+    def begin_redraw_visualization(self) -> str | None:
+        if "PYREPL_VISUALIZE_REDRAWS" not in os.environ:
+            return None
+
+        palette = self._redraw_debug_palette
+        cycle = self._redraw_visual_cycle
+        style = palette[cycle % len(palette)]
+        self._redraw_visual_cycle = cycle + 1
+        trace(
+            "console.begin_redraw_visualization cycle={cycle} style={style!r}",
+            cycle=cycle,
+            style=style,
+        )
+        return style
+
+    @staticmethod
+    def visualize_redraw_text(text: str, style: str | None) -> str:
+        if style is None or not text:
+            return text
+        return style + text.replace("\x1b[0m", "\x1b[0m" + style) + "\x1b[0m"
+
     @abstractmethod
-    def refresh(self, screen: list[str], xy: tuple[int, int]) -> None: ...
+    def refresh(self, rendered_screen: RenderedScreen) -> None: ...
 
     @abstractmethod
     def prepare(self) -> None: ...

Lib/_pyrepl/reader.py

@@ -28,6 +28,7 @@
 from dataclasses import dataclass, field, fields
 
 from . import commands, console, input
+from .render import RenderCell, RenderLine, RenderedScreen
 from .utils import wlen, unbracket, disp_str, gen_colors, THEME
 from .trace import trace
 
@@ -207,7 +208,7 @@ class Reader:
     keymap: tuple[tuple[str, str], ...] = ()
     input_trans: input.KeymapTranslator = field(init=False)
     input_trans_stack: list[input.KeymapTranslator] = field(default_factory=list)
-    screen: list[str] = field(default_factory=list)
+    rendered_screen: RenderedScreen = field(init=False)
     screeninfo: list[tuple[int, list[int]]] = field(init=False)
     cxy: tuple[int, int] = field(init=False)
     lxy: tuple[int, int] = field(init=False)
@@ -218,7 +219,7 @@ class Reader:
     ## cached metadata to speed up screen refreshes
     @dataclass
     class RefreshCache:
-        screen: list[str] = field(default_factory=list)
+        render_lines: list[RenderLine] = field(default_factory=list)
         screeninfo: list[tuple[int, list[int]]] = field(init=False)
         line_end_offsets: list[int] = field(default_factory=list)
         pos: int = field(init=False)
@@ -228,11 +229,13 @@ class RefreshCache:
 
         def update_cache(self,
                          reader: Reader,
-                         screen: list[str],
+                         render_lines: list[RenderLine],
                          screeninfo: list[tuple[int, list[int]]],
+                         line_end_offsets: list[int],
             ) -> None:
-            self.screen = screen.copy()
+            self.render_lines = render_lines.copy()
             self.screeninfo = screeninfo.copy()
+            self.line_end_offsets = line_end_offsets.copy()
             self.pos = reader.pos
             self.cxy = reader.cxy
             self.dimensions = reader.console.width, reader.console.height
@@ -273,18 +276,23 @@ def __post_init__(self) -> None:
         self.screeninfo = [(0, [])]
         self.cxy = self.pos2xy()
         self.lxy = (self.pos, 0)
+        self.rendered_screen = RenderedScreen.empty()
         self.can_colorize = _colorize.can_colorize()
 
         self.last_refresh_cache.screeninfo = self.screeninfo
         self.last_refresh_cache.pos = self.pos
         self.last_refresh_cache.cxy = self.cxy
         self.last_refresh_cache.dimensions = (0, 0)
 
+    @property
+    def screen(self) -> list[str]:
+        return list(self.rendered_screen.screen_lines)
+
     def collect_keymap(self) -> tuple[tuple[KeySpec, CommandName], ...]:
         return default_keymap
 
-    def calc_screen(self) -> list[str]:
-        """Translate changes in self.buffer into changes in self.console.screen."""
+    def calc_screen(self) -> RenderedScreen:
+        """Translate changes in self.buffer into a structured rendered screen."""
         # Since the last call to calc_screen:
         # screen and screeninfo may differ due to a completion menu being shown
         # pos and cxy may differ due to edits, cursor movements, or completion menus
@@ -297,14 +305,9 @@ def calc_screen(self) -> list[str]:
         if self.last_refresh_cache.valid(self):
             offset, num_common_lines = self.last_refresh_cache.get_cached_location(self)
 
-        screen = self.last_refresh_cache.screen
-        del screen[num_common_lines:]
-
-        screeninfo = self.last_refresh_cache.screeninfo
-        del screeninfo[num_common_lines:]
-
-        last_refresh_line_end_offsets = self.last_refresh_cache.line_end_offsets
-        del last_refresh_line_end_offsets[num_common_lines:]
+        render_lines = self.last_refresh_cache.render_lines[:num_common_lines]
+        screeninfo = self.last_refresh_cache.screeninfo[:num_common_lines]
+        last_refresh_line_end_offsets = self.last_refresh_cache.line_end_offsets[:num_common_lines]
 
         pos = self.pos
         pos -= offset
@@ -339,7 +342,7 @@ def calc_screen(self) -> list[str]:
             while "\n" in prompt:
                 pre_prompt, _, prompt = prompt.partition("\n")
                 last_refresh_line_end_offsets.append(offset)
-                screen.append(pre_prompt)
+                render_lines.append(RenderLine.from_rendered_text(pre_prompt))
                 screeninfo.append((0, []))
             pos -= line_len + 1
             prompt, prompt_len = self.process_prompt(prompt)
@@ -348,7 +351,8 @@ def calc_screen(self) -> list[str]:
             if wrapcount == 0 or not char_widths:
                 offset += line_len + 1  # Takes all of the line plus the newline
                 last_refresh_line_end_offsets.append(offset)
-                screen.append(prompt + "".join(chars))
+                render_line = self._render_line(prompt, chars, char_widths)
+                render_lines.append(render_line)
                 screeninfo.append((prompt_len, char_widths))
             else:
                 pre = prompt
@@ -370,9 +374,14 @@ def calc_screen(self) -> list[str]:
                         post = ""
                         after = []
                     last_refresh_line_end_offsets.append(offset)
-                    render = pre + "".join(chars[:index_to_wrap_before]) + post
+                    render_line = self._render_line(
+                        pre,
+                        chars[:index_to_wrap_before],
+                        char_widths[:index_to_wrap_before],
+                        post,
+                    )
                     render_widths = char_widths[:index_to_wrap_before] + after
-                    screen.append(render)
+                    render_lines.append(render_line)
                     screeninfo.append((prelen, render_widths))
                     chars = chars[index_to_wrap_before:]
                     char_widths = char_widths[index_to_wrap_before:]
@@ -382,11 +391,35 @@ def calc_screen(self) -> list[str]:
         self.cxy = self.pos2xy()
         if self.msg:
             for mline in self.msg.split("\n"):
-                screen.append(mline)
+                render_lines.append(RenderLine.from_rendered_text(mline))
                 screeninfo.append((0, []))
 
-        self.last_refresh_cache.update_cache(self, screen, screeninfo)
-        return screen
+        self.rendered_screen = RenderedScreen(tuple(render_lines), self.cxy)
+        self.last_refresh_cache.update_cache(
+            self,
+            render_lines,
+            screeninfo,
+            last_refresh_line_end_offsets,
+        )
+        return self.rendered_screen
+
+    @staticmethod
+    def _render_line(
+        prefix: str,
+        chars: list[str],
+        char_widths: list[int],
+        suffix: str = "",
+    ) -> RenderLine:
+        cells: list[RenderCell] = []
+        if prefix:
+            cells.extend(RenderLine.from_rendered_text(prefix).cells)
+        cells.extend(
+            RenderCell(text, width, "\x1b" in text)
+            for text, width in zip(chars, char_widths)
+        )
+        if suffix:
+            cells.append(RenderCell(suffix, wlen(suffix), "\x1b" in suffix))
+        return RenderLine.from_cells(cells)
 
     @staticmethod
     def process_prompt(prompt: str) -> tuple[str, int]:
@@ -645,8 +678,17 @@ def update_screen(self) -> None:
     def refresh(self) -> None:
         """Recalculate and refresh the screen."""
         # this call sets up self.cxy, so call it first.
-        self.screen = self.calc_screen()
-        self.console.refresh(self.screen, self.cxy)
+        rendered_screen = self.calc_screen()
+        trace(
+            "reader.refresh cursor={cursor} lines={lines} "
+            "dims=({width},{height}) dirty={dirty}",
+            cursor=self.cxy,
+            lines=len(rendered_screen.lines),
+            width=self.console.width,
+            height=self.console.height,
+            dirty=self.dirty,
+        )
+        self.console.refresh(rendered_screen)
         self.dirty = False
 
     def do_cmd(self, cmd: tuple[str, list[str]]) -> None:

Lib/_pyrepl/trace.py

@@ -32,3 +32,9 @@ def trace(line: str, *k: object, **kw: object) -> None:
             line = line.format(*k, **kw)
         trace_file.write(line + "\n")
         trace_file.flush()
+
+
+def trace_text(text: str, limit: int = 60) -> str:
+    if len(text) > limit:
+        text = text[:limit] + "..."
+    return repr(text)