From 8a16d2e1f1be97d33b07ad922770bd64cef5a380 Mon Sep 17 00:00:00 2001 From: zwt13703 Date: Fri, 17 Jul 2026 17:35:07 +0800 Subject: [PATCH] feat: add Vita network gamepad --- .gitignore | 7 + LICENSE | 21 ++ README.md | 107 ++++++ docs/protocol.md | 28 ++ docs/tasks/task_detail_2026_07_17.md | 23 ++ pyproject.toml | 23 ++ tests/test_protocol.py | 50 +++ tests/test_receiver.py | 82 +++++ vita/CMakeLists.txt | 40 +++ vita/sce_sys/livearea/contents/template.xml | 9 + vita/src/main.c | 341 ++++++++++++++++++++ vitapad/__init__.py | 4 + vitapad/__main__.py | 4 + vitapad/backends/__init__.py | 34 ++ vitapad/backends/debug.py | 28 ++ vitapad/backends/macos.py | 181 +++++++++++ vitapad/backends/windows.py | 67 ++++ vitapad/cli.py | 55 ++++ vitapad/protocol.py | 79 +++++ vitapad/receiver.py | 106 ++++++ 20 files changed, 1289 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 docs/protocol.md create mode 100644 docs/tasks/task_detail_2026_07_17.md create mode 100644 pyproject.toml create mode 100644 tests/test_protocol.py create mode 100644 tests/test_receiver.py create mode 100644 vita/CMakeLists.txt create mode 100644 vita/sce_sys/livearea/contents/template.xml create mode 100644 vita/src/main.c create mode 100644 vitapad/__init__.py create mode 100644 vitapad/__main__.py create mode 100644 vitapad/backends/__init__.py create mode 100644 vitapad/backends/debug.py create mode 100644 vitapad/backends/macos.py create mode 100644 vitapad/backends/windows.py create mode 100644 vitapad/cli.py create mode 100644 vitapad/protocol.py create mode 100644 vitapad/receiver.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e687e7a --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +.venv/ +*.egg-info/ +vita/build/ +.DS_Store + diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..087998a --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Vita Gamepad contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..e84c325 --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# Vita Gamepad + +把一台已安装 HENkaku/enso 的 PlayStation Vita 通过局域网变成 Windows 或 +macOS 游戏手柄。 + +## 功能 + +- 同一 Wi-Fi 下自动发现电脑，无需在 Vita 上输入 IP +- 约 120 Hz 的 UDP 输入传输 +- 双摇杆、方向键、四个动作键、L1/R1、Start/Select +- 后触摸板左右半区模拟 L2/R2 +- 前触摸板底部左右角模拟 L3/R3 +- 断线 300 ms 后自动释放全部按键，避免卡键 +- Windows 输出标准 Xbox 360/XInput 虚拟手柄 +- macOS 输出通用 HID 游戏手柄 + +## 按键映射 + +| PS Vita | Xbox 游戏内名称 | PlayStation 名称 | +|---|---|---| +| × | A | Cross | +| ○ | B | Circle | +| □ | X | Square | +| △ | Y | Triangle | +| L / R | LB / RB | L1 / R1 | +| 后触摸板左 / 右 | LT / RT | L2 / R2 | +| Select / Start | Back / Start | Share / Options | +| 前触摸板左下 / 右下 | LS / RS | L3 / R3 | + +## 电脑端 + +需要 Python 3.10 或更高版本。建议先创建虚拟环境。 + +### Windows + +1. 安装 [ViGEmBus](https://github.com/nefarius/ViGEmBus/releases)。 +2. 在 PowerShell 中运行： + +```powershell +py -m venv .venv +.\.venv\Scripts\Activate.ps1 +pip install -e ".[windows]" +vitapad +``` + +首次运行时允许 Windows 防火墙的“专用网络”访问。程序创建的是 Xbox 360 +控制器，Steam、Xbox 应用和绝大多数 Windows 游戏都能直接识别。 + +### macOS + +```bash +python3 -m venv .venv +source .venv/bin/activate +pip install -e . +sudo .venv/bin/vitapad +``` + +macOS 后端通过 `IOHIDUserDevice` 创建通用 HID 手柄，通常需要管理员权限。 +Apple 并未把用户态虚拟游戏手柄作为稳定的 Game Controller API 保证；个别只 +接受 Apple Game Controller Framework 设备的游戏可能不会显示它。Steam 游戏可 +在“控制器”设置中启用 Steam Input 以提高兼容性。 + +### 仅测试网络，不创建手柄 + +```bash +vitapad --backend debug +``` + +收到数据后会在终端显示按键和摇杆状态，适合排查防火墙或 Wi-Fi 隔离问题。 + +可选参数： + +```text +--port 5000 输入端口 +--discovery-port 5001 自动发现端口 +--bind 0.0.0.0 监听地址 +--allow 192.168.1.25 只接受指定 Vita IP +--timeout-ms 300 断线释放时间 +``` + +## PSVita 端构建与安装 + +需要安装 [VitaSDK](https://vitasdk.org/) 和 `vita2dlib`： + +```bash +cd vita +cmake -S . -B build +cmake --build build +``` + +将 `build/vita-gamepad.vpk` 传到 Vita 并用 VitaShell 安装。启动电脑端 +`vitapad` 后再打开 Vita Gamepad；两台设备必须在同一局域网，且路由器不能启用 +客户端/AP 隔离。 + +Vita 界面显示电脑 IP 和已发送包数。按住 `Start + Select` 两秒可退出。 + +## 网络协议 + +协议说明见 [docs/protocol.md](docs/protocol.md)。它只有固定 20 字节，不执行 +远程命令，也不接收来自电脑的代码。自动发现只发送一个固定文本信标。 + +## 开发测试 + +```bash +python -m unittest discover -s tests -v +``` + diff --git a/docs/protocol.md b/docs/protocol.md new file mode 100644 index 0000000..289f20e --- /dev/null +++ b/docs/protocol.md @@ -0,0 +1,28 @@ +# Vita Gamepad 网络协议 v1 + +电脑每秒向 UDP `5001` 广播一次 ASCII 文本 `VGPD_HOST_V1`。Vita 收到后，将该 +数据包的来源 IP 作为接收端，并向其 UDP `5000` 端口发送输入包。 + +所有多字节整数均为网络字节序。输入包固定为 20 字节： + +| 偏移 | 大小 | 字段 | 含义 | +|---:|---:|---|---| +| 0 | 4 | magic | `VGPD` | +| 4 | 1 | version | `1` | +| 5 | 1 | flags | 保留，当前为 0 | +| 6 | 2 | size | `20` | +| 8 | 4 | sequence | 每包递增，允许 32 位回绕 | +| 12 | 2 | buttons | 按键位图 | +| 14 | 1 | lx | 左摇杆 X，0–255，中心 128 | +| 15 | 1 | ly | 左摇杆 Y，0–255，中心 128 | +| 16 | 1 | rx | 右摇杆 X，0–255，中心 128 | +| 17 | 1 | ry | 右摇杆 Y，0–255，中心 128 | +| 18 | 1 | lt | 左扳机，0–255 | +| 19 | 1 | rt | 右扳机，0–255 | + +按键位从低到高依次为：A、B、X、Y、LB、RB、Back、Start、L3、R3、Up、 +Down、Left、Right；第 14、15 位保留。 + +协议不包含鉴权或加密，设计用途是可信的家庭局域网。可在电脑端使用 +`--allow ` 限制来源。 + diff --git a/docs/tasks/task_detail_2026_07_17.md b/docs/tasks/task_detail_2026_07_17.md new file mode 100644 index 0000000..9c2cede --- /dev/null +++ b/docs/tasks/task_detail_2026_07_17.md @@ -0,0 +1,23 @@ +# 任务执行摘要 + +## 会话 ID: vita-gamepad-initial +- [2026-07-17 17:23:15] +- **执行原因**: 从零实现 PSVita 在同一 Wi-Fi 下作为 Windows/macOS 游戏手柄使用的基础项目。 +- **执行过程**: + 1. 设计 20 字节低延迟 UDP 输入协议、局域网自动发现机制和断线松键保护。 + 2. 实现 PSVita 端双摇杆、物理按键及前后触摸板采集，并以约 120 Hz 发送。 + 3. 实现 Windows ViGEm/XInput 虚拟 Xbox 360 手柄后端。 + 4. 实现 macOS IOHIDUserDevice 通用 HID 游戏手柄后端。 + 5. 补充电脑端安装入口、VitaSDK 构建配置、按键映射、协议和排障文档。 + 6. 完成协议编解码、序列号回绕、UDP 接收、旧包过滤和断线保护测试。 +- **执行结果**: 已形成可构建的跨端 MVP；电脑端 6 项自动化测试全部通过。当前环境未安装 VitaSDK，VPK 需在配置 VitaSDK 的机器上完成编译；macOS 创建虚拟 HID 需要管理员权限。 + +## 会话 ID: vita-gamepad-publish +- [2026-07-17 17:34:57] +- **执行原因**: 将 Vita Gamepad 项目发布到用户指定的 Git 远程仓库。 +- **执行过程**: + 1. 检查本地工作区、当前分支和提交历史。 + 2. 确认目标远程仓库为空，避免覆盖既有提交。 + 3. 提交项目全部源码、测试与文档。 + 4. 配置远程地址并推送 main 分支。 +- **执行结果**: Vita Gamepad 项目已提交并推送至指定仓库。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..06568ff --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "vita-gamepad" +version = "0.1.0" +description = "Use a PlayStation Vita as a low-latency network gamepad" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +authors = [{ name = "Vita Gamepad contributors" }] +dependencies = [] + +[project.optional-dependencies] +windows = ["vgamepad>=0.1.0"] + +[project.scripts] +vitapad = "vitapad.cli:main" + +[tool.setuptools.packages.find] +include = ["vitapad*"] + diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..9eaa07d --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,50 @@ +import unittest + +from vitapad.protocol import ( + Buttons, + InputState, + PACKET_SIZE, + decode_packet, + encode_packet, + sequence_is_newer, +) + + +class ProtocolTests(unittest.TestCase): + def test_round_trip(self) -> None: + state = InputState( + sequence=123456, + buttons=Buttons.A | Buttons.UP, + lx=1, + ly=2, + rx=253, + ry=254, + lt=17, + rt=240, + ) + encoded = encode_packet(state) + self.assertEqual(len(encoded), PACKET_SIZE) + self.assertEqual(decode_packet(encoded), state) + + def test_rejects_bad_packets(self) -> None: + with self.assertRaises(ValueError): + decode_packet(b"") + packet = bytearray(encode_packet(InputState())) + packet[0] = 0 + with self.assertRaises(ValueError): + decode_packet(packet) + + def test_sequence_comparison_wraps(self) -> None: + self.assertTrue(sequence_is_newer(11, 10)) + self.assertFalse(sequence_is_newer(10, 10)) + self.assertFalse(sequence_is_newer(9, 10)) + self.assertTrue(sequence_is_newer(0, 0xFFFFFFFF)) + + def test_axis_validation(self) -> None: + with self.assertRaises(ValueError): + encode_packet(InputState(lx=256)) + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_receiver.py b/tests/test_receiver.py new file mode 100644 index 0000000..d8d219b --- /dev/null +++ b/tests/test_receiver.py @@ -0,0 +1,82 @@ +import socket +import threading +import time +import unittest + +from vitapad.protocol import Buttons, InputState, encode_packet +from vitapad.receiver import Receiver + + +class FakeBackend: + name = "fake" + + def __init__(self) -> None: + self.states: list[InputState] = [] + self.closed = False + + def update(self, state: InputState) -> None: + self.states.append(state) + + def close(self) -> None: + self.closed = True + + +def unused_udp_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class ReceiverTests(unittest.TestCase): + def test_receive_and_failsafe(self) -> None: + backend = FakeBackend() + port = unused_udp_port() + receiver = Receiver( + backend, + bind="127.0.0.1", + port=port, + discovery_port=unused_udp_port(), + timeout_ms=50, + ) + thread = threading.Thread(target=receiver.run) + thread.start() + try: + state = InputState(sequence=7, buttons=Buttons.A, lx=200) + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sender: + # Give the receiver thread time to bind. + time.sleep(0.02) + sender.sendto(encode_packet(state), ("127.0.0.1", port)) + deadline = time.monotonic() + 1.0 + while len(backend.states) < 2 and time.monotonic() < deadline: + time.sleep(0.01) + self.assertEqual(backend.states[0], state) + self.assertEqual(backend.states[1], InputState.neutral()) + finally: + receiver.stop() + thread.join(timeout=2) + self.assertFalse(thread.is_alive()) + self.assertTrue(backend.closed) + + def test_rejects_old_sequence(self) -> None: + backend = FakeBackend() + receiver = Receiver(backend) + receiver._connected_address = "127.0.0.1" + receiver._last_sequence = 20 + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + listener.bind(("127.0.0.1", 0)) + listener.settimeout(1) + sender.sendto( + encode_packet(InputState(sequence=19, buttons=Buttons.B)), + listener.getsockname(), + ) + receiver._receive_once(listener) + finally: + listener.close() + sender.close() + self.assertEqual(backend.states, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/vita/CMakeLists.txt b/vita/CMakeLists.txt new file mode 100644 index 0000000..3b05ddc --- /dev/null +++ b/vita/CMakeLists.txt @@ -0,0 +1,40 @@ +cmake_minimum_required(VERSION 3.16) + +if(NOT DEFINED CMAKE_TOOLCHAIN_FILE) + if(DEFINED ENV{VITASDK}) + set(CMAKE_TOOLCHAIN_FILE "$ENV{VITASDK}/share/vita.toolchain.cmake" + CACHE PATH "Vita toolchain file") + else() + message(FATAL_ERROR "Please define VITASDK to point to your SDK path") + endif() +endif() + +project(vita_gamepad C) +include("${VITASDK}/share/vita.cmake" REQUIRED) + +set(VITA_APP_NAME "Vita Gamepad") +set(VITA_TITLEID "VITAPAD01") +set(VITA_VERSION "01.00") + +add_executable(vita-gamepad src/main.c) +target_compile_options(vita-gamepad PRIVATE -Wall -Wextra -Werror -O2) +target_link_libraries(vita-gamepad + vita2d + SceDisplay_stub + SceGxm_stub + SceSysmodule_stub + SceCtrl_stub + SceTouch_stub + SceNet_stub + SceNetCtl_stub + ScePgf_stub + SceCommonDialog_stub + SceAppMgr_stub + m +) + +vita_create_self(vita-gamepad.self vita-gamepad UNSAFE) +vita_create_vpk(vita-gamepad.vpk ${VITA_TITLEID} vita-gamepad.self + VERSION ${VITA_VERSION} + NAME ${VITA_APP_NAME} +) diff --git a/vita/sce_sys/livearea/contents/template.xml b/vita/sce_sys/livearea/contents/template.xml new file mode 100644 index 0000000..efcc879 --- /dev/null +++ b/vita/sce_sys/livearea/contents/template.xml @@ -0,0 +1,9 @@ + + + + bg.png + + + startup.png + + diff --git a/vita/src/main.c b/vita/src/main.c new file mode 100644 index 0000000..eb500cb --- /dev/null +++ b/vita/src/main.c @@ -0,0 +1,341 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#define INPUT_PORT 5000 +#define DISCOVERY_PORT 5001 +#define SEND_INTERVAL_US 8333 +#define DISCOVERY_MAGIC "VGPD_HOST_V1" +#define PACKET_MAGIC "VGPD" +#define PACKET_VERSION 1 + +enum ProtocolButton { + BTN_A = 1u << 0, + BTN_B = 1u << 1, + BTN_X = 1u << 2, + BTN_Y = 1u << 3, + BTN_LB = 1u << 4, + BTN_RB = 1u << 5, + BTN_BACK = 1u << 6, + BTN_START = 1u << 7, + BTN_L3 = 1u << 8, + BTN_R3 = 1u << 9, + BTN_UP = 1u << 10, + BTN_DOWN = 1u << 11, + BTN_LEFT = 1u << 12, + BTN_RIGHT = 1u << 13, +}; + +typedef struct __attribute__((packed)) InputPacket { + char magic[4]; + uint8_t version; + uint8_t flags; + uint16_t size; + uint32_t sequence; + uint16_t buttons; + uint8_t lx; + uint8_t ly; + uint8_t rx; + uint8_t ry; + uint8_t lt; + uint8_t rt; +} InputPacket; + +_Static_assert(sizeof(InputPacket) == 20, "InputPacket must be exactly 20 bytes"); + +static uint8_t net_memory[1024 * 1024]; +static volatile bool exiting = false; + +static int exit_callback(int argc, const void *args) { + (void)argc; + (void)args; + exiting = true; + return 0; +} + +static int callback_thread(SceSize args, void *argp) { + (void)args; + (void)argp; + int callback = sceKernelCreateCallback("Exit callback", exit_callback, NULL); + sceKernelRegisterExitCallback(callback); + sceKernelDelayThreadCB(0xFFFFFFFF); + return 0; +} + +static void setup_callbacks(void) { + int thread = sceKernelCreateThread( + "callback thread", callback_thread, 0x10000100, 0x1000, 0, 0, NULL + ); + if (thread >= 0) { + sceKernelStartThread(thread, 0, NULL); + } +} + +static int init_network(void) { + int result = sceSysmoduleLoadModule(SCE_SYSMODULE_NET); + if (result < 0) { + return result; + } + SceNetInitParam params; + memset(¶ms, 0, sizeof(params)); + params.memory = net_memory; + params.size = sizeof(net_memory); + result = sceNetInit(¶ms); + if (result < 0) { + return result; + } + return sceNetCtlInit(); +} + +static int create_socket(void) { + int sock = sceNetSocket( + "vita-gamepad", SCE_NET_AF_INET, SCE_NET_SOCK_DGRAM, 0 + ); + if (sock < 0) { + return sock; + } + + int nonblocking = 1; + sceNetSetsockopt( + sock, SCE_NET_SOL_SOCKET, SCE_NET_SO_NBIO, + &nonblocking, sizeof(nonblocking) + ); + + SceNetSockaddrIn local; + memset(&local, 0, sizeof(local)); + local.sin_len = sizeof(local); + local.sin_family = SCE_NET_AF_INET; + local.sin_port = sceNetHtons(DISCOVERY_PORT); + local.sin_addr.s_addr = sceNetHtonl(SCE_NET_INADDR_ANY); + int result = sceNetBind(sock, (SceNetSockaddr *)&local, sizeof(local)); + if (result < 0) { + sceNetSocketClose(sock); + return result; + } + return sock; +} + +static bool find_host(int sock, SceNetSockaddrIn *host) { + char buffer[32]; + SceNetSockaddrIn source; + unsigned int source_length = sizeof(source); + int count = sceNetRecvfrom( + sock, buffer, sizeof(buffer), 0, + (SceNetSockaddr *)&source, &source_length + ); + if (count != (int)strlen(DISCOVERY_MAGIC) || + memcmp(buffer, DISCOVERY_MAGIC, strlen(DISCOVERY_MAGIC)) != 0) { + return false; + } + *host = source; + host->sin_port = sceNetHtons(INPUT_PORT); + return true; +} + +static uint16_t map_buttons(uint32_t vita_buttons) { + uint16_t buttons = 0; + if (vita_buttons & SCE_CTRL_CROSS) buttons |= BTN_A; + if (vita_buttons & SCE_CTRL_CIRCLE) buttons |= BTN_B; + if (vita_buttons & SCE_CTRL_SQUARE) buttons |= BTN_X; + if (vita_buttons & SCE_CTRL_TRIANGLE) buttons |= BTN_Y; + if (vita_buttons & SCE_CTRL_LTRIGGER) buttons |= BTN_LB; + if (vita_buttons & SCE_CTRL_RTRIGGER) buttons |= BTN_RB; + if (vita_buttons & SCE_CTRL_SELECT) buttons |= BTN_BACK; + if (vita_buttons & SCE_CTRL_START) buttons |= BTN_START; + if (vita_buttons & SCE_CTRL_UP) buttons |= BTN_UP; + if (vita_buttons & SCE_CTRL_DOWN) buttons |= BTN_DOWN; + if (vita_buttons & SCE_CTRL_LEFT) buttons |= BTN_LEFT; + if (vita_buttons & SCE_CTRL_RIGHT) buttons |= BTN_RIGHT; + return buttons; +} + +static void read_touch(uint16_t *buttons, uint8_t *lt, uint8_t *rt) { + SceTouchData touch; + *lt = 0; + *rt = 0; + + memset(&touch, 0, sizeof(touch)); + if (sceTouchPeek(SCE_TOUCH_PORT_BACK, &touch, 1) >= 0) { + for (uint32_t i = 0; i < touch.reportNum; ++i) { + if (touch.report[i].x < 960) { + *lt = 255; + } else { + *rt = 255; + } + } + } + + memset(&touch, 0, sizeof(touch)); + if (sceTouchPeek(SCE_TOUCH_PORT_FRONT, &touch, 1) >= 0) { + for (uint32_t i = 0; i < touch.reportNum; ++i) { + if (touch.report[i].y >= 850 && touch.report[i].x < 480) { + *buttons |= BTN_L3; + } else if (touch.report[i].y >= 850 && + touch.report[i].x > 1440) { + *buttons |= BTN_R3; + } + } + } +} + +static void make_packet( + InputPacket *packet, const SceCtrlData *pad, uint32_t sequence +) { + memset(packet, 0, sizeof(*packet)); + memcpy(packet->magic, PACKET_MAGIC, sizeof(packet->magic)); + packet->version = PACKET_VERSION; + packet->size = sceNetHtons(sizeof(*packet)); + packet->sequence = sceNetHtonl(sequence); + uint16_t buttons = map_buttons(pad->buttons); + read_touch(&buttons, &packet->lt, &packet->rt); + packet->buttons = sceNetHtons(buttons); + packet->lx = pad->lx; + packet->ly = pad->ly; + packet->rx = pad->rx; + packet->ry = pad->ry; +} + +static void format_ip(const SceNetSockaddrIn *address, char *out, size_t size) { + uint32_t ip = sceNetNtohl(address->sin_addr.s_addr); + snprintf( + out, size, "%u.%u.%u.%u", + (unsigned)((ip >> 24) & 0xff), (unsigned)((ip >> 16) & 0xff), + (unsigned)((ip >> 8) & 0xff), (unsigned)(ip & 0xff) + ); +} + +static void draw_ui( + vita2d_pgf *font, bool connected, const char *host_ip, + uint32_t sent, int network_error +) { + vita2d_start_drawing(); + vita2d_clear_screen(); + vita2d_draw_rectangle(0, 0, 960, 544, RGBA8(18, 24, 38, 255)); + vita2d_draw_rectangle(0, 0, 960, 8, RGBA8(38, 190, 125, 255)); + vita2d_pgf_draw_text( + font, 54, 92, RGBA8(255, 255, 255, 255), 1.6f, "Vita Gamepad" + ); + if (network_error < 0) { + char error[96]; + snprintf(error, sizeof(error), "Network error: 0x%08X", network_error); + vita2d_pgf_draw_text( + font, 54, 176, RGBA8(255, 105, 105, 255), 1.0f, error + ); + } else if (!connected) { + vita2d_pgf_draw_text( + font, 54, 176, RGBA8(245, 200, 80, 255), 1.0f, + "Waiting for a computer on the same Wi-Fi..." + ); + vita2d_pgf_draw_text( + font, 54, 218, RGBA8(180, 190, 205, 255), 0.8f, + "Start 'vitapad' on Windows or macOS." + ); + } else { + char status[128]; + snprintf(status, sizeof(status), "Connected: %s", host_ip); + vita2d_pgf_draw_text( + font, 54, 176, RGBA8(80, 225, 155, 255), 1.0f, status + ); + snprintf(status, sizeof(status), "Packets sent: %u", (unsigned)sent); + vita2d_pgf_draw_text( + font, 54, 218, RGBA8(180, 190, 205, 255), 0.8f, status + ); + } + vita2d_pgf_draw_text( + font, 54, 390, RGBA8(180, 190, 205, 255), 0.75f, + "Rear touch: L2 / R2 Front bottom corners: L3 / R3" + ); + vita2d_pgf_draw_text( + font, 54, 438, RGBA8(140, 150, 165, 255), 0.7f, + "Hold START + SELECT for 2 seconds to exit" + ); + vita2d_end_drawing(); + vita2d_swap_buffers(); +} + +int main(void) { + setup_callbacks(); + sceCtrlSetSamplingMode(SCE_CTRL_MODE_ANALOG_WIDE); + sceTouchSetSamplingState(SCE_TOUCH_PORT_FRONT, SCE_TOUCH_SAMPLING_STATE_START); + sceTouchSetSamplingState(SCE_TOUCH_PORT_BACK, SCE_TOUCH_SAMPLING_STATE_START); + + vita2d_init(); + vita2d_set_clear_color(RGBA8(18, 24, 38, 255)); + vita2d_pgf *font = vita2d_load_default_pgf(); + + int network_result = init_network(); + bool network_initialized = network_result >= 0; + int sock = network_result >= 0 ? create_socket() : network_result; + if (sock < 0) { + network_result = sock; + } + + bool connected = false; + SceNetSockaddrIn host; + memset(&host, 0, sizeof(host)); + char host_ip[32] = ""; + uint32_t sequence = 0; + uint32_t sent = 0; + uint32_t exit_frames = 0; + uint32_t draw_divider = 0; + + while (!exiting) { + if (sock >= 0 && find_host(sock, &host)) { + connected = true; + format_ip(&host, host_ip, sizeof(host_ip)); + } + + SceCtrlData pad; + memset(&pad, 0, sizeof(pad)); + if (sceCtrlPeekBufferPositive(0, &pad, 1) > 0) { + if ((pad.buttons & (SCE_CTRL_START | SCE_CTRL_SELECT)) == + (SCE_CTRL_START | SCE_CTRL_SELECT)) { + exit_frames++; + if (exit_frames >= 240) exiting = true; + } else { + exit_frames = 0; + } + + if (connected) { + InputPacket packet; + make_packet(&packet, &pad, sequence++); + int result = sceNetSendto( + sock, &packet, sizeof(packet), 0, + (SceNetSockaddr *)&host, sizeof(host) + ); + if (result == (int)sizeof(packet)) { + sent++; + } else if (result < 0) { + connected = false; + } + } + } + + if (++draw_divider >= 4) { + draw_ui(font, connected, host_ip, sent, network_result); + draw_divider = 0; + } + sceKernelDelayThread(SEND_INTERVAL_US); + } + + if (sock >= 0) sceNetSocketClose(sock); + if (network_initialized) { + sceNetCtlTerm(); + sceNetTerm(); + sceSysmoduleUnloadModule(SCE_SYSMODULE_NET); + } + vita2d_free_pgf(font); + vita2d_fini(); + sceKernelExitProcess(0); + return 0; +} diff --git a/vitapad/__init__.py b/vitapad/__init__.py new file mode 100644 index 0000000..17c6c41 --- /dev/null +++ b/vitapad/__init__.py @@ -0,0 +1,4 @@ +"""Vita Gamepad desktop receiver.""" + +__version__ = "0.1.0" + diff --git a/vitapad/__main__.py b/vitapad/__main__.py new file mode 100644 index 0000000..b1fb266 --- /dev/null +++ b/vitapad/__main__.py @@ -0,0 +1,4 @@ +from vitapad.cli import main + +main() + diff --git a/vitapad/backends/__init__.py b/vitapad/backends/__init__.py new file mode 100644 index 0000000..0861653 --- /dev/null +++ b/vitapad/backends/__init__.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import sys +from typing import Protocol + +from vitapad.protocol import InputState + + +class GamepadBackend(Protocol): + name: str + + def update(self, state: InputState) -> None: ... + + def close(self) -> None: ... + + +def create_backend(name: str) -> GamepadBackend: + selected = name + if selected == "auto": + selected = "windows" if sys.platform == "win32" else "macos" if sys.platform == "darwin" else "debug" + if selected == "windows": + from .windows import WindowsGamepad + + return WindowsGamepad() + if selected == "macos": + from .macos import MacOSGamepad + + return MacOSGamepad() + if selected == "debug": + from .debug import DebugGamepad + + return DebugGamepad() + raise ValueError(f"unknown backend: {name}") + diff --git a/vitapad/backends/debug.py b/vitapad/backends/debug.py new file mode 100644 index 0000000..e3f5926 --- /dev/null +++ b/vitapad/backends/debug.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import time + +from vitapad.protocol import InputState + + +class DebugGamepad: + name = "debug" + + def __init__(self) -> None: + self._last_print = 0.0 + + def update(self, state: InputState) -> None: + now = time.monotonic() + if now - self._last_print >= 0.1: + print( + f"\rseq={state.sequence:<10} buttons=0x{state.buttons:04x} " + f"L=({state.lx:3},{state.ly:3}) R=({state.rx:3},{state.ry:3}) " + f"LT={state.lt:3} RT={state.rt:3}", + end="", + flush=True, + ) + self._last_print = now + + def close(self) -> None: + print() + diff --git a/vitapad/backends/macos.py b/vitapad/backends/macos.py new file mode 100644 index 0000000..2a92fb5 --- /dev/null +++ b/vitapad/backends/macos.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import ctypes +import ctypes.util +import struct + +from vitapad.protocol import Buttons, InputState + +# Report 1: 16 buttons, 8-way hat, four 16-bit axes and two 8-bit triggers. +REPORT_DESCRIPTOR = bytes( + [ + 0x05, 0x01, 0x09, 0x05, 0xA1, 0x01, 0x85, 0x01, + 0x05, 0x09, 0x19, 0x01, 0x29, 0x10, 0x15, 0x00, + 0x25, 0x01, 0x75, 0x01, 0x95, 0x10, 0x81, 0x02, + 0x05, 0x01, 0x09, 0x39, 0x15, 0x00, 0x25, 0x07, + 0x35, 0x00, 0x46, 0x3B, 0x01, 0x65, 0x14, 0x75, + 0x04, 0x95, 0x01, 0x81, 0x42, 0x65, 0x00, 0x75, + 0x04, 0x95, 0x01, 0x81, 0x03, 0x09, 0x30, 0x09, + 0x31, 0x09, 0x33, 0x09, 0x34, 0x15, 0x00, 0x26, + 0xFF, 0xFF, 0x75, 0x10, 0x95, 0x04, 0x81, 0x02, + 0x09, 0x32, 0x09, 0x35, 0x15, 0x00, 0x26, 0xFF, + 0x00, 0x75, 0x08, 0x95, 0x02, 0x81, 0x02, 0xC0, + ] +) + +BUTTON_ORDER = ( + Buttons.A, Buttons.B, Buttons.X, Buttons.Y, + Buttons.LB, Buttons.RB, Buttons.BACK, Buttons.START, + Buttons.L3, Buttons.R3, +) + + +def _hat(buttons: int) -> int: + up, down = bool(buttons & Buttons.UP), bool(buttons & Buttons.DOWN) + left, right = bool(buttons & Buttons.LEFT), bool(buttons & Buttons.RIGHT) + if up and right: + return 1 + if right and down: + return 3 + if down and left: + return 5 + if left and up: + return 7 + if up and not down: + return 0 + if right and not left: + return 2 + if down and not up: + return 4 + if left and not right: + return 6 + return 8 + + +class MacOSGamepad: + name = "Generic HID gamepad (IOHIDUserDevice)" + + def __init__(self) -> None: + cf_path = ctypes.util.find_library("CoreFoundation") + io_path = ctypes.util.find_library("IOKit") + if not cf_path or not io_path: + raise RuntimeError("找不到 macOS CoreFoundation/IOKit") + self._cf = ctypes.CDLL(cf_path) + self._io = ctypes.CDLL(io_path) + self._configure_functions() + properties = self._create_properties() + self._device = self._io.IOHIDUserDeviceCreate(None, properties) + self._cf.CFRelease(properties) + if not self._device: + raise RuntimeError( + "无法创建 macOS 虚拟 HID 手柄；请尝试使用 sudo 运行" + ) + + def _configure_functions(self) -> None: + self._cf.CFStringCreateWithCString.restype = ctypes.c_void_p + self._cf.CFStringCreateWithCString.argtypes = [ + ctypes.c_void_p, ctypes.c_char_p, ctypes.c_uint32 + ] + self._cf.CFDataCreate.restype = ctypes.c_void_p + self._cf.CFDataCreate.argtypes = [ + ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_long + ] + self._cf.CFNumberCreate.restype = ctypes.c_void_p + self._cf.CFNumberCreate.argtypes = [ + ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p + ] + self._cf.CFDictionaryCreateMutable.restype = ctypes.c_void_p + self._cf.CFDictionaryCreateMutable.argtypes = [ + ctypes.c_void_p, ctypes.c_long, ctypes.c_void_p, ctypes.c_void_p + ] + self._cf.CFDictionarySetValue.argtypes = [ + ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p + ] + self._cf.CFRelease.argtypes = [ctypes.c_void_p] + self._io.IOHIDUserDeviceCreate.restype = ctypes.c_void_p + self._io.IOHIDUserDeviceCreate.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + self._io.IOHIDUserDeviceHandleReport.argtypes = [ + ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_long + ] + self._io.IOHIDUserDeviceHandleReport.restype = ctypes.c_int + self._dictionary_key_callbacks = ctypes.c_byte.in_dll( + self._cf, "kCFTypeDictionaryKeyCallBacks" + ) + self._dictionary_value_callbacks = ctypes.c_byte.in_dll( + self._cf, "kCFTypeDictionaryValueCallBacks" + ) + + def _string(self, value: str) -> int: + # kCFStringEncodingUTF8 + return self._cf.CFStringCreateWithCString(None, value.encode(), 0x08000100) + + def _create_properties(self) -> int: + dictionary = self._cf.CFDictionaryCreateMutable( + None, + 0, + ctypes.byref(self._dictionary_key_callbacks), + ctypes.byref(self._dictionary_value_callbacks), + ) + retained: list[int] = [] + + def put_number(key: str, value: int) -> None: + key_ref = self._string(key) + raw = ctypes.c_int(value) + value_ref = self._cf.CFNumberCreate(None, 9, ctypes.byref(raw)) + self._cf.CFDictionarySetValue(dictionary, key_ref, value_ref) + retained.extend((key_ref, value_ref)) + + def put_string(key: str, value: str) -> None: + key_ref, value_ref = self._string(key), self._string(value) + self._cf.CFDictionarySetValue(dictionary, key_ref, value_ref) + retained.extend((key_ref, value_ref)) + + descriptor_array = (ctypes.c_uint8 * len(REPORT_DESCRIPTOR)).from_buffer_copy( + REPORT_DESCRIPTOR + ) + descriptor = self._cf.CFDataCreate( + None, descriptor_array, len(REPORT_DESCRIPTOR) + ) + descriptor_key = self._string("ReportDescriptor") + self._cf.CFDictionarySetValue(dictionary, descriptor_key, descriptor) + retained.extend((descriptor_key, descriptor)) + put_number("VendorID", 0x054C) + put_number("ProductID", 0x09CC) + put_number("VersionNumber", 0x0100) + put_number("PrimaryUsagePage", 0x01) + put_number("PrimaryUsage", 0x05) + put_string("Manufacturer", "Vita Gamepad") + put_string("Product", "PS Vita Network Gamepad") + for item in retained: + self._cf.CFRelease(item) + return dictionary + + def update(self, state: InputState) -> None: + hid_buttons = 0 + for index, mask in enumerate(BUTTON_ORDER): + if state.buttons & mask: + hid_buttons |= 1 << index + report = struct.pack( + " None: + if self._device: + self.update(InputState.neutral()) + self._cf.CFRelease(self._device) + self._device = None diff --git a/vitapad/backends/windows.py b/vitapad/backends/windows.py new file mode 100644 index 0000000..67e1f51 --- /dev/null +++ b/vitapad/backends/windows.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from vitapad.protocol import Buttons, InputState + +try: + import vgamepad as vg +except ImportError as exc: + raise RuntimeError( + 'Windows 后端需要 vgamepad，请运行: pip install -e ".[windows]"' + ) from exc + + +BUTTON_MAP = ( + (Buttons.A, vg.XUSB_BUTTON.XUSB_GAMEPAD_A), + (Buttons.B, vg.XUSB_BUTTON.XUSB_GAMEPAD_B), + (Buttons.X, vg.XUSB_BUTTON.XUSB_GAMEPAD_X), + (Buttons.Y, vg.XUSB_BUTTON.XUSB_GAMEPAD_Y), + (Buttons.LB, vg.XUSB_BUTTON.XUSB_GAMEPAD_LEFT_SHOULDER), + (Buttons.RB, vg.XUSB_BUTTON.XUSB_GAMEPAD_RIGHT_SHOULDER), + (Buttons.BACK, vg.XUSB_BUTTON.XUSB_GAMEPAD_BACK), + (Buttons.START, vg.XUSB_BUTTON.XUSB_GAMEPAD_START), + (Buttons.L3, vg.XUSB_BUTTON.XUSB_GAMEPAD_LEFT_THUMB), + (Buttons.R3, vg.XUSB_BUTTON.XUSB_GAMEPAD_RIGHT_THUMB), + (Buttons.UP, vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_UP), + (Buttons.DOWN, vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_DOWN), + (Buttons.LEFT, vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_LEFT), + (Buttons.RIGHT, vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_RIGHT), +) + + +def _axis(value: int, *, invert: bool = False) -> int: + if value >= 128: + result = round((value - 128) * 32767 / 127) + else: + result = round((value - 128) * 32768 / 128) + return -result if invert else result + + +class WindowsGamepad: + name = "Xbox 360 (ViGEm)" + + def __init__(self) -> None: + try: + self._pad = vg.VX360Gamepad() + except Exception as exc: + raise RuntimeError( + "无法创建虚拟手柄，请确认已安装 ViGEmBus 驱动" + ) from exc + + def update(self, state: InputState) -> None: + self._pad.reset() + for mask, button in BUTTON_MAP: + if state.buttons & mask: + self._pad.press_button(button=button) + self._pad.left_joystick( + x_value=_axis(state.lx), y_value=_axis(state.ly, invert=True) + ) + self._pad.right_joystick( + x_value=_axis(state.rx), y_value=_axis(state.ry, invert=True) + ) + self._pad.left_trigger(value=state.lt) + self._pad.right_trigger(value=state.rt) + self._pad.update() + + def close(self) -> None: + self.update(InputState.neutral()) + del self._pad diff --git a/vitapad/cli.py b/vitapad/cli.py new file mode 100644 index 0000000..287e18d --- /dev/null +++ b/vitapad/cli.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import argparse +import sys + +from vitapad.backends import create_backend +from vitapad.receiver import Receiver + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="vitapad", description="将 PSVita 输入转换为电脑虚拟游戏手柄" + ) + parser.add_argument( + "--backend", choices=("auto", "windows", "macos", "debug"), default="auto" + ) + parser.add_argument("--bind", default="0.0.0.0", help="UDP 监听地址") + parser.add_argument("--port", type=int, default=5000, help="输入端口") + parser.add_argument( + "--discovery-port", type=int, default=5001, help="自动发现广播端口" + ) + parser.add_argument("--allow", metavar="IP", help="只接受指定 PSVita IP") + parser.add_argument( + "--timeout-ms", type=int, default=300, help="断线释放按键时间" + ) + return parser + + +def main() -> None: + args = build_parser().parse_args() + if not 1 <= args.port <= 65535 or not 1 <= args.discovery_port <= 65535: + raise SystemExit("端口必须在 1 到 65535 之间") + if not 50 <= args.timeout_ms <= 10000: + raise SystemExit("--timeout-ms 必须在 50 到 10000 之间") + try: + backend = create_backend(args.backend) + receiver = Receiver( + backend=backend, + bind=args.bind, + port=args.port, + discovery_port=args.discovery_port, + allow=args.allow, + timeout_ms=args.timeout_ms, + ) + receiver.run() + except KeyboardInterrupt: + print("\n已停止") + except (OSError, RuntimeError, ValueError) as exc: + print(f"错误: {exc}", file=sys.stderr) + raise SystemExit(1) from exc + + +if __name__ == "__main__": + main() + diff --git a/vitapad/protocol.py b/vitapad/protocol.py new file mode 100644 index 0000000..f5eaf08 --- /dev/null +++ b/vitapad/protocol.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from dataclasses import dataclass +import struct + +MAGIC = b"VGPD" +VERSION = 1 +PACKET_SIZE = 20 +DISCOVERY_MAGIC = b"VGPD_HOST_V1" +PACKET_STRUCT = struct.Struct("!4sBBHIH6B") + + +class Buttons: + A = 1 << 0 + B = 1 << 1 + X = 1 << 2 + Y = 1 << 3 + LB = 1 << 4 + RB = 1 << 5 + BACK = 1 << 6 + START = 1 << 7 + L3 = 1 << 8 + R3 = 1 << 9 + UP = 1 << 10 + DOWN = 1 << 11 + LEFT = 1 << 12 + RIGHT = 1 << 13 + + +@dataclass(frozen=True, slots=True) +class InputState: + sequence: int = 0 + buttons: int = 0 + lx: int = 128 + ly: int = 128 + rx: int = 128 + ry: int = 128 + lt: int = 0 + rt: int = 0 + flags: int = 0 + + @classmethod + def neutral(cls) -> "InputState": + return cls() + + +def decode_packet(data: bytes) -> InputState: + if len(data) != PACKET_SIZE: + raise ValueError(f"invalid packet size: {len(data)}") + magic, version, flags, size, sequence, buttons, *axes = PACKET_STRUCT.unpack(data) + if magic != MAGIC: + raise ValueError("invalid packet magic") + if version != VERSION: + raise ValueError(f"unsupported protocol version: {version}") + if size != PACKET_SIZE: + raise ValueError(f"invalid declared packet size: {size}") + return InputState(sequence, buttons, *axes, flags) + + +def encode_packet(state: InputState) -> bytes: + values = (state.lx, state.ly, state.rx, state.ry, state.lt, state.rt) + if any(not 0 <= value <= 255 for value in values): + raise ValueError("axis and trigger values must be between 0 and 255") + return PACKET_STRUCT.pack( + MAGIC, + VERSION, + state.flags, + PACKET_SIZE, + state.sequence & 0xFFFFFFFF, + state.buttons & 0xFFFF, + *values, + ) + + +def sequence_is_newer(candidate: int, previous: int) -> bool: + """Compare wrapping unsigned 32-bit sequence numbers.""" + difference = (candidate - previous) & 0xFFFFFFFF + return 0 < difference < 0x80000000 + diff --git a/vitapad/receiver.py b/vitapad/receiver.py new file mode 100644 index 0000000..6568b30 --- /dev/null +++ b/vitapad/receiver.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import socket +import threading +import time + +from vitapad.backends import GamepadBackend +from vitapad.protocol import ( + DISCOVERY_MAGIC, + InputState, + decode_packet, + sequence_is_newer, +) + + +class Receiver: + def __init__( + self, + backend: GamepadBackend, + bind: str = "0.0.0.0", + port: int = 5000, + discovery_port: int = 5001, + allow: str | None = None, + timeout_ms: int = 300, + ) -> None: + self.backend = backend + self.bind = bind + self.port = port + self.discovery_port = discovery_port + self.allow = allow + self.timeout = timeout_ms / 1000.0 + self._stop = threading.Event() + self._last_packet = 0.0 + self._connected_address: str | None = None + self._last_sequence: int | None = None + self._neutral_sent = True + + def _broadcast_discovery(self) -> None: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as beacon: + beacon.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + while not self._stop.wait(1.0): + try: + beacon.sendto( + DISCOVERY_MAGIC, ("255.255.255.255", self.discovery_port) + ) + except OSError: + # Interfaces can disappear during sleep/network changes. + pass + + def run(self) -> None: + discovery = threading.Thread( + target=self._broadcast_discovery, name="discovery", daemon=True + ) + discovery.start() + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as listener: + listener.bind((self.bind, self.port)) + listener.settimeout(0.05) + print( + f"正在监听 {self.bind}:{self.port}，后端: {self.backend.name}\n" + "等待 PSVita（Ctrl+C 退出）..." + ) + try: + while not self._stop.is_set(): + self._receive_once(listener) + self._apply_failsafe() + finally: + self._stop.set() + discovery.join(timeout=1.2) + self.backend.close() + + def _receive_once(self, listener: socket.socket) -> None: + try: + data, address = listener.recvfrom(256) + except socket.timeout: + return + if self.allow and address[0] != self.allow: + return + try: + state = decode_packet(data) + except ValueError: + return + if self._connected_address != address[0]: + self._connected_address = address[0] + self._last_sequence = None + print(f"已连接 PSVita: {address[0]}") + if self._last_sequence is not None and not sequence_is_newer( + state.sequence, self._last_sequence + ): + return + self._last_sequence = state.sequence + self._last_packet = time.monotonic() + self._neutral_sent = False + self.backend.update(state) + + def _apply_failsafe(self) -> None: + if ( + not self._neutral_sent + and time.monotonic() - self._last_packet >= self.timeout + ): + self.backend.update(InputState.neutral()) + self._neutral_sent = True + print("连接超时，已释放全部按键；等待重连...") + + def stop(self) -> None: + self._stop.set() +