mirror of
https://github.com/python/cpython.git
synced 2026-04-20 19:01:14 +00:00
1033 lines
36 KiB
Python
1033 lines
36 KiB
Python
# 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)
|