cpython/Lib/_pyrepl/reader.py

#   Copyright 2000-2010 Michael Hudson-Doyle <micahel@gmail.com>
#                       Antonio Cuni
#                       Armin Rigo
#
#                        All Rights Reserved
#
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose is hereby granted without fee,
# provided that the above copyright notice appear in all copies and
# that both that copyright notice and this permission notice appear in
# supporting documentation.
#
# THE AUTHOR MICHAEL HUDSON DISCLAIMS ALL WARRANTIES WITH REGARD TO
# THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS, IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
# INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

from __future__ import annotations

import sys
import _colorize

from contextlib import contextmanager
from dataclasses import dataclass, field, fields, replace
from typing import Self

from . import commands, console, input
from .content import (
    ContentFragment,
    ContentLine,
    SourceLine,
    build_body_fragments,
    process_prompt as build_prompt_content,
)
from .layout import LayoutMap, LayoutResult, LayoutRow, WrappedRow, layout_content_lines
from .render import RenderCell, RenderLine, RenderedScreen, ScreenOverlay
from .utils import ANSI_ESCAPE_SEQUENCE, THEME, StyleRef, wlen, gen_colors
from .trace import trace


# types
Command = commands.Command
from .types import (
    Callback,
    CommandName,
    CursorXY,
    Dimensions,
    EventData,
    KeySpec,
    Keymap,
    ScreenInfoRow,
    SimpleContextManager,
)

type CommandClass = type[Command]
type CommandInput = tuple[CommandName | CommandClass, EventData]
type PromptCellCacheKey = tuple[str, bool]


# syntax classes
SYNTAX_WHITESPACE, SYNTAX_WORD, SYNTAX_SYMBOL = range(3)


def make_default_syntax_table() -> dict[str, int]:
    # XXX perhaps should use some unicodedata here?
    st: dict[str, int] = {}
    for c in map(chr, range(256)):
        st[c] = SYNTAX_SYMBOL
    for c in [a for a in map(chr, range(256)) if a.isalnum()]:
        st[c] = SYNTAX_WORD
    st["\n"] = st[" "] = SYNTAX_WHITESPACE
    return st


def make_default_commands() -> dict[CommandName, CommandClass]:
    result: dict[CommandName, CommandClass] = {}
    for v in vars(commands).values():
        if isinstance(v, type) and issubclass(v, Command) and v.__name__[0].islower():
            result[v.__name__] = v
            result[v.__name__.replace("_", "-")] = v
    return result


default_keymap: Keymap = tuple(
    [
        (r"\C-a", "beginning-of-line"),
        (r"\C-b", "left"),
        (r"\C-c", "interrupt"),
        (r"\C-d", "delete"),
        (r"\C-e", "end-of-line"),
        (r"\C-f", "right"),
        (r"\C-g", "cancel"),
        (r"\C-h", "backspace"),
        (r"\C-j", "accept"),
        (r"\<return>", "accept"),
        (r"\C-k", "kill-line"),
        (r"\C-l", "clear-screen"),
        (r"\C-m", "accept"),
        (r"\C-t", "transpose-characters"),
        (r"\C-u", "unix-line-discard"),
        (r"\C-w", "unix-word-rubout"),
        (r"\C-x\C-u", "upcase-region"),
        (r"\C-y", "yank"),
        *(() if sys.platform == "win32" else ((r"\C-z", "suspend"), )),
        (r"\M-b", "backward-word"),
        (r"\M-c", "capitalize-word"),
        (r"\M-d", "kill-word"),
        (r"\M-f", "forward-word"),
        (r"\M-l", "downcase-word"),
        (r"\M-t", "transpose-words"),
        (r"\M-u", "upcase-word"),
        (r"\M-y", "yank-pop"),
        (r"\M--", "digit-arg"),
        (r"\M-0", "digit-arg"),
        (r"\M-1", "digit-arg"),
        (r"\M-2", "digit-arg"),
        (r"\M-3", "digit-arg"),
        (r"\M-4", "digit-arg"),
        (r"\M-5", "digit-arg"),
        (r"\M-6", "digit-arg"),
        (r"\M-7", "digit-arg"),
        (r"\M-8", "digit-arg"),
        (r"\M-9", "digit-arg"),
        (r"\M-\n", "accept"),
        ("\\\\", "self-insert"),
        (r"\x1b[200~", "perform-bracketed-paste"),
        (r"\x03", "ctrl-c"),
    ]
    + [(c, "self-insert") for c in map(chr, range(32, 127)) if c != "\\"]
    + [(c, "self-insert") for c in map(chr, range(128, 256)) if c.isalpha()]
    + [
        (r"\<up>", "up"),
        (r"\<down>", "down"),
        (r"\<left>", "left"),
        (r"\C-\<left>", "backward-word"),
        (r"\<right>", "right"),
        (r"\C-\<right>", "forward-word"),
        (r"\<delete>", "delete"),
        (r"\x1b[3~", "delete"),
        (r"\<backspace>", "backspace"),
        (r"\M-\<backspace>", "backward-kill-word"),
        (r"\<end>", "end-of-line"),  # was 'end'
        (r"\<home>", "beginning-of-line"),  # was 'home'
        (r"\<f1>", "help"),
        (r"\<f2>", "show-history"),
        (r"\<f3>", "paste-mode"),
        (r"\EOF", "end"),  # the entries in the terminfo database for xterms
        (r"\EOH", "home"),  # seem to be wrong.  this is a less than ideal
        # workaround
    ]
)


