Improve performance of large pastes by batching text
This commit is contained in:
@@ -715,7 +715,7 @@ class File:
|
||||
def c(self, wch: str, margin: Margin) -> None:
|
||||
s = self.lines[self.y]
|
||||
self.lines[self.y] = s[:self.x] + wch + s[self.x:]
|
||||
self.x = self.x_hint = self.x + 1
|
||||
self.x = self.x_hint = self.x + len(wch)
|
||||
_restore_lines_eof_invariant(self.lines)
|
||||
|
||||
def finalize_previous_action(self) -> None:
|
||||
|
||||
@@ -24,7 +24,8 @@ def _edit(screen: Screen) -> EditResult:
|
||||
ret = Screen.DISPATCH[key.keyname](screen)
|
||||
if isinstance(ret, EditResult):
|
||||
return ret
|
||||
elif isinstance(key.wch, str) and key.wch.isprintable():
|
||||
elif key.keyname == b'STRING':
|
||||
assert isinstance(key.wch, str), key.wch
|
||||
screen.file.c(key.wch, screen.margin)
|
||||
else:
|
||||
screen.status.update(f'unknown key: {key}')
|
||||
|
||||
@@ -2,6 +2,7 @@ import curses
|
||||
import enum
|
||||
from typing import List
|
||||
from typing import Optional
|
||||
from typing import Tuple
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Union
|
||||
|
||||
@@ -98,20 +99,24 @@ class Prompt:
|
||||
def _resize(self) -> None:
|
||||
self._screen.resize()
|
||||
|
||||
def _reverse_search(self) -> Union[None, str, PromptResult]:
|
||||
reverse_s = ''
|
||||
reverse_idx = self._y
|
||||
while True:
|
||||
reverse_failed = False
|
||||
for search_idx in range(reverse_idx, -1, -1):
|
||||
if reverse_s in self._lst[search_idx]:
|
||||
reverse_idx = self._y = search_idx
|
||||
self._x = self._lst[search_idx].index(reverse_s)
|
||||
def _check_failed(self, idx: int, s: str) -> Tuple[bool, int]:
|
||||
failed = False
|
||||
for search_idx in range(idx, -1, -1):
|
||||
if s in self._lst[search_idx]:
|
||||
idx = self._y = search_idx
|
||||
self._x = self._lst[search_idx].index(s)
|
||||
break
|
||||
else:
|
||||
reverse_failed = True
|
||||
failed = True
|
||||
return failed, idx
|
||||
|
||||
if reverse_failed:
|
||||
def _reverse_search(self) -> Union[None, str, PromptResult]:
|
||||
reverse_s = ''
|
||||
idx = self._y
|
||||
while True:
|
||||
fail, idx = self._check_failed(idx, reverse_s)
|
||||
|
||||
if fail:
|
||||
base = f'{self._prompt}(failed reverse-search)`{reverse_s}`'
|
||||
else:
|
||||
base = f'{self._prompt}(reverse-search)`{reverse_s}`'
|
||||
@@ -123,14 +128,17 @@ class Prompt:
|
||||
self._screen.resize()
|
||||
elif key.keyname == b'KEY_BACKSPACE' or key.keyname == b'^H':
|
||||
reverse_s = reverse_s[:-1]
|
||||
elif isinstance(key.wch, str) and key.wch.isprintable():
|
||||
reverse_s += key.wch
|
||||
elif key.keyname == b'^R':
|
||||
reverse_idx = max(0, reverse_idx - 1)
|
||||
idx = max(0, idx - 1)
|
||||
elif key.keyname == b'^C':
|
||||
return self._screen.status.cancelled()
|
||||
elif key.keyname == b'^M':
|
||||
return self._s
|
||||
elif key.keyname == b'STRING':
|
||||
assert isinstance(key.wch, str), key.wch
|
||||
for c in key.wch:
|
||||
reverse_s += c
|
||||
failed, idx = self._check_failed(idx, reverse_s)
|
||||
else:
|
||||
self._x = len(self._s)
|
||||
return None
|
||||
@@ -167,7 +175,7 @@ class Prompt:
|
||||
|
||||
def _c(self, c: str) -> None:
|
||||
self._s = self._s[:self._x] + c + self._s[self._x:]
|
||||
self._x += 1
|
||||
self._x += len(c)
|
||||
|
||||
def run(self) -> Union[PromptResult, str]:
|
||||
while True:
|
||||
@@ -178,5 +186,6 @@ class Prompt:
|
||||
ret = Prompt.DISPATCH[key.keyname](self)
|
||||
if ret is not None:
|
||||
return ret
|
||||
elif isinstance(key.wch, str) and key.wch.isprintable():
|
||||
elif key.keyname == b'STRING':
|
||||
assert isinstance(key.wch, str), key.wch
|
||||
self._c(key.wch)
|
||||
|
||||
@@ -78,6 +78,7 @@ class Screen:
|
||||
self.cut_buffer: Tuple[str, ...] = ()
|
||||
self.cut_selection = False
|
||||
self._resize_cb: Optional[Callable[[], None]] = None
|
||||
self._buffered_input: Union[int, str, None] = None
|
||||
|
||||
@property
|
||||
def file(self) -> File:
|
||||
@@ -97,29 +98,55 @@ class Screen:
|
||||
s = f' {VERSION_STR} {files}{centered}{files}'
|
||||
self.stdscr.insstr(0, 0, s, curses.A_REVERSE)
|
||||
|
||||
def _get_char(self) -> Key:
|
||||
wch = self.stdscr.get_wch()
|
||||
if isinstance(wch, str) and wch == '\x1b':
|
||||
def _get_sequence(self, wch: str) -> str:
|
||||
self.stdscr.nodelay(True)
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
new_wch = self.stdscr.get_wch()
|
||||
if isinstance(new_wch, str):
|
||||
wch += new_wch
|
||||
else: # pragma: no cover (impossible?)
|
||||
curses.unget_wch(new_wch)
|
||||
c = self.stdscr.get_wch()
|
||||
if isinstance(c, str):
|
||||
wch += c
|
||||
else: # pragma: no cover (race)
|
||||
self._buffered_input = c
|
||||
break
|
||||
except curses.error:
|
||||
break
|
||||
finally:
|
||||
self.stdscr.nodelay(False)
|
||||
return wch
|
||||
|
||||
def _get_string(self, wch: str) -> str:
|
||||
self.stdscr.nodelay(True)
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
c = self.stdscr.get_wch()
|
||||
if isinstance(c, str) and c.isprintable():
|
||||
wch += c
|
||||
else:
|
||||
self._buffered_input = c
|
||||
break
|
||||
except curses.error:
|
||||
break
|
||||
finally:
|
||||
self.stdscr.nodelay(False)
|
||||
return wch
|
||||
|
||||
def _get_char(self) -> Key:
|
||||
if self._buffered_input is not None:
|
||||
wch, self._buffered_input = self._buffered_input, None
|
||||
else:
|
||||
wch = self.stdscr.get_wch()
|
||||
if isinstance(wch, str) and wch == '\x1b':
|
||||
wch = self._get_sequence(wch)
|
||||
if len(wch) == 2:
|
||||
return Key(wch, f'M-{wch[1]}'.encode())
|
||||
elif len(wch) > 1:
|
||||
keyname = SEQUENCE_KEYNAME.get(wch, b'unknown')
|
||||
return Key(wch, keyname)
|
||||
elif isinstance(wch, str) and wch.isprintable():
|
||||
wch = self._get_string(wch)
|
||||
return Key(wch, b'STRING')
|
||||
elif wch == '\x7f': # pragma: no cover (macos)
|
||||
keyname = curses.keyname(curses.KEY_BACKSPACE)
|
||||
return Key(wch, keyname)
|
||||
|
||||
@@ -277,7 +277,7 @@ class DeferredRunner:
|
||||
elif s.startswith('M-'):
|
||||
return [KeyPress('\x1b'), KeyPress(s[2:]), CursesError()]
|
||||
else:
|
||||
return [KeyPress(k) for k in s]
|
||||
return [*(KeyPress(k) for k in s), CursesError()]
|
||||
|
||||
def press(self, s):
|
||||
self._ops.extend(self._expand_key(s))
|
||||
@@ -287,7 +287,7 @@ class DeferredRunner:
|
||||
self.press('Enter')
|
||||
|
||||
def answer_no_if_modified(self):
|
||||
self._ops.append(KeyPress('n'))
|
||||
self.press('n')
|
||||
|
||||
@contextlib.contextmanager
|
||||
def resize(self, *, width, height):
|
||||
@@ -344,7 +344,7 @@ class DeferredRunner:
|
||||
# we have already exited -- check remaining things
|
||||
# KeyPress with failing condition or error
|
||||
for i in range(self._i, len(self._ops)):
|
||||
if self._ops[i] != KeyPress('n'):
|
||||
if self._ops[i] not in {KeyPress('n'), CursesError()}:
|
||||
raise AssertionError(self._ops[i:])
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user