gh-148105: _pyrepl: switch console refresh to structured rendered screens (#146584)

Author: pablogsal

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):
@@ -450,6 +464,7 @@ def do(self) -> None:
 
         with self.reader.suspend():
             self.reader.msg = _sitebuiltins._Helper()()  # type: ignore[assignment]
+        self.reader.invalidate_prompt()
 
 
 class invalid_key(Command):
@@ -470,22 +485,24 @@ def do(self) -> None:
         from .pager import get_pager
         from site import gethistoryfile
 
+        # After the pager exits, the screen state is unknown (Unix may
+        # restore via alternate screen, Windows shows pager output).
+        # Clear and force a full redraw at the end for consistency.
+        self.reader.console.clear()
+
         history = os.linesep.join(self.reader.history[:])
         self.reader.console.restore()
         pager = get_pager()
         pager(history, gethistoryfile())
         self.reader.console.prepare()
 
-        # 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
+        self.reader.invalidate_full()
 
 
 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,17 +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
-TYPE_CHECKING = False
 if TYPE_CHECKING:
-    from .types import KeySpec, CommandName, CompletionAction
+    from .types import CommandName, CompletionAction, Keymap, KeySpec
 
 
 def prefix(wordlist: list[str], j: int = 0) -> str:
@@ -175,6 +176,8 @@ def do(self) -> None:
                 r.cmpltn_action = None  # consumed
                 if msg:
                     r.msg = msg
+                    r.cmpltn_message_visible = True
+                    r.invalidate_message()
             else:  # other input since last tab: cancel action
                 r.cmpltn_action = None
 
@@ -192,7 +195,8 @@ def do(self) -> None:
             completion = stripcolor(completions[0])
             if completions_unchangable and len(completion) == len(stem):
                 r.msg = "[ sole completion ]"
-                r.dirty = True
+                r.cmpltn_message_visible = True
+                r.invalidate_message()
             r.insert(completion[len(stem):])
         else:
             clean_completions = [stripcolor(word) for word in completions]
@@ -201,19 +205,23 @@ def do(self) -> None:
                 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.cmpltn_message_visible = False
+                    r.invalidate_message()
+                r.invalidate_overlay()
             elif not r.cmpltn_menu_visible:
-                r.cmpltn_message_visible = True
                 if stem + p in clean_completions:
                     r.msg = "[ complete but not unique ]"
-                    r.dirty = True
+                    r.cmpltn_message_visible = True
+                    r.invalidate_message()
                 else:
                     r.msg = "[ not unique ]"
-                    r.dirty = True
+                    r.cmpltn_message_visible = True
+                    r.invalidate_message()
 
         if r.cmpltn_action:
             if r.msg and r.cmpltn_message_visible:
@@ -223,7 +231,7 @@ def do(self) -> None:
             else:
                 r.msg = r.cmpltn_action[0]
                 r.cmpltn_message_visible = True
-                r.dirty = True
+                r.invalidate_message()
 
 
 class self_insert(commands.self_insert):
@@ -243,6 +251,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()
 
@@ -272,7 +281,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'),)
 
@@ -281,25 +290,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()
-        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