@dataclass(frozen=True, slots=True)
class RefreshInvalidation:
    """Which parts of the screen need to be recomputed on the next refresh."""

    cursor_only: bool = False
    buffer_from_pos: int | None = None
    prompt: bool = False
    layout: bool = False
    theme: bool = False
    message: bool = False
    overlay: bool = False
    full: bool = False

    @classmethod
    def empty(cls) -> Self:
        return cls()

    @property
    def needs_screen_refresh(self) -> bool:
        return any(
            (
                self.buffer_from_pos is not None,
                self.prompt,
                self.layout,
                self.theme,
                self.message,
                self.overlay,
                self.full,
            )
        )

    @property
    def is_cursor_only(self) -> bool:
        return self.cursor_only and not self.needs_screen_refresh

    @property
    def buffer_rebuild_from_pos(self) -> int | None:
        if self.full or self.prompt or self.layout or self.theme:
            return 0
        return self.buffer_from_pos

    def with_cursor(self) -> Self:
        if self.needs_screen_refresh:
            return self
        return replace(self, cursor_only=True)

    def with_buffer(self, from_pos: int) -> Self:
        current = from_pos
        if self.buffer_from_pos is not None:
            current = min(current, self.buffer_from_pos)
        return replace(self, cursor_only=False, buffer_from_pos=current)

    def with_prompt(self) -> Self:
        return replace(self, cursor_only=False, prompt=True)

    def with_layout(self) -> Self:
        return replace(self, cursor_only=False, layout=True)

    def with_theme(self) -> Self:
        return replace(self, cursor_only=False, theme=True)

    def with_message(self) -> Self:
        return replace(self, cursor_only=False, message=True)

    def with_overlay(self) -> Self:
        return replace(self, cursor_only=False, overlay=True)

    def with_full(self) -> Self:
        return replace(self, cursor_only=False, full=True)


