mirror of
https://github.com/python/cpython.git
synced 2026-04-21 03:10:52 +00:00
1068 lines
34 KiB
Python
1068 lines
34 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 errno
|
|
import os
|
|
import re
|
|
import select
|
|
import signal
|
|
import struct
|
|
import termios
|
|
import time
|
|
import types
|
|
import platform
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from fcntl import ioctl
|
|
from typing import TYPE_CHECKING, cast, overload
|
|
|
|
from . import terminfo
|
|
from .console import Console, Event
|
|
from .fancy_termios import tcgetattr, tcsetattr, TermState
|
|
from .render import (
|
|
EMPTY_RENDER_LINE,
|
|
LineUpdate,
|
|
RenderLine,
|
|
RenderedScreen,
|
|
requires_cursor_resync,
|
|
diff_render_lines,
|
|
render_cells,
|
|
)
|
|
from .trace import trace, trace_text
|
|
from .unix_eventqueue import EventQueue
|
|
|
|
# declare posix optional to allow None assignment on other platforms
|
|
posix: types.ModuleType | None
|
|
try:
|
|
import posix
|
|
except ImportError:
|
|
posix = None
|
|
|
|
# types
|
|
if TYPE_CHECKING:
|
|
from typing import AbstractSet, IO, Literal
|
|
|
|
type _MoveFunc = Callable[[int, int], None]
|
|
type _PendingWrite = tuple[str | bytes, bool]
|
|
|
|
|
|
class InvalidTerminal(RuntimeError):
|
|
def __init__(self, message: str) -> None:
|
|
super().__init__(errno.EIO, message)
|
|
|
|
|
|
_error = (termios.error, InvalidTerminal)
|
|
_error_codes_to_ignore = frozenset([errno.EIO, errno.ENXIO, errno.EPERM])
|
|
|
|
SIGWINCH_EVENT = "repaint"
|
|
|
|
FIONREAD = getattr(termios, "FIONREAD", None)
|
|
TIOCGWINSZ = getattr(termios, "TIOCGWINSZ", None)
|
|
|
|
# ------------ start of baudrate definitions ------------
|
|
|
|
# Add (possibly) missing baudrates (check termios man page) to termios
|
|
|
|
|
|
def add_baudrate_if_supported(dictionary: dict[int, int], rate: int) -> None:
|
|
baudrate_name = "B%d" % rate
|
|
if hasattr(termios, baudrate_name):
|
|
dictionary[getattr(termios, baudrate_name)] = rate
|
|
|
|
|
|
# Check the termios man page (Line speed) to know where these
|
|
# values come from.
|
|
potential_baudrates = [
|
|
0,
|
|
110,
|
|
115200,
|
|
1200,
|
|
134,
|
|
150,
|
|
1800,
|
|
19200,
|
|
200,
|
|
230400,
|
|
2400,
|
|
300,
|
|
38400,
|
|
460800,
|
|
4800,
|
|
50,
|
|
57600,
|
|
600,
|
|
75,
|
|
9600,
|
|
]
|
|
|
|
ratedict: dict[int, int] = {}
|
|
for rate in potential_baudrates:
|
|
add_baudrate_if_supported(ratedict, rate)
|
|
|
|
# Clean up variables to avoid unintended usage
|
|
del rate, add_baudrate_if_supported
|
|
|
|
# ------------ end of baudrate definitions ------------
|
|
|
|
delayprog = re.compile(b"\\$<([0-9]+)((?:/|\\*){0,2})>")
|
|
|
|
try:
|
|
poll: type[select.poll] = select.poll
|
|
except AttributeError:
|
|
# this is exactly the minimum necessary to support what we
|
|
# do with poll objects
|
|
class MinimalPoll:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def register(self, fd, flag):
|
|
self.fd = fd
|
|
|
|
# note: The 'timeout' argument is received as *milliseconds*
|
|
def poll(self, timeout: float | None = None) -> list[int]:
|
|
if timeout is None:
|
|
r, w, e = select.select([self.fd], [], [])
|
|
else:
|
|
r, w, e = select.select([self.fd], [], [], timeout / 1000)
|
|
return r
|
|
|
|
poll = MinimalPoll # type: ignore[assignment]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class UnixRefreshPlan:
|
|
"""Instructions for updating the terminal after a screen change.
|
|
|
|
After the user types ``e`` to complete ``name``::
|
|
|
|
Before: >>> def greet(nam|):
|
|
â–²
|
|
LineUpdate here: insert_char "e"
|
|
|
|
After: >>> def greet(name|):
|
|
â–²
|
|
|
|
Only the changed cells are sent to the terminal; unchanged rows
|
|
are skipped entirely.
|
|
"""
|
|
|
|
grow_lines: int
|
|
"""Number of blank lines to append at the bottom to accommodate new content."""
|
|
use_tall_mode: bool
|
|
"""Use absolute cursor addressing via ``cup`` instead of relative moves.
|
|
Activated when content exceeds one screen height."""
|
|
offset: int
|
|
"""Vertical scroll offset: the buffer row displayed at the top of the terminal window."""
|
|
reverse_scroll: int
|
|
"""Number of lines to scroll backwards (content moves down)."""
|
|
forward_scroll: int
|
|
"""Number of lines to scroll forwards (content moves up)."""
|
|
line_updates: tuple[LineUpdate, ...]
|
|
cleared_lines: tuple[int, ...]
|
|
"""Row indices to erase (old content with no replacement)."""
|
|
rendered_screen: RenderedScreen
|
|
cursor: tuple[int, int]
|
|
|
|
|
|
class UnixConsole(Console):
|
|
__buffer: list[_PendingWrite]
|
|
__gone_tall: bool
|
|
__move: _MoveFunc
|
|
__offset: int
|
|
|
|
def __init__(
|
|
self,
|
|
f_in: IO[bytes] | int = 0,
|
|
f_out: IO[bytes] | int = 1,
|
|
term: str = "",
|
|
encoding: str = "",
|
|
):
|
|
"""
|
|
Initialize the UnixConsole.
|
|
|
|
Parameters:
|
|
- f_in (int or file-like object): Input file descriptor or object.
|
|
- f_out (int or file-like object): Output file descriptor or object.
|
|
- term (str): Terminal name.
|
|
- encoding (str): Encoding to use for I/O operations.
|
|
"""
|
|
super().__init__(f_in, f_out, term, encoding)
|
|
|
|
self.pollob = poll()
|
|
self.pollob.register(self.input_fd, select.POLLIN)
|
|
self.terminfo = terminfo.TermInfo(term or None)
|
|
self.term = term
|
|
self.is_apple_terminal = (
|
|
platform.system() == "Darwin"
|
|
and os.getenv("TERM_PROGRAM") == "Apple_Terminal"
|
|
)
|
|
|
|
try:
|
|
self.__input_fd_set(tcgetattr(self.input_fd), ignore=frozenset())
|
|
except _error as e:
|
|
raise RuntimeError(f"termios failure ({e.args[1]})")
|
|
|
|
@overload
|
|
def _my_getstr(
|
|
cap: str, optional: Literal[False] = False
|
|
) -> bytes: ...
|
|
|
|
@overload
|
|
def _my_getstr(cap: str, optional: bool) -> bytes | None: ...
|
|
|
|
def _my_getstr(cap: str, optional: bool = False) -> bytes | None:
|
|
r = self.terminfo.get(cap)
|
|
if not optional and r is None:
|
|
raise InvalidTerminal(
|
|
f"terminal doesn't have the required {cap} capability"
|
|
)
|
|
return r
|
|
|
|
self._bel = _my_getstr("bel")
|
|
self._civis = _my_getstr("civis", optional=True)
|
|
self._clear = _my_getstr("clear")
|
|
self._cnorm = _my_getstr("cnorm", optional=True)
|
|
self._cub = _my_getstr("cub", optional=True)
|
|
self._cub1 = _my_getstr("cub1", optional=True)
|
|
self._cud = _my_getstr("cud", optional=True)
|
|
self._cud1 = _my_getstr("cud1", optional=True)
|
|
self._cuf = _my_getstr("cuf", optional=True)
|
|
self._cuf1 = _my_getstr("cuf1", optional=True)
|
|
self._cup = _my_getstr("cup")
|
|
self._cuu = _my_getstr("cuu", optional=True)
|
|
self._cuu1 = _my_getstr("cuu1", optional=True)
|
|
self._dch1 = _my_getstr("dch1", optional=True)
|
|
self._dch = _my_getstr("dch", optional=True)
|
|
self._el = _my_getstr("el")
|
|
self._hpa = _my_getstr("hpa", optional=True)
|
|
self._ich = _my_getstr("ich", optional=True)
|
|
self._ich1 = _my_getstr("ich1", optional=True)
|
|
self._ind = _my_getstr("ind", optional=True)
|
|
self._pad = _my_getstr("pad", optional=True)
|
|
self._ri = _my_getstr("ri", optional=True)
|
|
self._rmkx = _my_getstr("rmkx", optional=True)
|
|
self._smkx = _my_getstr("smkx", optional=True)
|
|
|
|
self.__setup_movement()
|
|
|
|
self.event_queue = EventQueue(
|
|
self.input_fd, self.encoding, self.terminfo
|
|
)
|
|
self.cursor_visible = True
|
|
|
|
signal.signal(signal.SIGCONT, self._sigcont_handler)
|
|
|
|
def _sigcont_handler(self, signum, frame):
|
|
self.restore()
|
|
self.prepare()
|
|
|
|
def __read(self, n: int) -> bytes:
|
|
return os.read(self.input_fd, n)
|
|
|
|
def change_encoding(self, encoding: str) -> None:
|
|
"""
|
|
Change the encoding used for I/O operations.
|
|
|
|
Parameters:
|
|
- encoding (str): New encoding to use.
|
|
"""
|
|
self.encoding = encoding
|
|
|
|
def refresh(self, rendered_screen: RenderedScreen) -> None:
|
|
"""
|
|
Refresh the console screen.
|
|
|
|
Parameters:
|
|
- rendered_screen: Structured rendered screen contents and cursor.
|
|
"""
|
|
c_xy = rendered_screen.cursor
|
|
trace(
|
|
"unix.refresh start cursor={cursor} lines={lines} prev_lines={prev_lines} "
|
|
"offset={offset} posxy={posxy}",
|
|
cursor=c_xy,
|
|
lines=len(rendered_screen.composed_lines),
|
|
prev_lines=len(self._rendered_screen.composed_lines),
|
|
offset=self.__offset,
|
|
posxy=self.posxy,
|
|
)
|
|
plan = self.__plan_refresh(rendered_screen, c_xy)
|
|
self.__apply_refresh_plan(plan)
|
|
|
|
def __plan_refresh(
|
|
self,
|
|
rendered_screen: RenderedScreen,
|
|
c_xy: tuple[int, int],
|
|
) -> UnixRefreshPlan:
|
|
cx, cy = c_xy
|
|
height = self.height
|
|
old_offset = offset = self.__offset
|
|
prev_composed = self._rendered_screen.composed_lines
|
|
previous_lines = list(prev_composed)
|
|
next_lines = list(rendered_screen.composed_lines)
|
|
line_count = len(next_lines)
|
|
|
|
grow_lines = 0
|
|
if not self.__gone_tall:
|
|
grow_lines = max(
|
|
min(line_count, height) - len(prev_composed),
|
|
0,
|
|
)
|
|
previous_lines.extend([EMPTY_RENDER_LINE] * grow_lines)
|
|
elif len(previous_lines) < line_count:
|
|
previous_lines.extend([EMPTY_RENDER_LINE] * (line_count - len(previous_lines)))
|
|
|
|
use_tall_mode = self.__gone_tall or line_count > height
|
|
|
|
# we make sure the cursor is on the screen, and that we're
|
|
# using all of the screen if we can
|
|
if cy < offset:
|
|
offset = cy
|
|
elif cy >= offset + height:
|
|
offset = cy - height + 1
|
|
elif offset > 0 and line_count < offset + height:
|
|
offset = max(line_count - height, 0)
|
|
next_lines.append(EMPTY_RENDER_LINE)
|
|
|
|
oldscr = previous_lines[old_offset : old_offset + height]
|
|
newscr = next_lines[offset : offset + height]
|
|
|
|
reverse_scroll = 0
|
|
forward_scroll = 0
|
|
if old_offset > offset and self._ri:
|
|
reverse_scroll = old_offset - offset
|
|
for _ in range(reverse_scroll):
|
|
if oldscr:
|
|
oldscr.pop(-1)
|
|
oldscr.insert(0, EMPTY_RENDER_LINE)
|
|
elif old_offset < offset and self._ind:
|
|
forward_scroll = offset - old_offset
|
|
for _ in range(forward_scroll):
|
|
if oldscr:
|
|
oldscr.pop(0)
|
|
oldscr.append(EMPTY_RENDER_LINE)
|
|
|
|
line_updates: list[LineUpdate] = []
|
|
px, _ = self.posxy
|
|
for y, oldline, newline in zip(range(offset, offset + height), oldscr, newscr):
|
|
update = self.__plan_changed_line(y, oldline, newline, px)
|
|
if update is not None:
|
|
line_updates.append(update)
|
|
|
|
cleared_lines = tuple(range(offset + len(newscr), offset + len(oldscr)))
|
|
console_rendered_screen = RenderedScreen(tuple(next_lines), c_xy)
|
|
trace(
|
|
"unix.refresh plan grow={grow} tall={tall} offset={offset} "
|
|
"reverse_scroll={reverse_scroll} forward_scroll={forward_scroll} "
|
|
"updates={updates} clears={clears}",
|
|
grow=grow_lines,
|
|
tall=use_tall_mode,
|
|
offset=offset,
|
|
reverse_scroll=reverse_scroll,
|
|
forward_scroll=forward_scroll,
|
|
updates=len(line_updates),
|
|
clears=len(cleared_lines),
|
|
)
|
|
return UnixRefreshPlan(
|
|
grow_lines=grow_lines,
|
|
use_tall_mode=use_tall_mode,
|
|
offset=offset,
|
|
reverse_scroll=reverse_scroll,
|
|
forward_scroll=forward_scroll,
|
|
line_updates=tuple(line_updates),
|
|
cleared_lines=cleared_lines,
|
|
rendered_screen=console_rendered_screen,
|
|
cursor=(cx, cy),
|
|
)
|
|
|
|
def __apply_refresh_plan(self, plan: UnixRefreshPlan) -> None:
|
|
cx, cy = plan.cursor
|
|
trace(
|
|
"unix.refresh apply cursor={cursor} updates={updates} clears={clears}",
|
|
cursor=plan.cursor,
|
|
updates=len(plan.line_updates),
|
|
clears=len(plan.cleared_lines),
|
|
)
|
|
visual_style = self.begin_redraw_visualization()
|
|
screen_line_count = len(self._rendered_screen.composed_lines)
|
|
|
|
for _ in range(plan.grow_lines):
|
|
self.__hide_cursor()
|
|
if screen_line_count:
|
|
self.__move(0, screen_line_count - 1)
|
|
self.__write("\n")
|
|
self.posxy = 0, screen_line_count
|
|
screen_line_count += 1
|
|
|
|
if plan.use_tall_mode and not self.__gone_tall:
|
|
self.__gone_tall = True
|
|
self.__move = self.__move_tall
|
|
|
|
old_offset = self.__offset
|
|
if plan.reverse_scroll:
|
|
self.__hide_cursor()
|
|
self.__write_code(self._cup, 0, 0)
|
|
self.posxy = 0, old_offset
|
|
for _ in range(plan.reverse_scroll):
|
|
self.__write_code(self._ri)
|
|
elif plan.forward_scroll:
|
|
self.__hide_cursor()
|
|
self.__write_code(self._cup, self.height - 1, 0)
|
|
self.posxy = 0, old_offset + self.height - 1
|
|
for _ in range(plan.forward_scroll):
|
|
self.__write_code(self._ind)
|
|
|
|
self.__offset = plan.offset
|
|
|
|
for update in plan.line_updates:
|
|
self.__apply_line_update(update, visual_style)
|
|
|
|
for y in plan.cleared_lines:
|
|
self.__hide_cursor()
|
|
self.__move(0, y)
|
|
self.posxy = 0, y
|
|
self.__write_code(self._el)
|
|
|
|
self.__show_cursor()
|
|
self.move_cursor(cx, cy)
|
|
self.flushoutput()
|
|
self.sync_rendered_screen(plan.rendered_screen, self.posxy)
|
|
|
|
def move_cursor(self, x: int, y: int) -> None:
|
|
"""
|
|
Move the cursor to the specified position on the screen.
|
|
|
|
Parameters:
|
|
- x (int): X coordinate.
|
|
- y (int): Y coordinate.
|
|
"""
|
|
if y < self.__offset or y >= self.__offset + self.height:
|
|
trace(
|
|
"unix.move_cursor offscreen x={x} y={y} offset={offset} height={height}",
|
|
x=x,
|
|
y=y,
|
|
offset=self.__offset,
|
|
height=self.height,
|
|
)
|
|
self.event_queue.insert(Event("scroll", ""))
|
|
else:
|
|
trace("unix.move_cursor x={x} y={y}", x=x, y=y)
|
|
self.__move(x, y)
|
|
self.posxy = x, y
|
|
self.flushoutput()
|
|
|
|
def prepare(self) -> None:
|
|
"""
|
|
Prepare the console for input/output operations.
|
|
"""
|
|
trace("unix.prepare")
|
|
self.__buffer = []
|
|
|
|
self.__svtermstate = tcgetattr(self.input_fd)
|
|
raw = self.__svtermstate.copy()
|
|
raw.iflag &= ~(termios.INPCK | termios.ISTRIP | termios.IXON)
|
|
raw.oflag &= ~(termios.OPOST)
|
|
raw.cflag &= ~(termios.CSIZE | termios.PARENB)
|
|
raw.cflag |= termios.CS8
|
|
raw.iflag |= termios.BRKINT
|
|
raw.lflag &= ~(termios.ICANON | termios.ECHO | termios.IEXTEN)
|
|
raw.lflag |= termios.ISIG
|
|
raw.cc[termios.VMIN] = b"\x01"
|
|
raw.cc[termios.VTIME] = b"\x00"
|
|
self.__input_fd_set(raw)
|
|
|
|
# Apple Terminal will re-wrap lines for us unless we preempt the
|
|
# damage.
|
|
if self.is_apple_terminal:
|
|
os.write(self.output_fd, b"\033[?7l")
|
|
|
|
self.height, self.width = self.getheightwidth()
|
|
|
|
self.posxy = 0, 0
|
|
self.__gone_tall = False
|
|
self.__move = self.__move_short
|
|
self.__offset = 0
|
|
self.sync_rendered_screen(RenderedScreen.empty(), self.posxy)
|
|
|
|
self.__maybe_write_code(self._smkx)
|
|
|
|
try:
|
|
self.old_sigwinch = signal.signal(signal.SIGWINCH, self.__sigwinch)
|
|
except ValueError:
|
|
pass
|
|
|
|
self.__enable_bracketed_paste()
|
|
|
|
def restore(self) -> None:
|
|
"""
|
|
Restore the console to the default state
|
|
"""
|
|
trace("unix.restore")
|
|
self.__disable_bracketed_paste()
|
|
self.__maybe_write_code(self._rmkx)
|
|
self.flushoutput()
|
|
self.__input_fd_set(self.__svtermstate)
|
|
|
|
if self.is_apple_terminal:
|
|
os.write(self.output_fd, b"\033[?7h")
|
|
|
|
if hasattr(self, "old_sigwinch"):
|
|
try:
|
|
signal.signal(signal.SIGWINCH, self.old_sigwinch)
|
|
except ValueError as e:
|
|
import threading
|
|
if threading.current_thread() is threading.main_thread():
|
|
raise e
|
|
del self.old_sigwinch
|
|
|
|
def push_char(self, char: int | bytes) -> None:
|
|
"""
|
|
Push a character to the console event queue.
|
|
"""
|
|
trace("push char {char!r}", char=char)
|
|
self.event_queue.push(char)
|
|
|
|
def get_event(self, block: bool = True) -> Event | None:
|
|
"""
|
|
Get an event from the console event queue.
|
|
|
|
Parameters:
|
|
- block (bool): Whether to block until an event is available.
|
|
|
|
Returns:
|
|
- Event: Event object from the event queue.
|
|
"""
|
|
if not block and not self.wait(timeout=0):
|
|
return None
|
|
|
|
while self.event_queue.empty():
|
|
while True:
|
|
try:
|
|
self.push_char(self.__read(1))
|
|
except OSError as err:
|
|
if err.errno == errno.EINTR:
|
|
if not self.event_queue.empty():
|
|
return self.event_queue.get()
|
|
else:
|
|
continue
|
|
elif err.errno == errno.EIO:
|
|
raise SystemExit(errno.EIO)
|
|
else:
|
|
raise
|
|
else:
|
|
break
|
|
return self.event_queue.get()
|
|
|
|
def wait(self, timeout: float | None = None) -> bool:
|
|
"""
|
|
Wait for events on the console.
|
|
"""
|
|
return (
|
|
not self.event_queue.empty()
|
|
or bool(self.pollob.poll(timeout))
|
|
)
|
|
|
|
def set_cursor_vis(self, visible: bool) -> None:
|
|
"""
|
|
Set the visibility of the cursor.
|
|
|
|
Parameters:
|
|
- visible (bool): Visibility flag.
|
|
"""
|
|
if visible:
|
|
self.__show_cursor()
|
|
else:
|
|
self.__hide_cursor()
|
|
|
|
if TIOCGWINSZ:
|
|
|
|
def getheightwidth(self):
|
|
"""
|
|
Get the height and width of the console.
|
|
|
|
Returns:
|
|
- tuple: Height and width of the console.
|
|
"""
|
|
try:
|
|
return int(os.environ["LINES"]), int(os.environ["COLUMNS"])
|
|
except (KeyError, TypeError, ValueError):
|
|
try:
|
|
size = ioctl(self.input_fd, TIOCGWINSZ, b"\000" * 8)
|
|
except OSError:
|
|
return 25, 80
|
|
height, width = struct.unpack("hhhh", size)[0:2]
|
|
if not height:
|
|
return 25, 80
|
|
return height, width
|
|
|
|
else:
|
|
|
|
def getheightwidth(self):
|
|
"""
|
|
Get the height and width of the console.
|
|
|
|
Returns:
|
|
- tuple: Height and width of the console.
|
|
"""
|
|
try:
|
|
return int(os.environ["LINES"]), int(os.environ["COLUMNS"])
|
|
except (KeyError, TypeError, ValueError):
|
|
return 25, 80
|
|
|
|
def forgetinput(self):
|
|
"""
|
|
Discard any pending input on the console.
|
|
"""
|
|
termios.tcflush(self.input_fd, termios.TCIFLUSH)
|
|
|
|
def flushoutput(self):
|
|
"""
|
|
Flush the output buffer.
|
|
"""
|
|
for text, iscode in self.__buffer:
|
|
if iscode:
|
|
self.__tputs(text)
|
|
else:
|
|
os.write(self.output_fd, text.encode(self.encoding, "replace"))
|
|
del self.__buffer[:]
|
|
|
|
def finish(self):
|
|
"""
|
|
Finish console operations and flush the output buffer.
|
|
"""
|
|
rendered_lines = self._rendered_screen.composed_lines
|
|
y = len(rendered_lines) - 1
|
|
while y >= 0 and not rendered_lines[y].text:
|
|
y -= 1
|
|
self.__move(0, min(y, self.height + self.__offset - 1))
|
|
self.__write("\n\r")
|
|
self.flushoutput()
|
|
|
|
def beep(self):
|
|
"""
|
|
Emit a beep sound.
|
|
"""
|
|
self.__maybe_write_code(self._bel)
|
|
self.flushoutput()
|
|
|
|
if FIONREAD:
|
|
|
|
def getpending(self):
|
|
"""
|
|
Get pending events from the console event queue.
|
|
|
|
Returns:
|
|
- Event: Pending event from the event queue.
|
|
"""
|
|
e = Event("key", "", b"")
|
|
|
|
while not self.event_queue.empty():
|
|
e2 = self.event_queue.get()
|
|
e.data += e2.data
|
|
e.raw += e2.raw
|
|
|
|
amount = struct.unpack("i", ioctl(self.input_fd, FIONREAD, b"\0\0\0\0"))[0]
|
|
trace("getpending({a})", a=amount)
|
|
raw = self.__read(amount)
|
|
data = str(raw, self.encoding, "replace")
|
|
e.data += data
|
|
e.raw += raw
|
|
return e
|
|
|
|
else:
|
|
|
|
def getpending(self):
|
|
"""
|
|
Get pending events from the console event queue.
|
|
|
|
Returns:
|
|
- Event: Pending event from the event queue.
|
|
"""
|
|
e = Event("key", "", b"")
|
|
|
|
while not self.event_queue.empty():
|
|
e2 = self.event_queue.get()
|
|
e.data += e2.data
|
|
e.raw += e2.raw
|
|
|
|
amount = 10000
|
|
raw = self.__read(amount)
|
|
data = str(raw, self.encoding, "replace")
|
|
e.data += data
|
|
e.raw += raw
|
|
return e
|
|
|
|
def clear(self):
|
|
"""
|
|
Clear the console screen.
|
|
"""
|
|
trace("unix.clear")
|
|
self.__write_code(self._clear)
|
|
self.__gone_tall = True
|
|
self.__move = self.__move_tall
|
|
self.posxy = 0, 0
|
|
self.sync_rendered_screen(RenderedScreen.empty(), self.posxy)
|
|
|
|
@property
|
|
def input_hook(self):
|
|
# avoid inline imports here so the repl doesn't get flooded
|
|
# with import logging from -X importtime=2
|
|
if posix is not None and posix._is_inputhook_installed():
|
|
return posix._inputhook
|
|
|
|
def __enable_bracketed_paste(self) -> None:
|
|
os.write(self.output_fd, b"\x1b[?2004h")
|
|
|
|
def __disable_bracketed_paste(self) -> None:
|
|
os.write(self.output_fd, b"\x1b[?2004l")
|
|
|
|
def __setup_movement(self):
|
|
"""
|
|
Set up the movement functions based on the terminal capabilities.
|
|
"""
|
|
if 0 and self._hpa: # hpa don't work in windows telnet :-(
|
|
self.__move_x = self.__move_x_hpa
|
|
elif self._cub and self._cuf:
|
|
self.__move_x = self.__move_x_cub_cuf
|
|
elif self._cub1 and self._cuf1:
|
|
self.__move_x = self.__move_x_cub1_cuf1
|
|
else:
|
|
raise RuntimeError("insufficient terminal (horizontal)")
|
|
|
|
if self._cuu and self._cud:
|
|
self.__move_y = self.__move_y_cuu_cud
|
|
elif self._cuu1 and self._cud1:
|
|
self.__move_y = self.__move_y_cuu1_cud1
|
|
else:
|
|
raise RuntimeError("insufficient terminal (vertical)")
|
|
|
|
if self._dch1:
|
|
self.dch1 = self._dch1
|
|
elif self._dch:
|
|
self.dch1 = terminfo.tparm(self._dch, 1)
|
|
else:
|
|
self.dch1 = None
|
|
|
|
if self._ich1:
|
|
self.ich1 = self._ich1
|
|
elif self._ich:
|
|
self.ich1 = terminfo.tparm(self._ich, 1)
|
|
else:
|
|
self.ich1 = None
|
|
|
|
self.__move = self.__move_short
|
|
|
|
@staticmethod
|
|
def __cell_index_from_x(line: RenderLine, x_coord: int) -> int:
|
|
width = 0
|
|
index = 0
|
|
while index < len(line.cells) and width < x_coord:
|
|
width += line.cells[index].width
|
|
index += 1
|
|
return index
|
|
|
|
def __plan_changed_line(
|
|
self,
|
|
y: int,
|
|
oldline: RenderLine,
|
|
newline: RenderLine,
|
|
px_coord: int,
|
|
) -> LineUpdate | None:
|
|
# NOTE: The shared replace_char / replace_span / rewrite_suffix logic
|
|
# is duplicated in WindowsConsole.__plan_changed_line. Keep changes to
|
|
# these common cases synchronised between the two files. Yes, this is
|
|
# duplicated on purpose; the two backends agree just enough to make a
|
|
# shared helper a trap. Unix-only cases (insert_char, delete_then_insert)
|
|
# rely on terminal capabilities (ich1/dch1) that are unavailable on
|
|
# Windows.
|
|
diff = diff_render_lines(oldline, newline)
|
|
if diff is None:
|
|
return None
|
|
|
|
start_cell = diff.start_cell
|
|
start_x = diff.start_x
|
|
|
|
if (
|
|
self.ich1
|
|
and not diff.old_cells
|
|
and (visible_new_cells := tuple(
|
|
cell for cell in diff.new_cells if cell.width
|
|
))
|
|
and len(visible_new_cells) == 1
|
|
and all(cell.width == 0 for cell in diff.new_cells[1:])
|
|
and oldline.cells[start_cell:] == newline.cells[start_cell + 1 :]
|
|
):
|
|
px_cell = self.__cell_index_from_x(oldline, px_coord)
|
|
if (
|
|
y == self.posxy[1]
|
|
and start_x > self.posxy[0]
|
|
and oldline.cells[px_cell:start_cell]
|
|
== newline.cells[px_cell + 1 : start_cell + 1]
|
|
):
|
|
start_cell = px_cell
|
|
start_x = px_coord
|
|
planned_cells = diff.new_cells
|
|
changed_cell = visible_new_cells[0]
|
|
return LineUpdate(
|
|
kind="insert_char",
|
|
y=y,
|
|
start_cell=start_cell,
|
|
start_x=start_x,
|
|
cells=planned_cells,
|
|
char_width=changed_cell.width,
|
|
reset_to_margin=requires_cursor_resync(planned_cells),
|
|
)
|
|
|
|
if (
|
|
len(diff.old_cells) == 1
|
|
and len(diff.new_cells) == 1
|
|
and diff.old_cells[0].width == diff.new_cells[0].width
|
|
):
|
|
planned_cells = diff.new_cells
|
|
changed_cell = planned_cells[0]
|
|
return LineUpdate(
|
|
kind="replace_char",
|
|
y=y,
|
|
start_cell=start_cell,
|
|
start_x=start_x,
|
|
cells=planned_cells,
|
|
char_width=changed_cell.width,
|
|
reset_to_margin=requires_cursor_resync(planned_cells),
|
|
)
|
|
|
|
if diff.old_changed_width == diff.new_changed_width:
|
|
planned_cells = diff.new_cells
|
|
return LineUpdate(
|
|
kind="replace_span",
|
|
y=y,
|
|
start_cell=start_cell,
|
|
start_x=start_x,
|
|
cells=planned_cells,
|
|
char_width=diff.new_changed_width,
|
|
reset_to_margin=requires_cursor_resync(planned_cells),
|
|
)
|
|
|
|
if (
|
|
self.dch1
|
|
and self.ich1
|
|
and newline.width == self.width
|
|
and start_x < newline.width - 2
|
|
and newline.cells[start_cell + 1 : -1] == oldline.cells[start_cell:-2]
|
|
):
|
|
planned_cells = (newline.cells[start_cell],)
|
|
changed_cell = planned_cells[0]
|
|
return LineUpdate(
|
|
kind="delete_then_insert",
|
|
y=y,
|
|
start_cell=start_cell,
|
|
start_x=start_x,
|
|
cells=planned_cells,
|
|
char_width=changed_cell.width,
|
|
reset_to_margin=requires_cursor_resync(planned_cells),
|
|
)
|
|
|
|
suffix_cells = newline.cells[start_cell:]
|
|
return LineUpdate(
|
|
kind="rewrite_suffix",
|
|
y=y,
|
|
start_cell=start_cell,
|
|
start_x=start_x,
|
|
cells=suffix_cells,
|
|
char_width=sum(cell.width for cell in suffix_cells),
|
|
clear_eol=oldline.width > newline.width,
|
|
reset_to_margin=requires_cursor_resync(suffix_cells),
|
|
)
|
|
|
|
def __apply_line_update(
|
|
self,
|
|
update: LineUpdate,
|
|
visual_style: str | None = None,
|
|
) -> None:
|
|
text = render_cells(update.cells, visual_style) if visual_style else update.text
|
|
trace(
|
|
"unix.refresh update kind={kind} y={y} x={x} text={text} "
|
|
"clear_eol={clear_eol} reset_to_margin={reset}",
|
|
kind=update.kind,
|
|
y=update.y,
|
|
x=update.start_x,
|
|
text=trace_text(text),
|
|
clear_eol=update.clear_eol,
|
|
reset=update.reset_to_margin,
|
|
)
|
|
if update.kind == "insert_char":
|
|
self.__move(update.start_x, update.y)
|
|
self.__write_code(self.ich1)
|
|
self.__write(text)
|
|
self.posxy = update.start_x + update.char_width, update.y
|
|
elif update.kind in {"replace_char", "replace_span"}:
|
|
self.__move(update.start_x, update.y)
|
|
self.__write(text)
|
|
self.posxy = update.start_x + update.char_width, update.y
|
|
elif update.kind == "delete_then_insert":
|
|
self.__hide_cursor()
|
|
self.__move(self.width - 2, update.y)
|
|
self.posxy = self.width - 2, update.y
|
|
self.__write_code(self.dch1)
|
|
self.__move(update.start_x, update.y)
|
|
self.__write_code(self.ich1)
|
|
self.__write(text)
|
|
self.posxy = update.start_x + update.char_width, update.y
|
|
else:
|
|
self.__hide_cursor()
|
|
self.__move(update.start_x, update.y)
|
|
if update.clear_eol:
|
|
self.__write_code(self._el)
|
|
self.__write(text)
|
|
self.posxy = update.start_x + update.char_width, update.y
|
|
|
|
if update.reset_to_margin:
|
|
# Non-SGR terminal controls can affect the cursor position.
|
|
self.move_cursor(0, update.y)
|
|
|
|
def __write(self, text):
|
|
self.__buffer.append((text, False))
|
|
|
|
def __write_code(self, fmt, *args):
|
|
self.__buffer.append((terminfo.tparm(fmt, *args), True))
|
|
|
|
def __maybe_write_code(self, fmt, *args):
|
|
if fmt:
|
|
self.__write_code(fmt, *args)
|
|
|
|
def __move_y_cuu1_cud1(self, y):
|
|
assert self._cud1 is not None
|
|
assert self._cuu1 is not None
|
|
dy = y - self.posxy[1]
|
|
if dy > 0:
|
|
self.__write_code(dy * self._cud1)
|
|
elif dy < 0:
|
|
self.__write_code((-dy) * self._cuu1)
|
|
|
|
def __move_y_cuu_cud(self, y):
|
|
dy = y - self.posxy[1]
|
|
if dy > 0:
|
|
self.__write_code(self._cud, dy)
|
|
elif dy < 0:
|
|
self.__write_code(self._cuu, -dy)
|
|
|
|
def __move_x_hpa(self, x: int) -> None:
|
|
if x != self.posxy[0]:
|
|
self.__write_code(self._hpa, x)
|
|
|
|
def __move_x_cub1_cuf1(self, x: int) -> None:
|
|
assert self._cuf1 is not None
|
|
assert self._cub1 is not None
|
|
dx = x - self.posxy[0]
|
|
if dx > 0:
|
|
self.__write_code(self._cuf1 * dx)
|
|
elif dx < 0:
|
|
self.__write_code(self._cub1 * (-dx))
|
|
|
|
def __move_x_cub_cuf(self, x: int) -> None:
|
|
dx = x - self.posxy[0]
|
|
if dx > 0:
|
|
self.__write_code(self._cuf, dx)
|
|
elif dx < 0:
|
|
self.__write_code(self._cub, -dx)
|
|
|
|
def __move_short(self, x, y):
|
|
self.__move_x(x)
|
|
self.__move_y(y)
|
|
|
|
def __move_tall(self, x, y):
|
|
assert 0 <= y - self.__offset < self.height, y - self.__offset
|
|
self.__write_code(self._cup, y - self.__offset, x)
|
|
|
|
def __sigwinch(self, signum, frame):
|
|
self.event_queue.insert(Event("resize", ""))
|
|
|
|
def __hide_cursor(self):
|
|
if self.cursor_visible:
|
|
self.__maybe_write_code(self._civis)
|
|
self.cursor_visible = False
|
|
|
|
def __show_cursor(self):
|
|
if not self.cursor_visible:
|
|
self.__maybe_write_code(self._cnorm)
|
|
self.cursor_visible = True
|
|
|
|
def repaint(self):
|
|
composed = self._rendered_screen.composed_lines
|
|
trace(
|
|
"unix.repaint gone_tall={gone_tall} screen_lines={lines} offset={offset}",
|
|
gone_tall=self.__gone_tall,
|
|
lines=len(composed),
|
|
offset=self.__offset,
|
|
)
|
|
if not self.__gone_tall:
|
|
self.posxy = 0, self.posxy[1]
|
|
self.__write("\r")
|
|
ns = len(composed) * ["\000" * self.width]
|
|
else:
|
|
self.posxy = 0, self.__offset
|
|
self.__move(0, self.__offset)
|
|
ns = self.height * ["\000" * self.width]
|
|
self.sync_rendered_screen(
|
|
RenderedScreen.from_screen_lines(ns, self.posxy),
|
|
self.posxy,
|
|
)
|
|
|
|
def __tputs(self, fmt, prog=delayprog):
|
|
"""A Python implementation of the curses tputs function; the
|
|
curses one can't really be wrapped in a sane manner.
|
|
|
|
I have the strong suspicion that this is complexity that
|
|
will never do anyone any good."""
|
|
# using .get() means that things will blow up
|
|
# only if the bps is actually needed (which I'm
|
|
# betting is pretty unlikely)
|
|
bps = ratedict.get(self.__svtermstate.ospeed)
|
|
while True:
|
|
m = prog.search(fmt)
|
|
if not m:
|
|
os.write(self.output_fd, fmt)
|
|
break
|
|
x, y = m.span()
|
|
os.write(self.output_fd, fmt[:x])
|
|
fmt = fmt[y:]
|
|
delay = int(m.group(1))
|
|
if b"*" in m.group(2):
|
|
delay *= self.height
|
|
if self._pad and bps is not None:
|
|
nchars = (bps * delay) / 1000
|
|
os.write(self.output_fd, self._pad * nchars)
|
|
else:
|
|
time.sleep(float(delay) / 1000.0)
|
|
|
|
def __input_fd_set(
|
|
self,
|
|
state: TermState,
|
|
ignore: AbstractSet[int] = _error_codes_to_ignore,
|
|
) -> bool:
|
|
try:
|
|
tcsetattr(self.input_fd, termios.TCSADRAIN, state)
|
|
except termios.error as te:
|
|
if te.args[0] not in ignore:
|
|
raise
|
|
return False
|
|
else:
|
|
return True
|