Slurp the entire input buffer before refreshing display during bracketed paste

Author: ambv

Lib/_pyrepl/commands.py

@@ -31,6 +31,7 @@
 #  finishing
 # [completion]
 
+from .trace import trace
 
 # types
 if False:
@@ -471,19 +472,23 @@ def do(self) -> None:
 
 
 class paste_mode(Command):
-
     def do(self) -> None:
         self.reader.paste_mode = not self.reader.paste_mode
         self.reader.dirty = True
 
 
-class enable_bracketed_paste(Command):
-    def do(self) -> None:
-        self.reader.paste_mode = True
-        self.reader.in_bracketed_paste = True
-
-class disable_bracketed_paste(Command):
-    def do(self) -> None:
-        self.reader.paste_mode = False
-        self.reader.in_bracketed_paste = False
-        self.reader.dirty = True
+class perform_bracketed_paste(Command):
+    def do(self) -> None:
+        done = "\x1b[201~"
+        data = ""
+        import time
+        start = time.time()
+        trace("bracketed pasting starts")
+        while done not in data:
+            self.reader.console.wait(100)
+            ev = self.reader.console.getpending()
+            data += ev.data
+            trace("len(data) = {d}", d=len(data))
+        trace("bracketed pasting done in {s:.2f}s", s=time.time() - start)
+        self.reader.insert(data.replace(done, ""))
+        self.reader.last_refresh_cache.invalidated = True

Lib/_pyrepl/reader.py

@@ -103,8 +103,7 @@ def make_default_commands() -> dict[CommandName, type[Command]]:
         (r"\M-9", "digit-arg"),
         (r"\M-\n", "accept"),
         ("\\\\", "self-insert"),
-        (r"\x1b[200~", "enable_bracketed_paste"),
-        (r"\x1b[201~", "disable_bracketed_paste"),
+        (r"\x1b[200~", "perform-bracketed-paste"),
         (r"\x03", "ctrl-c"),
     ]
     + [(c, "self-insert") for c in map(chr, range(32, 127)) if c != "\\"]
@@ -202,7 +201,6 @@ class Reader:
     dirty: bool = False
     finished: bool = False
     paste_mode: bool = False
-    in_bracketed_paste: bool = False
     commands: dict[str, type[Command]] = field(default_factory=make_default_commands)
     last_command: type[Command] | None = None
     syntax_table: dict[str, int] = field(default_factory=make_default_syntax_table)
@@ -220,7 +218,6 @@ class Reader:
     ## cached metadata to speed up screen refreshes
     @dataclass
     class RefreshCache:
-        in_bracketed_paste: bool = False
         screen: list[str] = field(default_factory=list)
         screeninfo: list[tuple[int, list[int]]] = field(init=False)
         line_end_offsets: list[int] = field(default_factory=list)
@@ -234,7 +231,6 @@ def update_cache(self,
                          screen: list[str],
                          screeninfo: list[tuple[int, list[int]]],
             ) -> None:
-            self.in_bracketed_paste = reader.in_bracketed_paste
             self.screen = screen.copy()
             self.screeninfo = screeninfo.copy()
             self.pos = reader.pos
@@ -247,8 +243,7 @@ def valid(self, reader: Reader) -> bool:
                 return False
             dimensions = reader.console.width, reader.console.height
             dimensions_changed = dimensions != self.dimensions
-            paste_changed = reader.in_bracketed_paste != self.in_bracketed_paste
-            return not (dimensions_changed or paste_changed)
+            return not dimensions_changed
 
         def get_cached_location(self, reader: Reader) -> tuple[int, int]:
             if self.invalidated:
@@ -350,7 +345,7 @@ def calc_screen(self) -> list[str]:
             prompt, prompt_len = self.process_prompt(prompt)
             chars, char_widths = disp_str(line, colors, offset)
             wrapcount = (sum(char_widths) + prompt_len) // self.console.width
-            trace("wrapcount = {wrapcount}", wrapcount=wrapcount)
+            # trace("wrapcount = {wrapcount}", wrapcount=wrapcount)
             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)