@dataclass(slots=True)
class Reader:
    """The Reader class implements the bare bones of a command reader,
    handling such details as editing and cursor motion.  What it does
    not support are such things as completion or history support -
    these are implemented elsewhere.

    Instance variables of note include:

      * buffer:
        A per-character list containing all the characters that have been
        entered. Does not include color information.
      * console:
        Hopefully encapsulates the OS dependent stuff.
      * pos:
        A 0-based index into 'buffer' for where the insertion point
        is.
      * layout:
        A mapping between buffer positions and rendered rows/columns.
        It is the internal source of truth for cursor placement.
      * cxy, lxy:
        the position of the insertion point in screen ...
      * syntax_table:
        Dictionary mapping characters to 'syntax class'; read the
        emacs docs to see what this means :-)
      * commands:
        Dictionary mapping command names to command classes.
      * arg:
        The emacs-style prefix argument.  It will be None if no such
        argument has been provided.
      * kill_ring:
        The emacs-style kill-ring; manipulated with yank & yank-pop
      * ps1, ps2, ps3, ps4:
        prompts.  ps1 is the prompt for a one-line input; for a
        multiline input it looks like:
            ps2> first line of input goes here
            ps3> second and further
            ps3> lines get ps3
            ...
            ps4> and the last one gets ps4
        As with the usual top-level, you can set these to instances if
        you like; str() will be called on them (once) at the beginning
        of each command.  Don't put really long or newline containing
        strings here, please!
        This is just the default policy; you can change it freely by
        overriding get_prompt() (and indeed some standard subclasses
        do).
      * finished:
        handle1 will set this to a true value if a command signals
        that we're done.
    """

    console: console.Console

    ## state
    buffer: list[str] = field(default_factory=list)
    pos: int = 0
    ps1: str = "->> "
    ps2: str = "/>> "
    ps3: str = "|.. "
    ps4: str = R"\__ "
    kill_ring: list[list[str]] = field(default_factory=list)
    msg: str = ""
    arg: int | None = None
    finished: bool = False
    paste_mode: bool = False
    commands: dict[CommandName, CommandClass] = field(default_factory=make_default_commands)
    last_command: CommandClass | None = None
    syntax_table: dict[str, int] = field(default_factory=make_default_syntax_table)
    keymap: Keymap = ()
    input_trans: input.KeymapTranslator = field(init=False)
    input_trans_stack: list[input.KeymapTranslator] = field(default_factory=list)
    rendered_screen: RenderedScreen = field(init=False)
    layout: LayoutMap = field(init=False)
    cxy: CursorXY = field(init=False)
    lxy: CursorXY = field(init=False)
    scheduled_commands: list[CommandName] = field(default_factory=list)
    can_colorize: bool = False
    threading_hook: Callback | None = None
    invalidation: RefreshInvalidation = field(init=False)

    ## cached metadata to speed up screen refreshes
    @dataclass
    class RefreshCache:
        """Previously computed render/layout data for incremental refresh."""

        render_lines: list[RenderLine] = field(default_factory=list)
        layout_rows: list[LayoutRow] = field(default_factory=list)
        line_end_offsets: list[int] = field(default_factory=list)
        pos: int = 0
        dimensions: Dimensions = (0, 0)

        def update_cache(self,
                         reader: Reader,
                         render_lines: list[RenderLine],
                         layout_rows: list[LayoutRow],
                         line_end_offsets: list[int],
            ) -> None:
            self.render_lines = render_lines.copy()
            self.layout_rows = layout_rows.copy()
            self.line_end_offsets = line_end_offsets.copy()
            self.pos = reader.pos
            self.dimensions = reader.console.width, reader.console.height

        def valid(self, reader: Reader) -> bool:
            dimensions = reader.console.width, reader.console.height
            dimensions_changed = dimensions != self.dimensions
            return not dimensions_changed

        def get_cached_location(
            self,
            reader: Reader,
            buffer_from_pos: int | None = None,
            *,
            reuse_full: bool = False,
        ) -> tuple[int, int]:
            """Return (buffer_offset, num_reusable_lines) for incremental refresh.

            Three paths:
            - reuse_full (overlay/message-only): reuse all cached lines.
            - buffer_from_pos=None (full rebuild): rewind to common cursor pos.
            - explicit buffer_from_pos: reuse lines before that position.
            """
            if reuse_full:
                if self.line_end_offsets:
                    last_offset = self.line_end_offsets[-1]
                    if last_offset >= len(reader.buffer):
                        return last_offset, len(self.line_end_offsets)
                return 0, 0
            if buffer_from_pos is None:
                buffer_from_pos = min(reader.pos, self.pos)
            num_common_lines = len(self.line_end_offsets)
            while num_common_lines > 0:
                candidate = self.line_end_offsets[num_common_lines - 1]
                if buffer_from_pos > candidate:
                    break
                num_common_lines -= 1
            # Prompt-only leading rows consume no buffer content. Reusing them
            # in isolation causes the next incremental rebuild to emit them a
            # second time.
            while (
                num_common_lines > 0
                and self.layout_rows[num_common_lines - 1].buffer_advance == 0
            ):
                num_common_lines -= 1
            offset = self.line_end_offsets[num_common_lines - 1] if num_common_lines else 0
            return offset, num_common_lines

    last_refresh_cache: RefreshCache = field(default_factory=RefreshCache)

    def __post_init__(self) -> None:
        # Enable the use of `insert` without a `prepare` call - necessary to
        # facilitate the tab completion hack implemented for
        # <https://bugs.python.org/issue25660>.
        self.keymap = self.collect_keymap()
        self.input_trans = input.KeymapTranslator(
            self.keymap, invalid_cls="invalid-key", character_cls="self-insert"
        )
        self.layout = LayoutMap.empty()
        self.cxy = self.pos2xy()
        self.lxy = (self.pos, 0)
        self.rendered_screen = RenderedScreen.empty()
        self.can_colorize = _colorize.can_colorize()
        self.invalidation = RefreshInvalidation.empty()

        self.last_refresh_cache.layout_rows = list(self.layout.rows)
        self.last_refresh_cache.pos = self.pos

        self.last_refresh_cache.dimensions = (0, 0)

    @property
    def screen(self) -> list[str]:
        return list(self.rendered_screen.screen_lines)

    @property
    def screeninfo(self) -> list[ScreenInfoRow]:
        return self.layout.screeninfo

    def collect_keymap(self) -> Keymap:
        return default_keymap

    def calc_screen(self) -> RenderedScreen:
        """Translate the editable buffer into a base rendered screen."""
        num_common_lines = 0
        offset = 0
        if self.last_refresh_cache.valid(self):
            if (
                self.invalidation.buffer_from_pos is None
                and not (
                    self.invalidation.full
                    or self.invalidation.prompt
                    or self.invalidation.layout
                    or self.invalidation.theme
                )
                and (self.invalidation.message or self.invalidation.overlay)
            ):
                # Fast path: only overlays or messages changed.
                offset, num_common_lines = self.last_refresh_cache.get_cached_location(
                    self,
                    reuse_full=True,
                )
                assert not self.last_refresh_cache.line_end_offsets or (
                    self.last_refresh_cache.line_end_offsets[-1] >= len(self.buffer)
                ), "Buffer modified without invalidate_buffer() call"
            else:
                offset, num_common_lines = self.last_refresh_cache.get_cached_location(
                    self,
                    self._buffer_refresh_from_pos(),
                )

        base_render_lines = self.last_refresh_cache.render_lines[:num_common_lines]
        layout_rows = self.last_refresh_cache.layout_rows[:num_common_lines]
        last_refresh_line_end_offsets = self.last_refresh_cache.line_end_offsets[:num_common_lines]

        source_lines = self._build_source_lines(offset, num_common_lines)
        content_lines = self._build_content_lines(
            source_lines,
            prompt_from_cache=bool(offset and self.buffer[offset - 1] != "\n"),
        )
        layout_result = self._layout_content(content_lines, offset)
        base_render_lines.extend(self._render_wrapped_rows(layout_result.wrapped_rows))
        layout_rows.extend(layout_result.layout_map.rows)
        last_refresh_line_end_offsets.extend(layout_result.line_end_offsets)

        self.layout = LayoutMap(tuple(layout_rows))
        self.cxy = self.pos2xy()
        if not source_lines:
            # reuse_full path: _build_source_lines didn't run,
            # so lxy wasn't updated. Derive it from the buffer.
            self.lxy = self._compute_lxy()
        self.last_refresh_cache.update_cache(
            self,
            base_render_lines,
            layout_rows,
            last_refresh_line_end_offsets,
        )
        return RenderedScreen(tuple(base_render_lines), self.cxy)

    def _buffer_refresh_from_pos(self) -> int:
        """Return buffer position from which to rebuild content.

        Returns 0 (full rebuild) when no incremental position is known.
        """
        buffer_from_pos = self.invalidation.buffer_rebuild_from_pos
        if buffer_from_pos is not None:
            return buffer_from_pos
        return 0

    def _compute_lxy(self) -> CursorXY:
        """Derive logical cursor (col, lineno) from the buffer and pos."""
        text = "".join(self.buffer[:self.pos])
        lineno = text.count("\n")
        if lineno:
            col = self.pos - text.rindex("\n") - 1
        else:
            col = self.pos
        return col, lineno

    def _build_source_lines(
        self,
        offset: int,
        first_lineno: int,
    ) -> tuple[SourceLine, ...]:
        if offset == len(self.buffer) and (offset > 0 or first_lineno > 0):
            return ()

        pos = self.pos - offset
        lines = "".join(self.buffer[offset:]).split("\n")
        cursor_found = False
        lines_beyond_cursor = 0
        source_lines: list[SourceLine] = []
        current_offset = offset

        for line_index, line in enumerate(lines):
            lineno = first_lineno + line_index
            has_newline = line_index < len(lines) - 1
            line_len = len(line)
            cursor_index: int | None = None
            if 0 <= pos <= line_len:
                cursor_index = pos
                self.lxy = pos, lineno
                cursor_found = True
            elif cursor_found:
                lines_beyond_cursor += 1
                if lines_beyond_cursor > self.console.height:
                    break

            source_lines.append(
                SourceLine(
                    lineno=lineno,
                    text=line,
                    start_offset=current_offset,
                    has_newline=has_newline,
                    cursor_index=cursor_index,
                )
            )
            pos -= line_len + 1
            current_offset += line_len + (1 if has_newline else 0)

        return tuple(source_lines)

    def _build_content_lines(
        self,
        source_lines: tuple[SourceLine, ...],
        *,
        prompt_from_cache: bool,
    ) -> tuple[ContentLine, ...]:
        if self.can_colorize:
            colors = list(gen_colors(self.get_unicode()))
        else:
            colors = None
        trace("colors = {colors}", colors=colors)

        content_lines: list[ContentLine] = []
        for source_line in source_lines:
            if prompt_from_cache:
                prompt_from_cache = False
                prompt = ""
            else:
                prompt = self.get_prompt(source_line.lineno, source_line.cursor_on_line)
            content_lines.append(
                ContentLine(
                    source=source_line,
                    prompt=build_prompt_content(prompt),
                    body=build_body_fragments(
                        source_line.text,
                        colors,
                        source_line.start_offset,
                    ),
                )
            )
        return tuple(content_lines)

    def _layout_content(
        self,
        content_lines: tuple[ContentLine, ...],
        offset: int,
    ) -> LayoutResult:
        return layout_content_lines(content_lines, self.console.width, offset)

    def _render_wrapped_rows(
        self,
        wrapped_rows: tuple[WrappedRow, ...],
    ) -> list[RenderLine]:
        return [
            self._render_line(
                row.prompt_text,
                row.fragments,
                row.suffix,
            )
            for row in wrapped_rows
        ]

    def _render_message_lines(self) -> tuple[RenderLine, ...]:
        if not self.msg:
            return ()
        width = self.console.width
        render_lines: list[RenderLine] = []
        for message_line in self.msg.split("\n"):
            # If self.msg is larger than console width, make it fit.
            # TODO: try to split between words?
            if not message_line:
                render_lines.append(RenderLine.from_rendered_text(""))
                continue
            for offset in range(0, len(message_line), width):
                render_lines.append(
                    RenderLine.from_rendered_text(message_line[offset : offset + width])
                )
        return tuple(render_lines)

    def get_screen_overlays(self) -> tuple[ScreenOverlay, ...]:
        return ()

    def compose_rendered_screen(self, base_screen: RenderedScreen) -> RenderedScreen:
        overlays = list(self.get_screen_overlays())
        message_lines = self._render_message_lines()
        if message_lines:
            overlays.append(ScreenOverlay(len(base_screen.lines), message_lines))
        if not overlays:
            return base_screen
        return RenderedScreen(base_screen.lines, base_screen.cursor, tuple(overlays))

    _prompt_cell_cache: dict[PromptCellCacheKey, tuple[RenderCell, ...]] = field(
        init=False, default_factory=dict, repr=False
    )

    def _render_line(
        self,
        prefix: str,
        fragments: tuple[ContentFragment, ...],
        suffix: str = "",
    ) -> RenderLine:
        cells: list[RenderCell] = []
        if prefix:
            cache_key = (prefix, self.can_colorize)
            cached = self._prompt_cell_cache.get(cache_key)
            if cached is None:
                prompt_cells = RenderLine.from_rendered_text(prefix).cells
                if self.can_colorize and prompt_cells and not ANSI_ESCAPE_SEQUENCE.search(prefix):
                    prompt_style = StyleRef.from_tag("prompt", THEME()["prompt"])
                    prompt_cells = tuple(
                        RenderCell(
                            cell.text,
                            cell.width,
                            style=prompt_style if cell.text else cell.style,
                            controls=cell.controls,
                        )
                        for cell in prompt_cells
                    )
                self._prompt_cell_cache[cache_key] = prompt_cells
                cached = prompt_cells
            cells.extend(cached)
        cells.extend(
            RenderCell(fragment.text, fragment.width, style=fragment.style)
            for fragment in fragments
        )
        if suffix:
            cells.extend(RenderLine.from_rendered_text(suffix).cells)
        return RenderLine.from_cells(cells)

    @staticmethod
    def process_prompt(prompt: str) -> tuple[str, int]:
        r"""Return a tuple with the prompt string and its visible length.

        The prompt string has the zero-width brackets recognized by shells
        (\x01 and \x02) removed.  The length ignores anything between those
        brackets as well as any ANSI escape sequences.
        """
        prompt_content = build_prompt_content(prompt)
        return prompt_content.text, prompt_content.width

    def bow(self, p: int | None = None) -> int:
        """Return the 0-based index of the word break preceding p most
        immediately.

        p defaults to self.pos; word boundaries are determined using
        self.syntax_table."""
        if p is None:
            p = self.pos
        st = self.syntax_table
        b = self.buffer
        p -= 1
        while p >= 0 and st.get(b[p], SYNTAX_WORD) != SYNTAX_WORD:
            p -= 1
        while p >= 0 and st.get(b[p], SYNTAX_WORD) == SYNTAX_WORD:
            p -= 1
        return p + 1

    def eow(self, p: int | None = None) -> int:
        """Return the 0-based index of the word break following p most
        immediately.

        p defaults to self.pos; word boundaries are determined using
        self.syntax_table."""
        if p is None:
            p = self.pos
        st = self.syntax_table
        b = self.buffer
        while p < len(b) and st.get(b[p], SYNTAX_WORD) != SYNTAX_WORD:
            p += 1
        while p < len(b) and st.get(b[p], SYNTAX_WORD) == SYNTAX_WORD:
            p += 1
        return p

    def bol(self, p: int | None = None) -> int:
        """Return the 0-based index of the line break preceding p most
        immediately.

        p defaults to self.pos."""
        if p is None:
            p = self.pos
        b = self.buffer
        p -= 1
        while p >= 0 and b[p] != "\n":
            p -= 1
        return p + 1

    def eol(self, p: int | None = None) -> int:
        """Return the 0-based index of the line break following p most
        immediately.

        p defaults to self.pos."""
        if p is None:
            p = self.pos
        b = self.buffer
        while p < len(b) and b[p] != "\n":
            p += 1
        return p

    def max_column(self, y: int) -> int:
        """Return the last x-offset for line y"""
        return self.layout.max_column(y)

    def max_row(self) -> int:
        return self.layout.max_row()

    def get_arg(self, default: int = 1) -> int:
        """Return any prefix argument that the user has supplied,
        returning 'default' if there is None.  Defaults to 1.
        """
        if self.arg is None:
            return default
        return self.arg

    def get_prompt(self, lineno: int, cursor_on_line: bool) -> str:
        """Return what should be in the left-hand margin for line
        'lineno'."""
        if self.arg is not None and cursor_on_line:
            prompt = f"(arg: {self.arg}) "
        elif self.paste_mode:
            prompt = "(paste) "
        elif "\n" in self.buffer:
            if lineno == 0:
                prompt = self.ps2
            elif self.ps4 and lineno == self.buffer.count("\n"):
                prompt = self.ps4
            else:
                prompt = self.ps3
        else:
            prompt = self.ps1
        return prompt

    def push_input_trans(self, itrans: input.KeymapTranslator) -> None:
        self.input_trans_stack.append(self.input_trans)
        self.input_trans = itrans

    def pop_input_trans(self) -> None:
        self.input_trans = self.input_trans_stack.pop()

    def setpos_from_xy(self, x: int, y: int) -> None:
        """Set pos according to coordinates x, y"""
        self.pos = self.layout.xy_to_pos(x, y)

    def pos2xy(self) -> CursorXY:
        """Return the x, y coordinates of position 'pos'."""
        assert 0 <= self.pos <= len(self.buffer)
        return self.layout.pos_to_xy(self.pos)

    def insert(self, text: str | list[str]) -> None:
        """Insert 'text' at the insertion point."""
        start = self.pos
        self.buffer[self.pos : self.pos] = list(text)
        self.pos += len(text)
        self.invalidate_buffer(start)

    def invalidate_cursor(self) -> None:
        self.invalidation = self.invalidation.with_cursor()

    def invalidate_buffer(self, from_pos: int) -> None:
        self.invalidation = self.invalidation.with_buffer(from_pos)

    def invalidate_prompt(self) -> None:
        self._prompt_cell_cache.clear()
        self.invalidation = self.invalidation.with_prompt()

    def invalidate_layout(self) -> None:
        self.invalidation = self.invalidation.with_layout()

    def invalidate_theme(self) -> None:
        self._prompt_cell_cache.clear()
        self.invalidation = self.invalidation.with_theme()

    def invalidate_message(self) -> None:
        self.invalidation = self.invalidation.with_message()

    def invalidate_overlay(self) -> None:
        self.invalidation = self.invalidation.with_overlay()

    def invalidate_full(self) -> None:
        self.invalidation = self.invalidation.with_full()

    def clear_invalidation(self) -> None:
        self.invalidation = RefreshInvalidation.empty()

    def update_cursor(self) -> None:
        """Move the cursor to reflect changes in self.pos"""
        self.cxy = self.pos2xy()
        trace("update_cursor({pos}) = {cxy}", pos=self.pos, cxy=self.cxy)
        self.console.move_cursor(*self.cxy)

    def after_command(self, cmd: Command) -> None:
        """This function is called to allow post command cleanup."""
        if getattr(cmd, "kills_digit_arg", True):
            if self.arg is not None:
                self.invalidate_prompt()
            self.arg = None

    def prepare(self) -> None:
        """Get ready to run.  Call restore when finished.  You must not
        write to the console in between the calls to prepare and
        restore."""
        try:
            self.console.prepare()
            self.arg = None
            self.finished = False
            del self.buffer[:]
            self.pos = 0
            self.layout = LayoutMap.empty()
            self.cxy = self.pos2xy()
            self.lxy = (self.pos, 0)
            self.rendered_screen = RenderedScreen.empty()
            self.invalidate_full()
            self.last_command = None
            base_screen = self.calc_screen()
            self.rendered_screen = self.compose_rendered_screen(base_screen)
            self.invalidation = RefreshInvalidation.empty()
        except BaseException:
            self.restore()
            raise

        while self.scheduled_commands:
            cmd = self.scheduled_commands.pop()
            self.do_cmd((cmd, []))

    def last_command_is(self, cls: CommandClass) -> bool:
        if not self.last_command:
            return False
        return issubclass(cls, self.last_command)

    def restore(self) -> None:
        """Clean up after a run."""
        self.console.restore()

    @contextmanager
    def suspend(self) -> SimpleContextManager:
        """A context manager to delegate to another reader."""
        prev_state = {f.name: getattr(self, f.name) for f in fields(self)}
        try:
            self.restore()
            yield
        finally:
            for arg in ("msg", "ps1", "ps2", "ps3", "ps4", "paste_mode"):
                setattr(self, arg, prev_state[arg])
            self.prepare()

    @contextmanager
    def suspend_colorization(self) -> SimpleContextManager:
        try:
            old_can_colorize = self.can_colorize
            self.can_colorize = False
            yield
        finally:
            self.can_colorize = old_can_colorize

    def finish(self) -> None:
        """Called when a command signals that we're finished."""
        pass

    def error(self, msg: str = "none") -> None:
        self.msg = "! " + msg + " "
        self.invalidate_message()
        self.console.beep()

    def update_screen(self) -> None:
        if self.invalidation.is_cursor_only:
            self.update_cursor()
            self.clear_invalidation()
        elif self.invalidation.needs_screen_refresh:
            self.refresh()

    def refresh(self) -> None:
        """Recalculate and refresh the screen."""
        self.console.height, self.console.width = self.console.getheightwidth()
        # this call sets up self.cxy, so call it first.
        base_screen = self.calc_screen()
        rendered_screen = self.compose_rendered_screen(base_screen)
        self.rendered_screen = rendered_screen
        trace(
            "reader.refresh cursor={cursor} lines={lines} "
            "dims=({width},{height}) invalidation={invalidation}",
            cursor=self.cxy,
            lines=len(rendered_screen.composed_lines),
            width=self.console.width,
            height=self.console.height,
            invalidation=self.invalidation,
        )
        self.console.refresh(rendered_screen)
        self.clear_invalidation()

    def do_cmd(self, cmd: CommandInput) -> None:
        """`cmd` is a tuple of "event_name" and "event", which in the current
        implementation is always just the "buffer" which happens to be a list
        of single-character strings."""

        trace("received command {cmd}", cmd=cmd)
        if isinstance(cmd[0], str):
            command_type = self.commands.get(cmd[0], commands.invalid_command)
        elif isinstance(cmd[0], type):
            command_type = cmd[0]
        else:
            return  # nothing to do

        command = command_type(self, *cmd)  # type: ignore[arg-type]
        command.do()

        self.after_command(command)
        if (
            not self.invalidation.needs_screen_refresh
            and not self.invalidation.is_cursor_only
        ):
            self.invalidate_cursor()
        self.update_screen()

        if command_type is not commands.digit_arg:
            self.last_command = command_type

        self.finished = bool(command.finish)
        if self.finished:
            self.console.finish()
            self.finish()

    def run_hooks(self) -> None:
        threading_hook = self.threading_hook
        if threading_hook is None and 'threading' in sys.modules:
            from ._threading_handler import install_threading_hook
            install_threading_hook(self)
        if threading_hook is not None:
            try:
                threading_hook()
            except Exception:
                pass

        input_hook = self.console.input_hook
        if input_hook:
            try:
                input_hook()
            except Exception:
                pass

    def handle1(self, block: bool = True) -> bool:
        """Handle a single event.  Wait as long as it takes if block
        is true (the default), otherwise return False if no event is
        pending."""

        if self.msg:
            self.msg = ""
            self.invalidate_message()

        while True:
            # We use the same timeout as in readline.c: 100ms
            self.run_hooks()
            self.console.wait(100)
            event = self.console.get_event(block=False)
            if not event:
                if block:
                    continue
                return False

            translate = True

            if event.evt == "key":
                self.input_trans.push(event)
            elif event.evt == "scroll":
                self.invalidate_full()
                self.refresh()
                return True
            elif event.evt == "resize":
                self.invalidate_full()
                self.refresh()
                return True
            else:
                translate = False

            if translate:
                cmd = self.input_trans.get()
            else:
                cmd = [event.evt, event.data]

            if cmd is None:
                if block:
                    continue
                return False

            self.do_cmd(cmd)
            return True

    def push_char(self, char: int | bytes) -> None:
        self.console.push_char(char)
        self.handle1(block=False)

    def readline(self, startup_hook: Callback | None = None) -> str:
        """Read a line.  The implementation of this method also shows
        how to drive Reader if you want more control over the event
        loop."""
        self.prepare()
        try:
            if startup_hook is not None:
                startup_hook()
            self.refresh()
            while not self.finished:
                self.handle1()
            return self.get_unicode()

        finally:
            self.restore()

    def bind(self, spec: KeySpec, command: CommandName) -> None:
        self.keymap = self.keymap + ((spec, command),)
        self.input_trans = input.KeymapTranslator(
            self.keymap, invalid_cls="invalid-key", character_cls="self-insert"
        )

    def get_unicode(self) -> str:
        """Return the current buffer as a unicode string."""
        return "".join(self.buffer)