xiaoli-cli bot
6a56b262d4
- dglab_app_simulator.py 模拟 APP 端行为 - 自动连接、自动接受配对 - 实时显示强度/波形指令 - 回传强度状态（与真机一致） - README 添加测试教程
267 lines
8.7 KiB
Python
267 lines
8.7 KiB
Python
"""
|
|
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()
|