| """Pure Python curses-like terminal capability queries.""" |
| |
| from dataclasses import dataclass, field |
| import errno |
| import os |
| from pathlib import Path |
| import re |
| import struct |
| |
| |
| # Terminfo constants |
| MAGIC16 = 0o432 # Magic number for 16-bit terminfo format |
| MAGIC32 = 0o1036 # Magic number for 32-bit terminfo format |
| |
| # Special values for absent/cancelled capabilities |
| ABSENT_BOOLEAN = -1 |
| ABSENT_NUMERIC = -1 |
| CANCELLED_NUMERIC = -2 |
| ABSENT_STRING = None |
| CANCELLED_STRING = None |
| |
| |
| # Standard string capability names from ncurses Caps file |
| # This matches the order used by ncurses when compiling terminfo |
| # fmt: off |
| _STRING_NAMES: tuple[str, ...] = ( |
| "cbt", "bel", "cr", "csr", "tbc", "clear", "el", "ed", "hpa", "cmdch", |
| "cup", "cud1", "home", "civis", "cub1", "mrcup", "cnorm", "cuf1", "ll", |
| "cuu1", "cvvis", "dch1", "dl1", "dsl", "hd", "smacs", "blink", "bold", |
| "smcup", "smdc", "dim", "smir", "invis", "prot", "rev", "smso", "smul", |
| "ech", "rmacs", "sgr0", "rmcup", "rmdc", "rmir", "rmso", "rmul", "flash", |
| "ff", "fsl", "is1", "is2", "is3", "if", "ich1", "il1", "ip", "kbs", "ktbc", |
| "kclr", "kctab", "kdch1", "kdl1", "kcud1", "krmir", "kel", "ked", "kf0", |
| "kf1", "kf10", "kf2", "kf3", "kf4", "kf5", "kf6", "kf7", "kf8", "kf9", |
| "khome", "kich1", "kil1", "kcub1", "kll", "knp", "kpp", "kcuf1", "kind", |
| "kri", "khts", "kcuu1", "rmkx", "smkx", "lf0", "lf1", "lf10", "lf2", "lf3", |
| "lf4", "lf5", "lf6", "lf7", "lf8", "lf9", "rmm", "smm", "nel", "pad", "dch", |
| "dl", "cud", "ich", "indn", "il", "cub", "cuf", "rin", "cuu", "pfkey", |
| "pfloc", "pfx", "mc0", "mc4", "mc5", "rep", "rs1", "rs2", "rs3", "rf", "rc", |
| "vpa", "sc", "ind", "ri", "sgr", "hts", "wind", "ht", "tsl", "uc", "hu", |
| "iprog", "ka1", "ka3", "kb2", "kc1", "kc3", "mc5p", "rmp", "acsc", "pln", |
| "kcbt", "smxon", "rmxon", "smam", "rmam", "xonc", "xoffc", "enacs", "smln", |
| "rmln", "kbeg", "kcan", "kclo", "kcmd", "kcpy", "kcrt", "kend", "kent", |
| "kext", "kfnd", "khlp", "kmrk", "kmsg", "kmov", "knxt", "kopn", "kopt", |
| "kprv", "kprt", "krdo", "kref", "krfr", "krpl", "krst", "kres", "ksav", |
| "kspd", "kund", "kBEG", "kCAN", "kCMD", "kCPY", "kCRT", "kDC", "kDL", |
| "kslt", "kEND", "kEOL", "kEXT", "kFND", "kHLP", "kHOM", "kIC", "kLFT", |
| "kMSG", "kMOV", "kNXT", "kOPT", "kPRV", "kPRT", "kRDO", "kRPL", "kRIT", |
| "kRES", "kSAV", "kSPD", "kUND", "rfi", "kf11", "kf12", "kf13", "kf14", |
| "kf15", "kf16", "kf17", "kf18", "kf19", "kf20", "kf21", "kf22", "kf23", |
| "kf24", "kf25", "kf26", "kf27", "kf28", "kf29", "kf30", "kf31", "kf32", |
| "kf33", "kf34", "kf35", "kf36", "kf37", "kf38", "kf39", "kf40", "kf41", |
| "kf42", "kf43", "kf44", "kf45", "kf46", "kf47", "kf48", "kf49", "kf50", |
| "kf51", "kf52", "kf53", "kf54", "kf55", "kf56", "kf57", "kf58", "kf59", |
| "kf60", "kf61", "kf62", "kf63", "el1", "mgc", "smgl", "smgr", "fln", "sclk", |
| "dclk", "rmclk", "cwin", "wingo", "hup","dial", "qdial", "tone", "pulse", |
| "hook", "pause", "wait", "u0", "u1", "u2", "u3", "u4", "u5", "u6", "u7", |
| "u8", "u9", "op", "oc", "initc", "initp", "scp", "setf", "setb", "cpi", |
| "lpi", "chr", "cvr", "defc", "swidm", "sdrfq", "sitm", "slm", "smicm", |
| "snlq", "snrmq", "sshm", "ssubm", "ssupm", "sum", "rwidm", "ritm", "rlm", |
| "rmicm", "rshm", "rsubm", "rsupm", "rum", "mhpa", "mcud1", "mcub1", "mcuf1", |
| "mvpa", "mcuu1", "porder", "mcud", "mcub", "mcuf", "mcuu", "scs", "smgb", |
| "smgbp", "smglp", "smgrp", "smgt", "smgtp", "sbim", "scsd", "rbim", "rcsd", |
| "subcs", "supcs", "docr", "zerom", "csnm", "kmous", "minfo", "reqmp", |
| "getm", "setaf", "setab", "pfxl", "devt", "csin", "s0ds", "s1ds", "s2ds", |
| "s3ds", "smglr", "smgtb", "birep", "binel", "bicr", "colornm", "defbi", |
| "endbi", "setcolor", "slines", "dispc", "smpch", "rmpch", "smsc", "rmsc", |
| "pctrm", "scesc", "scesa", "ehhlm", "elhlm", "elohlm", "erhlm", "ethlm", |
| "evhlm", "sgr1", "slength", "OTi2", "OTrs", "OTnl", "OTbc", "OTko", "OTma", |
| "OTG2", "OTG3", "OTG1", "OTG4", "OTGR", "OTGL", "OTGU", "OTGD", "OTGH", |
| "OTGV", "OTGC","meml", "memu", "box1" |
| ) |
| # fmt: on |
| |
| |
| def _get_terminfo_dirs() -> list[Path]: |
| """Get list of directories to search for terminfo files. |
| |
| Based on ncurses behavior in: |
| - ncurses/tinfo/db_iterator.c:_nc_next_db() |
| - ncurses/tinfo/read_entry.c:_nc_read_entry() |
| """ |
| dirs = [] |
| |
| terminfo = os.environ.get("TERMINFO") |
| if terminfo: |
| dirs.append(terminfo) |
| |
| try: |
| home = Path.home() |
| dirs.append(str(home / ".terminfo")) |
| except RuntimeError: |
| pass |
| |
| # Check TERMINFO_DIRS |
| terminfo_dirs = os.environ.get("TERMINFO_DIRS", "") |
| if terminfo_dirs: |
| for d in terminfo_dirs.split(":"): |
| if d: |
| dirs.append(d) |
| |
| dirs.extend( |
| [ |
| "/etc/terminfo", |
| "/lib/terminfo", |
| "/usr/lib/terminfo", |
| "/usr/share/terminfo", |
| "/usr/share/lib/terminfo", |
| "/usr/share/misc/terminfo", |
| "/usr/local/lib/terminfo", |
| "/usr/local/share/terminfo", |
| ] |
| ) |
| |
| return [Path(d) for d in dirs if Path(d).is_dir()] |
| |
| |
| def _validate_terminal_name_or_raise(terminal_name: str) -> None: |
| if not isinstance(terminal_name, str): |
| raise TypeError("`terminal_name` must be a string") |
| |
| if not terminal_name: |
| raise ValueError("`terminal_name` cannot be empty") |
| |
| if "\x00" in terminal_name: |
| raise ValueError("NUL character found in `terminal_name`") |
| |
| t = Path(terminal_name) |
| if len(t.parts) > 1: |
| raise ValueError("`terminal_name` cannot contain path separators") |
| |
| |
| def _read_terminfo_file(terminal_name: str) -> bytes: |
| """Find and read terminfo file for given terminal name. |
| |
| Terminfo files are stored in directories using the first character |
| of the terminal name as a subdirectory. |
| """ |
| _validate_terminal_name_or_raise(terminal_name) |
| first_char = terminal_name[0].lower() |
| filename = terminal_name |
| |
| for directory in _get_terminfo_dirs(): |
| path = directory / first_char / filename |
| if path.is_file(): |
| return path.read_bytes() |
| |
| # Try with hex encoding of first char (for special chars) |
| hex_dir = "%02x" % ord(first_char) |
| path = directory / hex_dir / filename |
| if path.is_file(): |
| return path.read_bytes() |
| |
| raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), filename) |
| |
| |
| # Hard-coded terminal capabilities for common terminals |
| # This is a minimal subset needed by PyREPL |
| _TERMINAL_CAPABILITIES = { |
| # ANSI/xterm-compatible terminals |
| "ansi": { |
| # Bell |
| "bel": b"\x07", |
| # Cursor movement |
| "cub": b"\x1b[%p1%dD", # Move cursor left N columns |
| "cud": b"\x1b[%p1%dB", # Move cursor down N rows |
| "cuf": b"\x1b[%p1%dC", # Move cursor right N columns |
| "cuu": b"\x1b[%p1%dA", # Move cursor up N rows |
| "cub1": b"\x08", # Move cursor left 1 column |
| "cud1": b"\n", # Move cursor down 1 row |
| "cuf1": b"\x1b[C", # Move cursor right 1 column |
| "cuu1": b"\x1b[A", # Move cursor up 1 row |
| "cup": b"\x1b[%i%p1%d;%p2%dH", # Move cursor to row, column |
| "hpa": b"\x1b[%i%p1%dG", # Move cursor to column |
| # Clear operations |
| "clear": b"\x1b[H\x1b[2J", # Clear screen and home cursor |
| "el": b"\x1b[K", # Clear to end of line |
| # Insert/delete |
| "dch": b"\x1b[%p1%dP", # Delete N characters |
| "dch1": b"\x1b[P", # Delete 1 character |
| "ich": b"\x1b[%p1%d@", # Insert N characters |
| "ich1": b"", # Insert 1 character |
| # Cursor visibility |
| "civis": b"\x1b[?25l", # Make cursor invisible |
| "cnorm": b"\x1b[?12l\x1b[?25h", # Make cursor normal (visible) |
| # Scrolling |
| "ind": b"\n", # Scroll up one line |
| "ri": b"\x1bM", # Scroll down one line |
| # Keypad mode |
| "smkx": b"\x1b[?1h\x1b=", # Enable keypad mode |
| "rmkx": b"\x1b[?1l\x1b>", # Disable keypad mode |
| # Padding (not used in modern terminals) |
| "pad": b"", |
| # Function keys and special keys |
| "kdch1": b"\x1b[3~", # Delete key |
| "kcud1": b"\x1bOB", # Down arrow |
| "kend": b"\x1bOF", # End key |
| "kent": b"\x1bOM", # Enter key |
| "khome": b"\x1bOH", # Home key |
| "kich1": b"\x1b[2~", # Insert key |
| "kcub1": b"\x1bOD", # Left arrow |
| "knp": b"\x1b[6~", # Page down |
| "kpp": b"\x1b[5~", # Page up |
| "kcuf1": b"\x1bOC", # Right arrow |
| "kcuu1": b"\x1bOA", # Up arrow |
| # Function keys F1-F20 |
| "kf1": b"\x1bOP", |
| "kf2": b"\x1bOQ", |
| "kf3": b"\x1bOR", |
| "kf4": b"\x1bOS", |
| "kf5": b"\x1b[15~", |
| "kf6": b"\x1b[17~", |
| "kf7": b"\x1b[18~", |
| "kf8": b"\x1b[19~", |
| "kf9": b"\x1b[20~", |
| "kf10": b"\x1b[21~", |
| "kf11": b"\x1b[23~", |
| "kf12": b"\x1b[24~", |
| "kf13": b"\x1b[1;2P", |
| "kf14": b"\x1b[1;2Q", |
| "kf15": b"\x1b[1;2R", |
| "kf16": b"\x1b[1;2S", |
| "kf17": b"\x1b[15;2~", |
| "kf18": b"\x1b[17;2~", |
| "kf19": b"\x1b[18;2~", |
| "kf20": b"\x1b[19;2~", |
| }, |
| # Dumb terminal - minimal capabilities |
| "dumb": { |
| "bel": b"\x07", # Bell |
| "cud1": b"\n", # Move down 1 row (newline) |
| "ind": b"\n", # Scroll up one line (newline) |
| }, |
| # Linux console |
| "linux": { |
| # Bell |
| "bel": b"\x07", |
| # Cursor movement |
| "cub": b"\x1b[%p1%dD", # Move cursor left N columns |
| "cud": b"\x1b[%p1%dB", # Move cursor down N rows |
| "cuf": b"\x1b[%p1%dC", # Move cursor right N columns |
| "cuu": b"\x1b[%p1%dA", # Move cursor up N rows |
| "cub1": b"\x08", # Move cursor left 1 column (backspace) |
| "cud1": b"\n", # Move cursor down 1 row (newline) |
| "cuf1": b"\x1b[C", # Move cursor right 1 column |
| "cuu1": b"\x1b[A", # Move cursor up 1 row |
| "cup": b"\x1b[%i%p1%d;%p2%dH", # Move cursor to row, column |
| "hpa": b"\x1b[%i%p1%dG", # Move cursor to column |
| # Clear operations |
| "clear": b"\x1b[H\x1b[J", # Clear screen and home cursor (different from ansi!) |
| "el": b"\x1b[K", # Clear to end of line |
| # Insert/delete |
| "dch": b"\x1b[%p1%dP", # Delete N characters |
| "dch1": b"\x1b[P", # Delete 1 character |
| "ich": b"\x1b[%p1%d@", # Insert N characters |
| "ich1": b"\x1b[@", # Insert 1 character |
| # Cursor visibility |
| "civis": b"\x1b[?25l\x1b[?1c", # Make cursor invisible |
| "cnorm": b"\x1b[?25h\x1b[?0c", # Make cursor normal |
| # Scrolling |
| "ind": b"\n", # Scroll up one line |
| "ri": b"\x1bM", # Scroll down one line |
| # Keypad mode |
| "smkx": b"\x1b[?1h\x1b=", # Enable keypad mode |
| "rmkx": b"\x1b[?1l\x1b>", # Disable keypad mode |
| # Function keys and special keys |
| "kdch1": b"\x1b[3~", # Delete key |
| "kcud1": b"\x1b[B", # Down arrow |
| "kend": b"\x1b[4~", # End key (different from ansi!) |
| "khome": b"\x1b[1~", # Home key (different from ansi!) |
| "kich1": b"\x1b[2~", # Insert key |
| "kcub1": b"\x1b[D", # Left arrow |
| "knp": b"\x1b[6~", # Page down |
| "kpp": b"\x1b[5~", # Page up |
| "kcuf1": b"\x1b[C", # Right arrow |
| "kcuu1": b"\x1b[A", # Up arrow |
| # Function keys |
| "kf1": b"\x1b[[A", |
| "kf2": b"\x1b[[B", |
| "kf3": b"\x1b[[C", |
| "kf4": b"\x1b[[D", |
| "kf5": b"\x1b[[E", |
| "kf6": b"\x1b[17~", |
| "kf7": b"\x1b[18~", |
| "kf8": b"\x1b[19~", |
| "kf9": b"\x1b[20~", |
| "kf10": b"\x1b[21~", |
| "kf11": b"\x1b[23~", |
| "kf12": b"\x1b[24~", |
| "kf13": b"\x1b[25~", |
| "kf14": b"\x1b[26~", |
| "kf15": b"\x1b[28~", |
| "kf16": b"\x1b[29~", |
| "kf17": b"\x1b[31~", |
| "kf18": b"\x1b[32~", |
| "kf19": b"\x1b[33~", |
| "kf20": b"\x1b[34~", |
| }, |
| } |
| |
| # Map common TERM values to capability sets |
| _TERM_ALIASES = { |
| "xterm": "ansi", |
| "xterm-color": "ansi", |
| "xterm-256color": "ansi", |
| "screen": "ansi", |
| "screen-256color": "ansi", |
| "tmux": "ansi", |
| "tmux-256color": "ansi", |
| "vt100": "ansi", |
| "vt220": "ansi", |
| "rxvt": "ansi", |
| "rxvt-unicode": "ansi", |
| "rxvt-unicode-256color": "ansi", |
| "unknown": "dumb", |
| } |
| |
| |
| @dataclass |
| class TermInfo: |
| terminal_name: str | bytes | None |
| fallback: bool = True |
| |
| _capabilities: dict[str, bytes] = field(default_factory=dict) |
| |
| def __post_init__(self) -> None: |
| """Initialize terminal capabilities for the given terminal type. |
| |
| Based on ncurses implementation in: |
| - ncurses/tinfo/lib_setup.c:setupterm() and _nc_setupterm() |
| - ncurses/tinfo/lib_setup.c:TINFO_SETUP_TERM() |
| |
| This version first attempts to read terminfo database files like ncurses, |
| then, if `fallback` is True, falls back to hardcoded capabilities for |
| common terminal types. |
| """ |
| # If termstr is None or empty, try to get from environment |
| if not self.terminal_name: |
| self.terminal_name = os.environ.get("TERM") or "ANSI" |
| |
| if isinstance(self.terminal_name, bytes): |
| self.terminal_name = self.terminal_name.decode("ascii") |
| |
| try: |
| self._parse_terminfo_file(self.terminal_name) |
| except (OSError, ValueError): |
| if not self.fallback: |
| raise |
| |
| term_type = _TERM_ALIASES.get( |
| self.terminal_name, self.terminal_name |
| ) |
| if term_type not in _TERMINAL_CAPABILITIES: |
| term_type = "dumb" |
| self._capabilities = _TERMINAL_CAPABILITIES[term_type].copy() |
| |
| def _parse_terminfo_file(self, terminal_name: str) -> None: |
| """Parse a terminfo file. |
| |
| Populate the _capabilities dict for easy retrieval |
| |
| Based on ncurses implementation in: |
| - ncurses/tinfo/read_entry.c:_nc_read_termtype() |
| - ncurses/tinfo/read_entry.c:_nc_read_file_entry() |
| - ncurses/tinfo/lib_ti.c:tigetstr() |
| """ |
| data = _read_terminfo_file(terminal_name) |
| too_short = f"TermInfo file for {terminal_name!r} too short" |
| offset = 12 |
| if len(data) < offset: |
| raise ValueError(too_short) |
| |
| magic, name_size, bool_count, num_count, str_count, str_size = ( |
| struct.unpack("<Hhhhhh", data[:offset]) |
| ) |
| |
| if magic == MAGIC16: |
| number_size = 2 |
| elif magic == MAGIC32: |
| number_size = 4 |
| else: |
| raise ValueError( |
| f"TermInfo file for {terminal_name!r} uses unknown magic" |
| ) |
| |
| # Skip data than PyREPL doesn't need: |
| # - names (`|`-separated ASCII strings) |
| # - boolean capabilities (bytes with value 0 or 1) |
| # - numbers (little-endian integers, `number_size` bytes each) |
| offset += name_size |
| offset += bool_count |
| if offset % 2: |
| # Align to even byte boundary for numbers |
| offset += 1 |
| offset += num_count * number_size |
| if offset > len(data): |
| raise ValueError(too_short) |
| |
| # Read string offsets |
| end_offset = offset + 2 * str_count |
| if offset > len(data): |
| raise ValueError(too_short) |
| string_offset_data = data[offset:end_offset] |
| string_offsets = [ |
| off for [off] in struct.iter_unpack("<h", string_offset_data) |
| ] |
| offset = end_offset |
| |
| # Read string table |
| if offset + str_size > len(data): |
| raise ValueError(too_short) |
| string_table = data[offset : offset + str_size] |
| |
| # Extract strings from string table |
| capabilities = {} |
| for cap, off in zip(_STRING_NAMES, string_offsets): |
| if off < 0: |
| # CANCELLED_STRING; we do not store those |
| continue |
| elif off < len(string_table): |
| # Find null terminator |
| end = string_table.find(0, off) |
| if end >= 0: |
| capabilities[cap] = string_table[off:end] |
| # in other cases this is ABSENT_STRING; we don't store those. |
| |
| # Note: we don't support extended capabilities since PyREPL doesn't |
| # need them. |
| |
| self._capabilities = capabilities |
| |
| def get(self, cap: str) -> bytes | None: |
| """Get terminal capability string by name. |
| """ |
| if not isinstance(cap, str): |
| raise TypeError(f"`cap` must be a string, not {type(cap)}") |
| |
| return self._capabilities.get(cap) |
| |
| |
| def tparm(cap_bytes: bytes, *params: int) -> bytes: |
| """Parameterize a terminal capability string. |
| |
| Based on ncurses implementation in: |
| - ncurses/tinfo/lib_tparm.c:tparm() |
| - ncurses/tinfo/lib_tparm.c:tparam_internal() |
| |
| The ncurses version implements a full stack-based interpreter for |
| terminfo parameter strings. This pure Python version implements only |
| the subset of parameter substitution operations needed by PyREPL: |
| - %i (increment parameters for 1-based indexing) |
| - %p[1-9]%d (parameter substitution) |
| - %p[1-9]%{n}%+%d (parameter plus constant) |
| """ |
| if not isinstance(cap_bytes, bytes): |
| raise TypeError(f"`cap` must be bytes, not {type(cap_bytes)}") |
| |
| result = cap_bytes |
| |
| # %i - increment parameters (1-based instead of 0-based) |
| increment = b"%i" in result |
| if increment: |
| result = result.replace(b"%i", b"") |
| |
| # Replace %p1%d, %p2%d, etc. with actual parameter values |
| for i in range(len(params)): |
| pattern = b"%%p%d%%d" % (i + 1) |
| if pattern in result: |
| value = params[i] |
| if increment: |
| value += 1 |
| result = result.replace(pattern, str(value).encode("ascii")) |
| |
| # Handle %p1%{1}%+%d (parameter plus constant) |
| # Used in some cursor positioning sequences |
| pattern_re = re.compile(rb"%p(\d)%\{(\d+)\}%\+%d") |
| matches = list(pattern_re.finditer(result)) |
| for match in reversed(matches): # reversed to maintain positions |
| param_idx = int(match.group(1)) |
| constant = int(match.group(2)) |
| value = params[param_idx] + constant |
| result = ( |
| result[: match.start()] |
| + str(value).encode("ascii") |
| + result[match.end() :] |
| ) |
| |
| return result |
