_pyrepl: switch console refresh to structured rendered screens

Lib/_pyrepl/commands.py

@@ -22,6 +22,7 @@
 from __future__ import annotations
 import os
 import time
+from typing import TYPE_CHECKING
 
 # Categories of actions:
 #  killing
@@ -32,10 +33,11 @@
 #  finishing
 # [completion]
 
+from .render import RenderedScreen
 from .trace import trace
 
 # types
-if False:
+if TYPE_CHECKING:
     from .historical_reader import HistoricalReader
 
 
@@ -74,7 +76,7 @@ def kill_range(self, start: int, end: int) -> None:
         else:
             r.kill_ring.append(text)
         r.pos = start
-        r.dirty = True
+        r.invalidate_buffer(start)
 
 
 class YankCommand(Command):
@@ -125,24 +127,27 @@ def do(self) -> None:
                     r.arg = 10 * r.arg - d
                 else:
                     r.arg = 10 * r.arg + d
-        r.dirty = True
+        r.invalidate_prompt()
 
 
 class clear_screen(Command):
     def do(self) -> None:
         r = self.reader
+        trace("command.clear_screen")
         r.console.clear()
-        r.dirty = True
+        r.invalidate_full()
 
 
 class refresh(Command):
     def do(self) -> None:
-        self.reader.dirty = True
+        trace("command.refresh")
+        self.reader.invalidate_full()
 
 
 class repaint(Command):
     def do(self) -> None:
-        self.reader.dirty = True
+        trace("command.repaint")
+        self.reader.invalidate_full()
         self.reader.console.repaint()
 
 
@@ -208,9 +213,10 @@ def do(self) -> None:
         repl = len(r.kill_ring[-1])
         r.kill_ring.insert(0, r.kill_ring.pop())
         t = r.kill_ring[-1]
+        start = r.pos - repl
         b[r.pos - repl : r.pos] = t
         r.pos = r.pos - repl + len(t)
-        r.dirty = True
+        r.invalidate_buffer(start)
 
 
 class interrupt(FinishCommand):
@@ -242,8 +248,9 @@ def do(self) -> None:
         r.console.prepare()
         r.pos = p
         # r.posxy = 0, 0  # XXX this is invalid
-        r.dirty = True
-        r.console.screen = []
+        r.invalidate_full()
+        trace("command.suspend sync_rendered_screen")
+        r.console.sync_rendered_screen(RenderedScreen.empty(), r.console.posxy)
 
 
 class up(MotionCommand):
@@ -369,14 +376,15 @@ class self_insert(EditCommand):
     def do(self) -> None:
         r = self.reader
         text = self.event * r.get_arg()
+        start = r.pos
         r.insert(text)
         if r.paste_mode:
             data = ""
             ev = r.console.getpending()
             data += ev.data
             if data:
                 r.insert(data)
-                r.last_refresh_cache.invalidated = True
+                r.invalidate_buffer(start)
 
 
 class insert_nl(EditCommand):
@@ -400,20 +408,23 @@ def do(self) -> None:
             del b[s]
             b.insert(t, c)
             r.pos = t
-            r.dirty = True
+            r.invalidate_buffer(s)
 
 
 class backspace(EditCommand):
     def do(self) -> None:
         r = self.reader
         b = r.buffer
+        changed_from: int | None = None
         for i in range(r.get_arg()):
             if r.pos > 0:
                 r.pos -= 1
                 del b[r.pos]
-                r.dirty = True
+                changed_from = r.pos if changed_from is None else min(changed_from, r.pos)
             else:
                 self.reader.error("can't backspace at start")
+        if changed_from is not None:
+            r.invalidate_buffer(changed_from)
 
 
 class delete(EditCommand):
@@ -431,12 +442,15 @@ def do(self) -> None:
                 r.console.finish()
                 raise EOFError
 
+        changed_from: int | None = None
         for i in range(r.get_arg()):
             if r.pos != len(b):
                 del b[r.pos]
-                r.dirty = True
+                changed_from = r.pos if changed_from is None else min(changed_from, r.pos)
             else:
                 self.reader.error("end of buffer")
+        if changed_from is not None:
+            r.invalidate_buffer(changed_from)
 
 
 class accept(FinishCommand):
@@ -478,14 +492,17 @@ 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):
     def do(self) -> None:
         self.reader.paste_mode = not self.reader.paste_mode
-        self.reader.dirty = True
+        self.reader.invalidate_prompt()
 
 
 class perform_bracketed_paste(Command):
@@ -502,4 +519,3 @@ def do(self) -> None:
             s=time.time() - start,
         )
         self.reader.insert(data.replace(done, ""))
-        self.reader.last_refresh_cache.invalidated = True

