from __future__ import annotations import argparse from collections import deque from dataclasses import asdict, dataclass from datetime import datetime from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import queue import secrets import threading import time from urllib.parse import parse_qs, urlparse import webbrowser from vitapad.backends import create_backend from vitapad.protocol import InputState from vitapad.receiver import Receiver @dataclass(frozen=True, slots=True) class LogEntry: id: int time: str level: str message: str class DashboardController: def __init__( self, backend_name: str, bind: str, port: int, discovery_port: int, allow: str | None, timeout_ms: int, ) -> None: self.backend_name = backend_name self._resolved_backend = backend_name self.bind = bind self.port = port self.discovery_port = discovery_port self.allow = allow self.timeout_ms = timeout_ms self._lock = threading.RLock() self._status = "paused" self._receiver: Receiver | None = None self._worker: threading.Thread | None = None self._started_at: float | None = None self._error: str | None = None self._stop_requested = False self._logs: deque[LogEntry] = deque(maxlen=500) self._next_log_id = 1 self._input = InputState.neutral() self._subscribers: set[queue.Queue[InputState]] = set() self.log("控制面板已启动，点击“开启使用”开始接收。") def log(self, message: str, level: str = "info") -> None: with self._lock: entry = LogEntry( id=self._next_log_id, time=datetime.now().strftime("%H:%M:%S"), level=level, message=message, ) self._next_log_id += 1 self._logs.append(entry) def start(self) -> tuple[bool, str]: with self._lock: if self._status in {"starting", "running"}: return False, "接收服务已经开启" if self._status == "stopping": return False, "服务正在暂停，请稍候" self._status = "starting" self._error = None self._stop_requested = False self.log("正在创建虚拟手柄并启动网络服务...") self._worker = threading.Thread( target=self._run_receiver, name="receiver", daemon=True ) self._worker.start() return True, "正在开启" def _run_receiver(self) -> None: try: backend = create_backend(self.backend_name) receiver = Receiver( backend=backend, bind=self.bind, port=self.port, discovery_port=self.discovery_port, allow=self.allow, timeout_ms=self.timeout_ms, log=self.log, on_input=self.handle_input, ) with self._lock: if self._stop_requested: backend.close() self.log("服务已暂停。") return self._resolved_backend = backend.name self._receiver = receiver self._status = "running" self._started_at = time.monotonic() self.log(f"服务已开启，虚拟手柄后端：{backend.name}", "success") receiver.run() except Exception as exc: with self._lock: self._error = str(exc) self.log(f"启动失败：{exc}", "error") finally: with self._lock: if self._stop_requested: self.log("服务已暂停，虚拟手柄已释放。", "success") self._receiver = None self._started_at = None self._status = "paused" def pause(self) -> tuple[bool, str]: with self._lock: if self._status == "paused": return False, "接收服务已经暂停" if self._status == "stopping": return False, "服务正在暂停" self._status = "stopping" self._stop_requested = True receiver = self._receiver self.log("正在暂停，释放全部手柄按键...") if receiver is not None: receiver.stop() return True, "正在暂停" def status(self) -> dict[str, object]: with self._lock: connected = ( self._receiver.connected_address if self._receiver else None ) uptime = ( int(time.monotonic() - self._started_at) if self._started_at is not None else 0 ) return { "status": self._status, "connected": connected, "uptime": uptime, "backend": self._resolved_backend, "inputPort": self.port, "discoveryPort": self.discovery_port, "error": self._error, "input": asdict(self._input), } def logs_after(self, after: int) -> list[dict[str, object]]: with self._lock: return [asdict(entry) for entry in self._logs if entry.id > after] def handle_input(self, state: InputState) -> None: with self._lock: self._input = state subscribers = tuple(self._subscribers) for subscriber in subscribers: try: subscriber.put_nowait(state) except queue.Full: try: subscriber.get_nowait() subscriber.put_nowait(state) except (queue.Empty, queue.Full): pass def subscribe(self) -> queue.Queue[InputState]: subscriber: queue.Queue[InputState] = queue.Queue(maxsize=16) with self._lock: self._subscribers.add(subscriber) subscriber.put_nowait(self._input) return subscriber def unsubscribe(self, subscriber: queue.Queue[InputState]) -> None: with self._lock: self._subscribers.discard(subscriber) HTML = r""" Vita Gamepad

Vita Gamepad

局域网虚拟游戏手柄控制台

已暂停
PSVita 连接
尚未连接
运行时间
00:00:00
输入端口
5000
虚拟手柄
自动选择

开启后，程序会广播发现信号并等待同一 Wi-Fi 下的 PSVita。

暂停时会立即发送中立状态、释放所有按键并移除虚拟手柄。再次开启即可重新连接。

请保持此程序运行；关闭这个网页不会停止后台接收服务。

实时按键测试

等待手柄输入
L1 R1 L2 R2 ▲ ▼ ◀ ▶ △ ○ × □ SELECT START L3 R3

运行日志

""" def make_handler( controller: DashboardController, token: str ) -> type[BaseHTTPRequestHandler]: page = HTML.replace("__TOKEN__", token).encode() class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, format: str, *args: object) -> None: return def _authorized(self) -> bool: return self.headers.get("X-Vita-Token") == token def _json(self, payload: object, status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload, ensure_ascii=False).encode() self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _events(self) -> None: subscriber = controller.subscribe() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Connection", "keep-alive") self.end_headers() try: while True: try: state = subscriber.get(timeout=10) payload = json.dumps( asdict(state), ensure_ascii=False, separators=(",", ":") ) self.wfile.write(f"data:{payload}\n\n".encode()) except queue.Empty: self.wfile.write(b": keepalive\n\n") self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass finally: self.close_connection = True controller.unsubscribe(subscriber) def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path == "/": self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(page))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(page) return if parsed.path == "/api/events": query_token = parse_qs(parsed.query).get("token", [""])[0] if not secrets.compare_digest(query_token, token): self._json({"error": "unauthorized"}, HTTPStatus.FORBIDDEN) return self._events() return if not self._authorized(): self._json({"error": "unauthorized"}, HTTPStatus.FORBIDDEN) return if parsed.path == "/api/status": self._json(controller.status()) return if parsed.path == "/api/logs": query = parse_qs(parsed.query) try: after = int(query.get("after", ["0"])[0]) except ValueError: after = 0 self._json({"logs": controller.logs_after(after)}) return self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) def do_POST(self) -> None: if not self._authorized(): self._json({"error": "unauthorized"}, HTTPStatus.FORBIDDEN) return if self.path == "/api/start": ok, message = controller.start() self._json({"ok": ok, "message": message}) return if self.path == "/api/pause": ok, message = controller.pause() self._json({"ok": ok, "message": message}) return self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) return Handler def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Vita Gamepad 图形控制面板") parser.add_argument( "--backend", choices=("auto", "windows", "macos", "debug"), default="auto" ) parser.add_argument("--bind", default="0.0.0.0") parser.add_argument("--port", type=int, default=5000) parser.add_argument("--discovery-port", type=int, default=5001) parser.add_argument("--allow") parser.add_argument("--timeout-ms", type=int, default=300) parser.add_argument("--ui-port", type=int, default=8765) parser.add_argument("--no-browser", action="store_true") return parser def main() -> None: args = build_parser().parse_args() controller = DashboardController( backend_name=args.backend, bind=args.bind, port=args.port, discovery_port=args.discovery_port, allow=args.allow, timeout_ms=args.timeout_ms, ) token = secrets.token_urlsafe(24) server = ThreadingHTTPServer( ("127.0.0.1", args.ui_port), make_handler(controller, token) ) server.daemon_threads = True url = f"http://127.0.0.1:{server.server_port}/" print(f"Vita Gamepad 控制面板：{url}") if not args.no_browser: threading.Timer(0.5, lambda: webbrowser.open(url)).start() try: server.serve_forever() except KeyboardInterrupt: pass finally: controller.pause() server.server_close() if __name__ == "__main__": main()