import time  # 导入时间模块，用于控制延时import vosk  # 导入 Vosk 语音识别库import json  # 导入 JSON 解析库import subprocess  # 用于调用外部进程（如语音合成）import ollama  # 导入 Ollama，用于聊天 AI 处理import asyncio  # 导入 asyncio 以支持异步操作import pyaudio  # 导入 PyAudio 处理音频输入import threading  # 导入多线程模块import re  # 导入正则表达式模块，用于文本处理from concurrent.futures import ThreadPoolExecutor  # 用于异步任务的线程池from board import *  # 导入 board 库，用于 I2C 设备的引脚定义import busio  # 用于 I2C 通信from PIL import Image, ImageDraw, ImageFont  # 用于处理 OLED 显示图像import adafruit_ssd1306  # 用于控制 SSD1306 OLED 屏幕

# 选择合适的字体font_path = "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc"  # OLED 屏幕显示的字体路径font = ImageFont.truetype(font_path, 16)  # 设置字体大小适应屏幕
# 初始化 I2C 设备i2c = busio.I2C(I2C2_SCL, I2C2_SDA)  # 通过 I2C2_SCL 和 I2C2_SDA 初始化 I2C 通信disp = adafruit_ssd1306.SSD1306_I2C(128, 64, i2c)  # 通过 I2C 初始化 128x64 分辨率的 OLED 屏幕
# 清空屏幕disp.fill(0)disp.show()
# 获取屏幕宽高width = disp.widthheight = disp.height
# 创建一个空白图像用于绘制image = Image.new("1", (width, height))draw = ImageDraw.Draw(image)
# 设置 Vosk 语音识别模型路径,根据自己的路径设置model_path = "/home/linaro/vosk-model-small-cn-0.22"model = vosk.Model(model_path)  # 加载 Vosk 语音识别模型recognizer = vosk.KaldiRecognizer(model, 16000)  # 以 16kHz 采样率初始化语音识别器
# 记录会话历史conversation_history = [    {"role": "system", "content": "你是嘉立创EDA-小嘉，是一个运行在泰山派上的离线本地大模型语音助手。"},]MAX_HISTORY_LENGTH = 1  # 限制会话历史的最大条数asr_running = False  # 语音识别时的动画状态控制变量

def show_on_oled(text):    draw.rectangle((0, 0, width, height), outline=0, fill=0)  # 清除屏幕    draw.text((0, 0), text[:8], font=font, fill=255)  # 仅显示前8个字符    disp.image(image)  # 更新屏幕内容    disp.show()

def draw_face(mouth_state=0):     """显示不同表情的脸部动画"""    draw.rectangle((0, 0, width, height), outline=0, fill=0)  # 清空屏幕
    # 绘制眼睛    draw.ellipse((32, 15, 40, 25), outline=255, fill=255)  # 左眼    draw.ellipse((88, 15, 96, 25), outline=255, fill=255)  # 右眼
    # 嘴巴动画    if mouth_state == 0:        draw.line((50,50, 78, 50), fill=255, width=2)  # 直线嘴巴    elif mouth_state == 1:        draw.arc((50, 40, 78, 60), start=0, end=180, fill=255)  # 微笑嘴巴    elif mouth_state == 2:        draw.ellipse((58, 50, 70, 60), outline=255, fill=255)  # 张嘴嘴巴
    disp.image(image)  # 更新屏幕内容    disp.show()

def asr_animation():    global asr_running    frame = 0    while asr_running:        draw_face(frame % 3)  # 按 0、1、2 的顺序切换嘴巴形态        frame += 1        time.sleep(0.2)  # 控制动画速度

def recognize_speech(p):    global asr_running    stream = p.open(format=pyaudio.paInt16,                    channels=1,                    rate=16000,                    input=True,                    frames_per_buffer=20480)    stream.start_stream()    print("正在识别语音...")    asr_running = True  # 启动动画    animation_thread = threading.Thread(target=asr_animation)  # 创建动画线程    animation_thread.start()    try:        while True:            data = stream.read(2048, exception_on_overflow=False)  # 读取音频数据            if recognizer.AcceptWaveform(data):  # 处理音频流                result = recognizer.Result()  # 获取识别结果                result_json = json.loads(result)  # 解析 JSON 数据                if "text" in result_json:                    text = result_json['text']  # 提取识别文本                    print(f"识别到的文字: {text}")                    asr_running = False  # 停止动画                    animation_thread.join()  # 等待动画线程结束                    draw_face(1)  # 识别完成后显示微笑表情                    time.sleep(1)
                    # 释放音频资源                    stream.stop_stream()                    stream.close()                    return text    except IOError:        pass
    # 释放资源    asr_running = False    animation_thread.join()    stream.stop_stream()    stream.close()

def clean_text(text):    return re.sub(r"[^\u4e00-\u9fa5，。！？]", "", text)

async def generate_and_play_text(input_text):    draw_face(1)  # 显示微笑表情    time.sleep(1)    conversation_history.append({"role": "user", "content": input_text})  # 记录用户输入    response = ollama.chat(model="qwen2:0.5b", messages=conversation_history, stream=True)  # AI 生成回答    generated_text = ""    for chunk in response:        if 'message' in chunk and 'content' in chunk['message']:            raw_text = chunk['message']['content']            clean_tts_text = clean_text(raw_text)  # 清理文本            generated_text += raw_text            show_on_oled(raw_text)            print(f"当前生成: {generated_text}")            if clean_tts_text:  # 确保有中文内容才播放
                # 显示在 OLED 上                subprocess.run(['espeak-ng', '-v', 'zh', clean_tts_text])  # 语音朗读
    conversation_history.append({"role": "assistant", "content": generated_text})  # 记录 AI 回应    draw_face(1)  # 结束后显示微笑    time.sleep(1)

async def main():    loop = asyncio.get_event_loop()  # 获取事件循环    executor = ThreadPoolExecutor(max_workers=2)  # 线程池用于并发任务    p = pyaudio.PyAudio()  # 创建 pyaudio 实例    while True:        recognized_text = await loop.run_in_executor(executor, recognize_speech, p)  # 识别语音        await generate_and_play_text(recognized_text)  # 生成回答并朗读    p.terminate()  # 释放音频资源if __name__ == "__main__":    asyncio.run(main())  # 运行主程序