"""跨平台非阻塞键盘读取。 Windows 用 ``msvcrt``,POSIX 用 ``termios`` + ``select``, 两端统一返回**同一套小写**的按键名,TUI 层无需关心平台差异。 按键名约定(重要) ------------------ 具名键一律**小写**,见 :data:`NAMED_KEYS`:``up`` / ``down`` / ``left`` / ``right`` / ``space`` / ``tab`` / ``esc`` / ``enter`` / ``backspace`` / ``home`` / ``end`` / ``delete`` / ``ctrl_c``;普通可打印字符返回其自身的 小写形式(例如 ``"p"``)。 这条约定是硬性的:两条平台路径必须返回**完全一样**的名字。历史上这里 Windows 返回大写(``"UP"``)、TUI 里写的是小写(``"up"``),于是方向键、 空格、Tab、ESC 全都静默失效,而且只在某个平台上才看得出来。 在非交互式环境(管道、重定向)下 ``available`` 为 False, TUI 会退化为无输入的只读模式,并把 :attr:`KeyReader.reason` 显示给玩家。 """ from __future__ import annotations import importlib.util import os import sys import time IS_WINDOWS = os.name == "nt" # ---------------------------------------------------------------------- 具名键 # 两端共用的规范化名字。全小写,避免"大写产出、小写消费"这类靠肉眼发现的错配。 CTRL_C = "ctrl_c" ENTER = "enter" TAB = "tab" BACKSPACE = "backspace" ESC = "esc" SPACE = "space" UP = "up" DOWN = "down" LEFT = "left" RIGHT = "right" HOME = "home" END = "end" DELETE = "delete" NAMED_KEYS = frozenset( { CTRL_C, ENTER, TAB, BACKSPACE, ESC, SPACE, UP, DOWN, LEFT, RIGHT, HOME, END, DELETE, } ) # Windows 扩展键:第一次 getwch() 拿到 \x00 / \xe0,第二次才是真正的键码 _WIN_EXTENDED = { "H": UP, "P": DOWN, "K": LEFT, "M": RIGHT, "G": HOME, "O": END, "S": DELETE, } # Windows 控制字符 _WIN_CTRL = { "\x03": CTRL_C, "\r": ENTER, "\n": ENTER, "\t": TAB, "\x08": BACKSPACE, "\x7f": BACKSPACE, "\x1b": ESC, " ": SPACE, } # POSIX 控制字符(ESC 单独处理,因为方向键等功能键也是它打头的) _POSIX_CTRL = { "\x03": CTRL_C, "\r": ENTER, "\n": ENTER, "\t": TAB, "\x08": BACKSPACE, "\x7f": BACKSPACE, " ": SPACE, } # POSIX 转义序列的"参数+终结符"(去掉开头的 ESC [ 或 ESC O): # 方向键是 ESC [ A~D,Home/End 有两套写法,Delete 是 ESC [ 3 ~ _POSIX_CSI = { "A": UP, "B": DOWN, "C": RIGHT, "D": LEFT, "H": HOME, "F": END, "1~": HOME, "3~": DELETE, "4~": END, } EXTENDED_WAIT = 0.02 # 等功能键第二个字节的最长时间(秒) PENDING_WAIT = 0.02 # 判断"单按 ESC"还是"序列开头"、以及等半个字符的时间 class KeyReader: """把终端的原始字节流转换成按键名。""" def __init__(self) -> None: self._fd: int | None = None self._saved_term = None self._raw = b"" # POSIX:上一次没解析完的字节(半个转义序列、半个汉字) self.reason = "" # 读不到键盘时的解释,用于提示玩家 self.available = False # 先把 stdin 取到局部变量:下面每处都要用它。同时它的类型被收窄到非 None, # 省掉反复 `sys.stdin.xxx` 时"可能是 None"的隐患。 stdin = sys.stdin if stdin is None or not stdin.isatty(): self.reason = "标准输入不是终端(管道或重定向)" return self.available = True if IS_WINDOWS: # Windows 靠 msvcrt 做非阻塞读;拿不到就认为终端不可用 if importlib.util.find_spec("msvcrt") is None: self.available = False self.reason = "找不到 msvcrt 模块" return try: import msvcrt # 探一次:不是真的 Win32 控制台(例如 Git Bash 的 mintty)这里会报错。 # 与其让玩家对着"按键没反应"发呆,不如把原因摆出来。 msvcrt.kbhit() except Exception: self.available = False self.reason = "当前终端不是 Win32 控制台,msvcrt 读不到按键" return try: import termios import tty # 用局部变量:self._fd 声明为 int | None,直接传进去会被判为可能为 None fd = stdin.fileno() self._fd = fd self._saved_term = termios.tcgetattr(fd) # cbreak: 立即返回按键,但保留 Ctrl-C 等信号 tty.setcbreak(fd) except Exception: self.available = False self._saved_term = None self.reason = "无法把终端切到 cbreak 模式" # ------------------------------------------------------------------ 读取 def poll(self) -> str | None: """非阻塞读取一个按键;没有按键时返回 None。""" if not self.available: return None return self._poll_windows() if IS_WINDOWS else self._poll_posix() # ---------------------------------------------------------------- Windows def _poll_windows(self) -> str | None: import msvcrt if not msvcrt.kbhit(): return None char = msvcrt.getwch() if char in ("\x00", "\xe0"): code = self._win_extended_code() # 认不出来的功能键就丢掉,不要退化成 ESC 之类的错误动作 return _WIN_EXTENDED.get(code) if code else None named = _WIN_CTRL.get(char) return named if named else char.lower() @staticmethod def _win_extended_code() -> str | None: """取扩展键的第二个字节;它通常紧跟其后,但偶尔会差几毫秒。""" import msvcrt deadline = time.monotonic() + EXTENDED_WAIT while True: if msvcrt.kbhit(): return msvcrt.getwch().upper() if time.monotonic() >= deadline: return None time.sleep(0.005) # ---------------------------------------------------------------- POSIX 侧 def _poll_posix(self) -> str | None: import select if self._fd is None: return None if not self._raw: try: ready, _, _ = select.select([self._fd], [], [], 0) except (OSError, ValueError): return None if not ready: return None if self._read_fd() is None: return None return self._parse_posix() def _read_fd(self) -> bytes | None: """把输入追加进 :attr:`_raw`。读不到返回 None。""" if self._fd is None: return None try: data = os.read(self._fd, 64) except OSError: return None if not data: return None # 一次读到的可能是好几个按键:全部留在缓冲区里逐个返回, # 不能像老实现那样只取第一个字符、把其余的悄悄丢掉。 self._raw += data return data def _read_ready(self, timeout: float) -> bool: """等一小会儿再读一次,用于判断不完整的转义序列与半个字符。""" import select if self._fd is None: return False try: ready, _, _ = select.select([self._fd], [], [], timeout) except (OSError, ValueError): return False return bool(ready) and self._read_fd() is not None def _parse_posix(self) -> str | None: raw = self._raw if not raw: return None if raw[0:1] == b"\x1b": seq = self._read_escape_sequence() if seq is None: return None # 还没读全,等下一轮 poll if seq == b"": return ESC # 就是单按 ESC return _POSIX_CSI.get(seq.decode("latin-1")) first = raw[0] if first < 0x80: self._raw = raw[1:] char = chr(first) named = _POSIX_CTRL.get(char) return named if named else char.lower() # 多字节 UTF-8:先凑齐整个字符再解码,免得半个汉字被解成乱码 length = 2 if first < 0xE0 else 3 if first < 0xF0 else 4 if len(raw) < length: self._read_ready(PENDING_WAIT) raw = self._raw if len(raw) < length: return None # 等下一轮 poll 再来 char = raw[:length].decode("utf-8", errors="replace") self._raw = raw[length:] return char.lower() def _read_escape_sequence(self) -> bytes | None: """解析 ESC 打头的序列。 单按 ESC 返回 ``b""``,还没读全返回 ``None``,否则返回去掉 ``ESC [`` / ``ESC O`` 之后的"参数+终结符"(例如 ``b"A"``、``b"3~"``)。 """ # 最多试两次:第一次先看看能不能直接判完,不行就等一小会儿再试。 for attempt in (0, 1): raw = self._raw if len(raw) == 1: # 单按 ESC 与序列的第一个字节长得一样,有歧义 if attempt: self._raw = b"" return b"" self._read_ready(PENDING_WAIT) continue if raw[1:2] not in (b"[", b"O"): # 不是标准转义序列:吃掉 ESC,剩下的字节留给下一轮 self._raw = raw[1:] return b"" end = self._csi_end(raw) if end is None: if attempt: self._raw = b"" return b"" # 等不到了,当单按 ESC 处理,别把按键吞掉 self._read_ready(PENDING_WAIT) continue self._raw = raw[end + 1 :] return raw[2 : end + 1] return None @staticmethod def _csi_end(raw: bytes) -> int | None: """CSI 序列的终结符下标(0x40~0x7E);还没读到就返回 None。""" index = 2 while index < len(raw): if 0x40 <= raw[index] <= 0x7E: return index index += 1 return None # ------------------------------------------------------------------ 清理 def close(self) -> None: if self._fd is not None and self._saved_term is not None: try: import termios termios.tcsetattr(self._fd, termios.TCSADRAIN, self._saved_term) except Exception: pass self._fd = None self._saved_term = None self._raw = b""