Lib/_pyrepl/completing_reader.py

@@ -21,16 +21,18 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
 
 import re
 from . import commands, console, reader
+from .render import RenderLine, ScreenOverlay
 from .reader import Reader
 
 
 # types
 Command = commands.Command
-if False:
-    from .types import KeySpec, CommandName
+if TYPE_CHECKING:
+    from .types import CommandName, KeySpec, Keymap
 
 
 def prefix(wordlist: list[str], j: int = 0) -> str:
@@ -180,27 +182,28 @@ def do(self) -> None:
         elif len(completions) == 1:
             if completions_unchangable and len(completions[0]) == len(stem):
                 r.msg = "[ sole completion ]"
-                r.dirty = True
+                r.invalidate_message()
             r.insert(completions[0][len(stem):])
         else:
             p = prefix(completions, len(stem))
             if p:
                 r.insert(p)
             if last_is_completer:
                 r.cmpltn_menu_visible = True
-                r.cmpltn_message_visible = False
                 r.cmpltn_menu, r.cmpltn_menu_end = build_menu(
                     r.console, completions, r.cmpltn_menu_end,
                     r.use_brackets, r.sort_in_column)
-                r.dirty = True
+                if r.msg:
+                    r.msg = ""
+                    r.invalidate_message()
+                r.invalidate_overlay()
             elif not r.cmpltn_menu_visible:
-                r.cmpltn_message_visible = True
                 if stem + p in completions:
                     r.msg = "[ complete but not unique ]"
-                    r.dirty = True
+                    r.invalidate_message()
                 else:
                     r.msg = "[ not unique ]"
-                    r.dirty = True
+                    r.invalidate_message()
 
 
 class self_insert(commands.self_insert):
@@ -220,6 +223,7 @@ def do(self) -> None:
                     r.cmpltn_menu, r.cmpltn_menu_end = build_menu(
                         r.console, completions, 0,
                         r.use_brackets, r.sort_in_column)
+                    r.invalidate_overlay()
                 else:
                     r.cmpltn_reset()
 
@@ -237,7 +241,6 @@ class CompletingReader(Reader):
     ### Instance variables
     cmpltn_menu: list[str] = field(init=False)
     cmpltn_menu_visible: bool = field(init=False)
-    cmpltn_message_visible: bool = field(init=False)
     cmpltn_menu_end: int = field(init=False)
     cmpltn_menu_choices: list[str] = field(init=False)
 
@@ -248,7 +251,7 @@ def __post_init__(self) -> None:
             self.commands[c.__name__] = c
             self.commands[c.__name__.replace('_', '-')] = c
 
-    def collect_keymap(self) -> tuple[tuple[KeySpec, CommandName], ...]:
+    def collect_keymap(self) -> Keymap:
         return super().collect_keymap() + (
             (r'\t', 'complete'),)
 
@@ -257,28 +260,26 @@ 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()
-        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
-            # 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
+    def get_screen_overlays(self) -> tuple[ScreenOverlay, ...]:
+        if not self.cmpltn_menu_visible:
+            return ()
+        return (
+            ScreenOverlay(
+                self.lxy[1] + 1,
+                tuple(RenderLine.from_rendered_text(line) for line in self.cmpltn_menu),
+                insert=True,
+            ),
+        )
 
     def finish(self) -> None:
         super().finish()
         self.cmpltn_reset()
 
     def cmpltn_reset(self) -> None:
+        if getattr(self, "cmpltn_menu_visible", False):
+            self.invalidate_overlay()
         self.cmpltn_menu = []
         self.cmpltn_menu_visible = False
-        self.cmpltn_message_visible = False
         self.cmpltn_menu_end = 0
         self.cmpltn_menu_choices = []
 

Lib/_pyrepl/console.py

@@ -19,23 +19,25 @@
 
 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 typing import TYPE_CHECKING
 
-
-TYPE_CHECKING = False
+from .render import RenderedScreen
+from .trace import trace
 
 if TYPE_CHECKING:
-    from typing import IO
-    from typing import Callable
+    from typing import Callable, IO
+
+    from .types import CursorXY
 
 
 @dataclass
@@ -47,10 +49,17 @@ class Event:
 
 @dataclass
 class Console(ABC):
-    posxy: tuple[int, int]
-    screen: list[str] = field(default_factory=list)
+    posxy: CursorXY = (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 +80,52 @@ 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: CursorXY | 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.composed_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
+
     @abstractmethod
-    def refresh(self, screen: list[str], xy: tuple[int, int]) -> None: ...
+    def refresh(self, rendered_screen: RenderedScreen) -> None: ...
 
     @abstractmethod
     def prepare(self) -> None: ...