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