SimpleFarm/farm/keys.py

"""跨平台非阻塞键盘读取。

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""