185 lines
5.7 KiB
Python
185 lines
5.7 KiB
Python
"""
|
|
串口发送工具: 通过串口向 micro:bit 发送对话命令
|
|
用法:
|
|
python chat_moe.py # 自动查找串口
|
|
python chat_moe.py COM3 # 指定串口
|
|
"""
|
|
import sys
|
|
import glob
|
|
import os
|
|
|
|
try:
|
|
import serial
|
|
import serial.tools.list_ports
|
|
except ImportError:
|
|
print("[ERROR] 需要安装 pyserial:")
|
|
print(" pip install pyserial")
|
|
sys.exit(1)
|
|
|
|
|
|
def find_serial():
|
|
"""自动查找 micro:bit 串口"""
|
|
ports = serial.tools.list_ports.comports()
|
|
for p in ports:
|
|
# 检查是否是 micro:bit (VID:PID = 0d28:0204)
|
|
if p.vid == 0x0d28 and p.pid == 0x0204:
|
|
return p.device
|
|
# 也检查名称包含 MICROBIT 的
|
|
if "MICROBIT" in p.description.upper():
|
|
return p.device
|
|
return None
|
|
|
|
|
|
def send_command(port, text):
|
|
"""发送文本并转换为ID命令"""
|
|
VOCAB_CHARS = (
|
|
"猫狗鸟鱼猪牛马鸡虫羊兔鸭鹅"
|
|
"坐跑飞游看吃睡走听来叫爬跳唱画读做"
|
|
"在上里到"
|
|
"的是好了"
|
|
"吗什么怎哪"
|
|
"你我他她们它"
|
|
"蓝绿红黄白"
|
|
"天地水火花草树木日月星云春夏秋冬"
|
|
"大小多高远快"
|
|
"有没会不"
|
|
"人头脚手眼腿"
|
|
"垫肉书子"
|
|
"一三只两"
|
|
"热冷落开"
|
|
"自然动植物质色样颜得都"
|
|
"你好吗谢谢再见早晚安静"
|
|
"很棒爱喜欢乐伤心气"
|
|
"名字叫谁和问答应"
|
|
"聊天话今天说"
|
|
"客用"
|
|
"bit"
|
|
"\n"
|
|
)
|
|
CH2I = {c: i for i, c in enumerate(VOCAB_CHARS)}
|
|
ctx_ids = [str(CH2I[c]) for c in text if c in CH2I]
|
|
if not ctx_ids:
|
|
print(f"[warn] 无法识别: '{text}'")
|
|
return
|
|
cmd = "GEN:" + ",".join(ctx_ids) + "|15"
|
|
print(f" 发送: {cmd}")
|
|
port.write((cmd + "\n").encode())
|
|
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("BitFlow MoE v4 对话工具")
|
|
print("=" * 60)
|
|
|
|
# 串口
|
|
if len(sys.argv) > 1:
|
|
port_name = sys.argv[1]
|
|
else:
|
|
print("\n[search] 正在查找 micro:bit 串口 ...")
|
|
port_name = find_serial()
|
|
if port_name is None:
|
|
print("[ERROR] 未找到 micro:bit 串口!")
|
|
print(" 请确保:")
|
|
print(" 1. micro:bit 已通过 USB 连接")
|
|
print(" 2. main.py 已在 micro:bit 上运行")
|
|
print(" 3. 如已知串口, 可指定: python chat_moe.py COM3")
|
|
return
|
|
|
|
print(f"\n[connect] 打开串口: {port_name} @115200")
|
|
try:
|
|
ser = serial.Serial(port_name, 115200, timeout=2)
|
|
except Exception as e:
|
|
print(f"[ERROR] 无法打开串口: {e}")
|
|
return
|
|
print(" 连接成功!")
|
|
print()
|
|
|
|
# 等待 ready
|
|
ready = False
|
|
for _ in range(20):
|
|
if ser.in_waiting:
|
|
line = ser.readline().decode('utf-8', errors='ignore').strip()
|
|
print(f"< {line}")
|
|
if "READY" in line:
|
|
ready = True
|
|
print()
|
|
break
|
|
if not ready:
|
|
print("[warn] 未收到 READY 信号，micro:bit 可能还在启动...")
|
|
|
|
# 常用短语列表
|
|
phrases = [
|
|
"你好", "你好吗", "谢谢", "再见", "早安", "晚安",
|
|
"你叫什么名字", "你喜欢什么", "你喜欢猫吗", "你是谁",
|
|
"你会什么", "你开心吗", "你在做什么", "今天好吗",
|
|
"什么会飞", "什么会跑", "什么会游", "什么吃草",
|
|
"天是什么颜色", "猫吃什么", "猫是什么",
|
|
]
|
|
|
|
print("=" * 60)
|
|
print("常用短语 (输入序号发送, 或直接输入文字, 输入 q 退出):")
|
|
print("=" * 60)
|
|
for i, p in enumerate(phrases):
|
|
print(f" {i:2d}. {p}")
|
|
print()
|
|
|
|
try:
|
|
while True:
|
|
try:
|
|
query = input(">>> ").strip()
|
|
except EOFError:
|
|
break
|
|
|
|
if query.lower() == 'q':
|
|
break
|
|
|
|
if query == '':
|
|
continue
|
|
|
|
# 检查是否是序号
|
|
try:
|
|
idx = int(query)
|
|
if 0 <= idx < len(phrases):
|
|
print(f"\n 选择: {phrases[idx]}")
|
|
send_command(ser, phrases[idx])
|
|
# 读取输出
|
|
for _ in range(5):
|
|
if ser.in_waiting:
|
|
line = ser.readline().decode('utf-8', errors='ignore').strip()
|
|
if line.startswith('O:'):
|
|
ids = [int(x) for x in line[2:].split(',') if x.strip()]
|
|
text = "".join(VOCAB_CHARS.get(i, '?') for i in ids)
|
|
print(f"\n 回答: {text}")
|
|
print()
|
|
break
|
|
else:
|
|
print(f" {line}")
|
|
continue
|
|
except ValueError:
|
|
pass
|
|
|
|
# 直接输入文字
|
|
send_command(ser, query)
|
|
# 读取输出
|
|
for _ in range(5):
|
|
if ser.in_waiting:
|
|
line = ser.readline().decode('utf-8', errors='ignore').strip()
|
|
if line.startswith('O:'):
|
|
ids = [int(x) for x in line[2:].split(',') if x.strip()]
|
|
text = "".join(VOCAB_CHARS.get(i, '?') for i in ids)
|
|
print(f"\n 回答: {text}")
|
|
print()
|
|
break
|
|
else:
|
|
print(f" {line}")
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
ser.close()
|
|
print("\n[done] 已断开连接")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|