feat: 添加 DG-Lab APP 模拟器，支持无真机测试
- dglab_app_simulator.py 模拟 APP 端行为 - 自动连接、自动接受配对 - 实时显示强度/波形指令 - 回传强度状态（与真机一致） - README 添加测试教程
This commit is contained in:
xiaoli-cli bot committed
1 parent
742b9e4b41
commit
6a56b262d4
2 files changed
+293
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user