@@ -484,7 +479,7 @@ def get_prompt(self, lineno: int, cursor_on_line: bool) -> str:
         'lineno'."""
         if self.arg is not None and cursor_on_line:
             prompt = f"(arg: {self.arg}) "
-        elif self.paste_mode and not self.in_bracketed_paste:
+        elif self.paste_mode:
             prompt = "(paste) "
         elif "\n" in self.buffer:
             if lineno == 0:
@@ -639,9 +634,6 @@ def update_screen(self) -> None:
 
     def refresh(self) -> None:
         """Recalculate and refresh the screen."""
-        if self.in_bracketed_paste and self.buffer and not self.buffer[-1] == "\n":
-            return
-
         # this call sets up self.cxy, so call it first.
         self.screen = self.calc_screen()
         self.console.refresh(self.screen, self.cxy)

Lib/_pyrepl/readline.py

@@ -276,10 +276,6 @@ def do(self) -> None:
         r = self.reader  # type: ignore[assignment]
         r.dirty = True  # this is needed to hide the completion menu, if visible
 
-        if self.reader.in_bracketed_paste:
-            r.insert("\n")
-            return
-
         # if there are already several lines and the cursor
         # is not on the last one, always insert a new \n.
         text = r.get_unicode()

Lib/_pyrepl/simple_interact.py

@@ -157,7 +157,6 @@ def maybe_run_command(statement: str) -> bool:
             r.pos = len(r.get_unicode())
             r.dirty = True
             r.refresh()
-            r.in_bracketed_paste = False
             console.write("\nKeyboardInterrupt\n")
             console.resetbuffer()
         except MemoryError:

Lib/_pyrepl/unix_console.py

@@ -150,8 +150,6 @@ def __init__(
 
         self.pollob = poll()
         self.pollob.register(self.input_fd, select.POLLIN)
-        self.input_buffer = b""
-        self.input_buffer_pos = 0
         curses.setupterm(term or None, self.output_fd)
         self.term = term
 
@@ -199,22 +197,8 @@ def _my_getstr(cap: str, optional: bool = False) -> bytes | None:
         self.event_queue = EventQueue(self.input_fd, self.encoding)
         self.cursor_visible = 1
 
-    def more_in_buffer(self) -> bool:
-        return bool(
-            self.input_buffer
-            and self.input_buffer_pos < len(self.input_buffer)
-        )
-
     def __read(self, n: int) -> bytes:
-        if not self.more_in_buffer():
-            self.input_buffer = os.read(self.input_fd, 10000)
-
-        ret = self.input_buffer[self.input_buffer_pos : self.input_buffer_pos + n]
-        self.input_buffer_pos += len(ret)
-        if self.input_buffer_pos >= len(self.input_buffer):
-            self.input_buffer = b""
-            self.input_buffer_pos = 0
-        return ret
+        return os.read(self.input_fd, n)
 
 
     def change_encoding(self, encoding: str) -> None:
@@ -422,7 +406,6 @@ def wait(self, timeout: float | None = None) -> bool:
         """
         return (
             not self.event_queue.empty()
-            or self.more_in_buffer()
             or bool(self.pollob.poll(timeout))
         )
 
@@ -525,6 +508,7 @@ def getpending(self):
                 e.raw += e.raw
 
             amount = struct.unpack("i", ioctl(self.input_fd, FIONREAD, b"\0\0\0\0"))[0]
+            trace("getpending({a})", a=amount)
             raw = self.__read(amount)
             data = str(raw, self.encoding, "replace")
             e.data += data

Lib/_pyrepl/utils.py

@@ -304,5 +304,5 @@ def disp_str(
         # the next call to `disp_str()` will revive it.
         chars[-1] += TAG_TO_ANSI["SYNC"]
 
-    trace("disp_str({buffer}) = {s}, {b}", buffer=repr(buffer), s=chars, b=char_widths)
+    # trace("disp_str({buffer}) = {s}, {b}", buffer=repr(buffer), s=chars, b=char_widths)
     return chars, char_widths