| # 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. | |
| 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, ColorSpan, THEME, StyleRef, gen_colors | |
| from .trace import trace | |
| # types | |
| Command = commands.Command | |
| from collections.abc import Callable, Iterator | |
| 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 | |
| gen_colors: Callable[[str], Iterator[ColorSpan]] = gen_colors | |
| 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(self.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) |