diff --git a/dglab_ws/README.md b/dglab_ws/README.md index 36e84af..6280455 100644 --- a/dglab_ws/README.md +++ b/dglab_ws/README.md @@ -154,6 +154,33 @@ A: 实际强度不会超过软上限。比如软上限 100，设为 150 实际 **Q: 波形数据格式是什么？** A: 每条 8 字节 HEX，前 4 字节是频率，后 4 字节是强度。详见 [官方波形说明](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/blob/main/coyote/extra/README.md)。 +## 测试（无需真机） + +本插件附带一个 **APP 模拟器**，可以模拟 DG-LAB APP 端行为，无需真实设备即可测试。 + +### 测试步骤 + +``` +终端1: 启动 WebSocket 服务端 + cd socket/v2/backend && npm start + +终端2: 启动 APP 模拟器 + python dglab_app_simulator.py ws://127.0.0.1:9999 + +终端3: 在 xiaoli-cli 中测试 + /dglab connect ws://127.0.0.1:9999 + /dglab qr + /dglab strength a + 10 + /dglab pulse a 30 50 5 + /dglab status +``` + +模拟器会： +- 模拟 APP 连接 WebSocket 服务端 +- 自动接受配对请求 +- 实时显示收到的强度/波形指令 +- 回传强度状态（和真机行为一致） + ## 参考资料 - [DG-LAB 官方开源仓库](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE) diff --git a/dglab_ws/dglab_app_simulator.py b/dglab_ws/dglab_app_simulator.py new file mode 100644 index 0000000..e88b781 --- /dev/null +++ b/dglab_ws/dglab_app_simulator.py @@ -0,0 +1,266 @@ +""" +DG-Lab 郊狼 APP 模拟器 +模拟 DG-LAB APP 端连接 WebSocket 服务端，用于测试 dglab_ws 插件 + +用法: + python dglab_app_simulator.py [ws_url] + + ws_url: WebSocket 服务端地址，默认 ws://127.0.0.1:9999 + +流程: + 1. 先启动 WebSocket 服务端 (socket/v2/backend) + 2. 运行本模拟器，它会自动连接并等待配对 + 3. 在 xiaoli-cli 中用 /dglab connect 连接同一服务端 + 4. 用 /dglab qr 生成二维码，本模拟器会自动响应绑定请求 + 5. 开始测试各种命令 +""" + +import json +import sys +import time +import uuid +import threading + +try: + import websocket +except ImportError: + print("❌ 需要安装 websocket-client: pip install websocket-client") + sys.exit(1) + + +class DGLabAppSimulator: + """模拟 DG-LAB APP""" + + def __init__(self, ws_url="ws://127.0.0.1:9999"): + self.ws_url = ws_url + self.ws = None + self.target_id = "" + self.client_id = "" # 绑定的第三方终端 ID + self.connected = False + self.paired = False + + # 模拟设备状态 + self.strength_a = 0 + self.strength_b = 0 + self.limit_a = 100 + self.limit_b = 100 + + def start(self): + """启动模拟器""" + self.target_id = str(uuid.uuid4()) + connect_url = f"{self.ws_url}/{self.target_id}" + + print(f"🐺 DG-Lab APP 模拟器") + print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━") + print(f" 服务端: {self.ws_url}") + print(f" targetId: {self.target_id}") + print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━") + print(f"⏳ 正在连接...") + + self.ws = websocket.WebSocketApp( + connect_url, + on_message=self._on_message, + on_error=self._on_error, + on_close=self._on_close, + on_open=self._on_open + ) + + # 心跳线程 + threading.Thread(target=self._heartbeat, daemon=True).start() + + try: + self.ws.run_forever() + except KeyboardInterrupt: + print("\n👋 已退出") + + def _on_open(self, ws): + self.connected = True + print(f"✅ 已连接到 WebSocket 服务端") + print(f"📱 等待第三方终端发起配对...") + print(f" (在 xiaoli-cli 中执行: /dglab connect {self.ws_url})") + print() + + def _on_close(self, ws, code, reason): + self.connected = False + self.paired = False + print(f"🔌 连接已断开 (code={code})") + + def _on_error(self, ws, error): + print(f"❌ 错误: {error}") + + def _on_message(self, ws, raw): + try: + data = json.loads(raw) + except json.JSONDecodeError: + return + + msg_type = data.get("type", "") + msg = data.get("message", "") + cid = data.get("clientId", "") + tid = data.get("targetId", "") + + if msg_type == "bind": + self._handle_bind(data) + + elif msg_type == "msg": + self._handle_msg(data) + + elif msg_type == "heartbeat": + # 回复心跳 + pass + + elif msg_type == "error": + print(f" ❌ 服务端错误: {msg}") + + elif msg_type == "break": + self.paired = False + self.client_id = "" + print(f" ⚠️ 第三方终端已断开") + + def _handle_bind(self, data): + """处理绑定请求""" + msg = data.get("message", "") + cid = data.get("clientId", "") + tid = data.get("targetId", "") + + if msg == "200": + self.paired = True + self.client_id = cid + print(f" ✅ 配对成功! 与终端 {cid[:8]}... 绑定") + print(f" 现在可以在 xiaoli-cli 中使用 /dglab 命令了") + print() + self._print_status() + elif msg == "400": + print(f" ❌ 配对失败: ID 已被绑定") + else: + # 新的绑定请求，APP 端自动接受 + print(f" 📨 收到绑定请求: clientId={cid[:8]}...") + # 回复绑定成功 + bind_ok = { + "type": "bind", + "clientId": cid, + "targetId": self.target_id, + "message": "200" + } + ws.send(json.dumps(bind_ok)) + self.paired = True + self.client_id = cid + print(f" ✅ 已自动接受配对") + + def _handle_msg(self, data): + """处理来自第三方终端的指令""" + msg = data.get("message", "") + + if msg.startswith("strength-"): + self._handle_strength(msg) + elif msg.startswith("pulse-"): + self._handle_pulse(msg) + elif msg.startswith("clear-"): + self._handle_clear(msg) + else: + print(f" 📨 收到指令: {msg}") + + def _handle_strength(self, msg): + """处理强度指令: strength-通道+模式+数值""" + try: + parts = msg.replace("strength-", "").split("+") + channel = int(parts[0]) # 1=A, 2=B + mode = int(parts[1]) # 0=减少, 1=增加, 2=设为 + value = int(parts[2]) + + ch_name = "A" if channel == 1 else "B" + + if mode == 0: # 减少 + if channel == 1: + self.strength_a = max(0, self.strength_a - value) + else: + self.strength_b = max(0, self.strength_b - value) + print(f" 📉 通道{ch_name} 强度 -{value}") + + elif mode == 1: # 增加 + if channel == 1: + self.strength_a = min(self.limit_a, self.strength_a + value) + else: + self.strength_b = min(self.limit_b, self.strength_b + value) + print(f" 📈 通道{ch_name} 强度 +{value}") + + elif mode == 2: # 设为 + if channel == 1: + self.strength_a = min(self.limit_a, value) + else: + self.strength_b = min(self.limit_b, value) + print(f" 🎯 通道{ch_name} 强度设为 {value}") + + # 回传当前强度 + self._send_strength_feedback() + + except (ValueError, IndexError) as e: + print(f" ❌ 强度指令解析失败: {msg} ({e})") + + def _handle_pulse(self, msg): + """处理波形指令: pulse-A:[hex_data,...]""" + try: + content = msg.replace("pulse-", "") + ch = content[0] # A or B + json_str = content[2:] # 跳过 "A:" + data = json.loads(json_str) + count = len(data) + print(f" 🌊 通道{ch} 收到 {count} 条波形数据") + if count > 0: + sample = data[0] + print(f" 示例: {sample}") + except Exception as e: + print(f" ❌ 波形指令解析失败: {msg} ({e})") + + def _handle_clear(self, msg): + """处理清空指令: clear-1 / clear-2""" + channel = msg.replace("clear-", "") + if channel == "1": + print(f" 🗑️ 通道A 波形队列已清空") + elif channel == "2": + print(f" 🗑️ 通道B 波形队列已清空") + + def _send_strength_feedback(self): + """回传当前强度状态""" + if not self.paired or not self.ws: + return + feedback = { + "type": "msg", + "clientId": self.client_id, + "targetId": self.target_id, + "message": f"strength-{self.strength_a}+{self.strength_b}+{self.limit_a}+{self.limit_b}" + } + try: + self.ws.send(json.dumps(feedback)) + self._print_status() + except Exception: + pass + + def _print_status(self): + """打印当前状态""" + print(f" ┌─────────────────────────┐") + print(f" │ 通道A: {self.strength_a:>3}/{self.limit_a:<3} │") + print(f" │ 通道B: {self.strength_b:>3}/{self.limit_b:<3} │") + print(f" └─────────────────────────┘") + + def _heartbeat(self): + """心跳""" + while self.connected: + time.sleep(60) + if self.ws and self.connected: + try: + hb = { + "type": "heartbeat", + "clientId": self.client_id, + "targetId": self.target_id, + "message": "ping" + } + self.ws.send(json.dumps(hb)) + except Exception: + break + + +if __name__ == "__main__": + url = sys.argv[1] if len(sys.argv) > 1 else "ws://127.0.0.1:9999" + sim = DGLabAppSimulator(url) + sim.start()