Add esc_timeout parameter to BaseEventQueue

Author: tanloong

Lib/_pyrepl/base_eventqueue.py

@@ -24,20 +24,26 @@
 See unix_eventqueue and windows_eventqueue for subclasses.
 """
 
+import os
 from collections import deque
 
 from . import keymap
 from .console import Event
 from .trace import trace
 
+ESC_TIMEOUT_DEFAULT = 0.1
+
 class BaseEventQueue:
-    def __init__(self, encoding: str, keymap_dict: dict[bytes, str]) -> None:
+    def __init__(self, encoding: str, keymap_dict: dict[bytes, str],
+                 esc_timeout: float | None = None) -> None:
         self.compiled_keymap = keymap.compile_keymap(keymap_dict)
         self.keymap = self.compiled_keymap
         trace("keymap {k!r}", k=self.keymap)
         self.encoding = encoding
         self.events: deque[Event] = deque()
         self.buf = bytearray()
+        default = float(os.environ.get("PYREPL_ESC_TIMEOUT", ESC_TIMEOUT_DEFAULT))
+        self.esc_timeout = esc_timeout if esc_timeout is not None else default
 
     def get(self) -> Event | None:
         """
@@ -63,25 +69,13 @@ def pending(self) -> bool:
         knows only as a prefix of a longer sequence has been pushed, but
         before the rest of that sequence arrives.  Consoles use this to
         decide whether the next read should block indefinitely or only for
-        the escape timeout (see ``flush``).
+        the escape timeout.
         """
         return self.keymap is not self.compiled_keymap
 
     def flush(self) -> None:
         """
         Finalize any pending, incomplete input as discrete events.
-
-        This implements the classic "escape-time" disambiguation used by
-        TUIs such as ncurses, Vim and Tmux: when we have received the start
-        of an escape sequence (e.g. a lone ESC) but no further bytes arrive
-        within the escape timeout, we stop waiting and emit the bytes
-        accumulated so far as a complete key instead.
-
-        The behavior mirrors that of an unrecognized escape sequence: the
-        leading ESC is emitted as a key and any remaining bytes are fed
-        back through ``push`` so they can start another sequence or be
-        treated as ordinary text.  Calling this when there is nothing
-        pending is a no-op.
         """
         if self.keymap is self.compiled_keymap:
             # Nothing pending: either idle, or a complete key was already

Lib/_pyrepl/unix_eventqueue.py

@@ -69,9 +69,10 @@ def get_terminal_keycodes(ti: TermInfo) -> dict[bytes, str]:
 
 
 class EventQueue(BaseEventQueue):
-    def __init__(self, fd: int, encoding: str, ti: TermInfo) -> None:
+    def __init__(self, fd: int, encoding: str, ti: TermInfo,
+                 esc_timeout: float | None = None) -> None:
         keycodes = get_terminal_keycodes(ti)
         if os.isatty(fd):
             backspace = tcgetattr(fd)[6][VERASE]
             keycodes[backspace] = "backspace"
-        BaseEventQueue.__init__(self, encoding, keycodes)
+        BaseEventQueue.__init__(self, encoding, keycodes, esc_timeout)

Lib/_pyrepl/windows_eventqueue.py

@@ -38,5 +38,5 @@
 }
 
 class EventQueue(BaseEventQueue):
-    def __init__(self, encoding: str) -> None:
-        BaseEventQueue.__init__(self, encoding, VT_MAP)
+    def __init__(self, encoding: str, esc_timeout: float | None = None) -> None:
+        BaseEventQueue.__init__(self, encoding, VT_MAP, esc_timeout)