Initial commit: BitFlow MoE - 微型中文 MoE 语言模型 for micro:bit
This commit is contained in:
shuiyu67 committed
commit
586917e75c
14 files changed
+2522
No files matched your search
+184
@@ -0,0 +1,184 @@
|
||||
"""
|
||||
串口发送工具: 通过串口向 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()
|
||||
Reference in new issue
Block a user