pyright(basic 模式)从 15 errors / 10 warnings 降到 0/0: - pyproject: pythonPlatform="All"(termios/msvcrt 双平台分支不再误报), reportUnsupportedDunderAll=false(__init__ 惰性导出属正常写法) - keys.py: 捕获 sys.stdin 局部变量 + 提前返回,消除 fileno() 的 None 收窄告警 - server/app.py: lifespan 改用 @asynccontextmanager 并标注 AsyncIterator[None] - server/actions.py: 显式标注 result: dict[str, Any],消除 isinstance 冗余报错 - server/__main__.py: 用 importlib.util.find_spec 探测 websockets - tests/test_server.py: 同上,uvicorn 探测改为 find_spec - tools/preview_tui.py: TYPE_CHECKING 下引入 RemoteFarm,修正返回类型标注 - farm/tui.py: 取消未使用变量,统一 round/取整写法 其他: - README: 新增第 11 节「工程化:uv + ruff + pyright」,快速开始全部改用 uv sync / uv run;原第 11 节起顺延为 12 - requirements.txt: 由 uv export --no-dev --no-hashes 重新生成(锁定版本) - 清理临时日志文件 验收:ruff check 全通过 / ruff format 26 文件无改动 / pyright 0 错误 / unittest 97 项全部通过
151 lines
4.2 KiB
Python
151 lines
4.2 KiB
Python
"""跨平台非阻塞键盘读取。
|
||
|
||
Windows 用 ``msvcrt``,POSIX 用 ``termios`` + ``select``,
|
||
两端统一返回规范化的按键名,TUI 层无需关心平台差异。
|
||
|
||
在非交互式环境(管道、重定向)下 ``available`` 为 False,
|
||
TUI 会退化为无输入的只读模式。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import os
|
||
import sys
|
||
|
||
IS_WINDOWS = os.name == "nt"
|
||
|
||
# Windows 下的扩展键(以 \x00 / \xe0 开头)
|
||
_WIN_EXTENDED = {
|
||
"H": "UP",
|
||
"P": "DOWN",
|
||
"K": "LEFT",
|
||
"M": "RIGHT",
|
||
"G": "HOME",
|
||
"O": "END",
|
||
"S": "DELETE",
|
||
}
|
||
|
||
_WIN_CTRL = {
|
||
"\x03": "CTRL_C",
|
||
"\r": "ENTER",
|
||
"\n": "ENTER",
|
||
"\t": "TAB",
|
||
"\x08": "BACKSPACE",
|
||
"\x1b": "ESC",
|
||
" ": "SPACE",
|
||
}
|
||
|
||
|
||
class KeyReader:
|
||
"""把终端的原始字节流转换成按键名。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._fd: int | None = None
|
||
self._saved_term = None
|
||
|
||
# 先把 stdin 取到局部变量:下面每处都要用它。同时它的类型被收窄到非 None,
|
||
# 省掉反复 `sys.stdin.xxx` 时"可能是 None"的隐患。
|
||
stdin = sys.stdin
|
||
if stdin is None or not stdin.isatty():
|
||
self.available = False
|
||
return
|
||
self.available = True
|
||
|
||
if IS_WINDOWS:
|
||
# Windows 靠 msvcrt 做非阻塞读;拿不到就认为终端不可用
|
||
if importlib.util.find_spec("msvcrt") is None:
|
||
self.available = False
|
||
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
|
||
|
||
# ------------------------------------------------------------------ 读取
|
||
def poll(self) -> str | None:
|
||
"""非阻塞读取一个按键;没有按键时返回 None。"""
|
||
if not self.available:
|
||
return None
|
||
return self._poll_windows() if IS_WINDOWS else self._poll_posix()
|
||
|
||
def _poll_windows(self) -> str | None:
|
||
import msvcrt
|
||
|
||
if not msvcrt.kbhit():
|
||
return None
|
||
char = msvcrt.getwch()
|
||
|
||
if char in ("\x00", "\xe0"):
|
||
if not msvcrt.kbhit():
|
||
return None
|
||
return _WIN_EXTENDED.get(msvcrt.getwch().upper())
|
||
if char in _WIN_CTRL:
|
||
return _WIN_CTRL[char]
|
||
return char.lower()
|
||
|
||
def _poll_posix(self) -> str | None:
|
||
import select
|
||
|
||
if self._fd is None:
|
||
return None
|
||
try:
|
||
ready, _, _ = select.select([self._fd], [], [], 0)
|
||
except (OSError, ValueError):
|
||
return None
|
||
if not ready:
|
||
return None
|
||
try:
|
||
data = os.read(self._fd, 8)
|
||
except OSError:
|
||
return None
|
||
if not data:
|
||
return None
|
||
|
||
text = data.decode("utf-8", errors="ignore")
|
||
if not text:
|
||
return None
|
||
|
||
# 方向键:ESC [ A/B/C/D,或 ESC O A/B/C/D
|
||
if text.startswith("\x1b["):
|
||
return {"A": "UP", "B": "DOWN", "C": "RIGHT", "D": "LEFT"}.get(text[2:3])
|
||
if text.startswith("\x1bO"):
|
||
return {"A": "UP", "B": "DOWN", "C": "RIGHT", "D": "LEFT"}.get(text[2:3])
|
||
|
||
char = text[0]
|
||
mapping = {
|
||
"\x03": "CTRL_C",
|
||
"\r": "ENTER",
|
||
"\n": "ENTER",
|
||
"\t": "TAB",
|
||
"\x7f": "BACKSPACE",
|
||
"\x08": "BACKSPACE",
|
||
"\x1b": "ESC",
|
||
" ": "SPACE",
|
||
}
|
||
if char in mapping:
|
||
return mapping[char]
|
||
return char.lower()
|
||
|
||
# ------------------------------------------------------------------ 清理
|
||
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
|