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_runningframe = 0while asr_running:draw_face(frame % 3)# 按 0、1、2 的顺序切换嘴巴形态frame += 1time.sleep(0.2)# 控制动画速度

def recognize_speech(p):global asr_runningstream = 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 textexcept IOError:pass
# 释放资源asr_running = Falseanimation_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_textshow_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())# 运行主程序