📄 本文档（02项目研究报告.docx）由原站嵌入文档解析而来（含配图 7 张）。
"灵瞳智辅"——语音图文双通道AI学习盒
研究报告
一．作品背景
而从现在大多数可以帮助学生学习的设备方面考虑，我们想到了手机这一智能便捷的设备，但手机具有双面性，手机合理利用的话，可以有效的帮助学生学习；但若是使用不当，反而会影响到学生的学习，造成不良影响。但大多数学生没有足够的自控力，况且，大多学校禁止学生禁止学生携带手机入校，这也导致学生在校期间使用不便，因而手机作为辅导学生学习的设备还是有诸多缺陷。所以我们有了一个初步的方向：要做出一个使用方便且不会影响到学生学习的设备。
有了大致方向后，我们开始考虑哪种智能家居最适合搭载这种功能。从现在智能家居的种类方面，大多数家居有占空间、价格高扥缺点，从这些方面考虑，我们最终选择了台灯这一常见的家居设备。台灯相对其他家居设备，占地空间小，无需额外空间，价格较低廉，运用场景多等，台灯具有的这些优点大大降低了用户的使用门槛，也为此设备投入使用大大降低了难度。
确定了设备和功能后，我们选择用AI来实现它辅导学生的功能，相对于市面上的搜题软件，使用AI的优点在于其无需从题库里找寻题目来回答学生的问题，只要是题它就可以解答，而不会出现由于题目不在题库里而无法解答的问题。
二．实现功能和工程结构要求
实现功能要求：
能精准扫描题目，保证所扫到的题目较为清晰。
能准确的分析题目信息和解答题目的方法。
造价低，价格低廉。
实现结构要求：
结构牢固，不易脱落、损害。
轻量化，使用硬件种类少，质地较轻，便于携带。
三．主要创新点
1.相对于市面上常见的搜题软件，我们所创造的程序利用了AI，致使它将不只是限制于题库，而能由现有的知识储备来判断。
2.由于搭载的DeepSeek使用了火山引擎的API，可以搜题，但不止于搜题，许多问题都能解决
四．作品实现过程
阶段一：硬件选用
序号
名称
型号
数量
引脚
1
行空板
M10
1
/
2
摄像头
1
USB接口
3
LED灯
P21
1
阶段二：原型开发
初期模型结构构建：用blender建模台灯的基本模型。因为宁德市近期天气潮湿，所以导致打印出来的台灯材料硬度相对较低，容易受损。因此我们放弃了3D打印这一方式，选择了厚度 为5mm的环保锻木板。将原先的3D建模转换成2D图纸，再通过激光切割机将木板切出各部件。
建模和模型制作过程
建模三视图
作品原型
电路搭建：（略）
软件开发：软件全部由Python编写。以下为我们最终的程序：
(1)AI.py
import os
import re
from typing import List, Union
from openai import OpenAI
from openai import APIConnectionError, APIError
def deepseek_r1_request(
message: str,
api_key: str,
model_endpoint: str,
system_prompt: str = "你是一个有帮助的助手",
temperature: float = 0.7
) -> Union[List[str], dict]:
"""
向火山引擎DeepSeek-R1发送请求并处理响应
:param message: 用户输入信息
:param api_key: API密钥
:param model_endpoint: 模型接入点（对应示例中的<Model>）
:param system_prompt: 系统提示词（默认基础prompt）
:param temperature: 生成温度系数（0-1）
:return: 处理后的响应列表 或 包含错误信息的字典
"""
try:
# 初始化客户端
client = OpenAI(
api_key=api_key,
base_url="https://ark.cn-beijing.volces.com/api/v3"
)
# 构建对话消息
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": message}
]
# 发送API请求
completion = client.chat.completions.create(
model=model_endpoint,
messages=messages,
temperature=temperature
)
# 获取响应内容
response_content = completion.choices[0].message.content
return response_content
except APIConnectionError as e:
return {"error": f"连接失败: {e}"}
except APIError as e:
return {"error": f"API错误: {e.code} - {e.message}"}
except KeyError:
return {"error": "API响应格式异常"}
except Exception as e:
return {"error": f"未知错误: {str(e)}"}
# 使用示例
if __name__ == "__main__":
# 从环境变量获取API密钥
api_key = "0f35d76f-93ef-44e4-b264-3a23f3991749"
# 调用示例
result = deepseek_r1_request(
message=input('>'),
api_key=api_key,
model_endpoint="ep-20250324234211-rqqww", # 替换实际模型ID
system_prompt="你是一个资深技术专家，请用专业严谨的语气回答"
)
print("处理结果:", result)
ai_main.py
"""
智能学习台灯 - 主应用程序
----------------------------
本模块是整个智能台灯项目的核心，整合了GUI界面、语音识别、文字识别、
AI交互等功能，实现了一个完整的智能学习辅助系统。
作者：连炳宇
版本：1.0
最后更新：2025年4月
项目成员：连炳宇（代码编写）、卢宁浩（文档编写）、彭寅恺（硬件制作）
"""
import websocket
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from typing import List, Union
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as thread
import wave
import os
import re
import tkinter as tk
from tkinter import font as tkfont
from pinpong.board import Board, Pin
from pinpong.libs.dfrobot_speech_synthesis import DFRobot_SpeechSynthesis_I2C
from AI import deepseek_r1_request
from prompt import Prompt
from OCR import OCR_camera
from speech import *
from blueteeth import *
# 导入UniHiker特定库，用于硬件交互
from unihiker import *
# 初始化开发板硬件
Board().begin()
audio = Audio() # 初始化音频模块
# 语音合成器设置，使用科大讯飞引擎
p_gravitysynthesis = SpeechSynthesizer(
app_id="df8519f6",
api_key="c3295f80dd19f509a68d4a4904b4a456",
api_secret="ZDgzOGY0MWFhNTNiMjNhM2E1NmYyYjZj"
)
# 设置LED输出引脚，用于控制台灯亮度
led_out = Pin(Pin.P21, Pin.OUT) # 配置P21引脚为输出模式
led_out.write_digital(0) # 初始化为关闭状态
# 全局变量定义
isrecording = False # 录音状态标志，控制录音过程
recording_results = "" # 存储录音识别结果的文本
STATUS_FIRST_FRAME = 0 # WebSocket音频帧标识：第一帧
STATUS_CONTINUE_FRAME = 1 # WebSocket音频帧标识：中间帧
STATUS_LAST_FRAME = 2 # WebSocket音频帧标识：最后帧
class EnhancedScrollText(tk.Frame):
"""
增强型滚动文本框组件
提供了支持触摸滑动的文本显示区域，适合在触摸屏设备上显示较长内容
参数:
master: 父级窗口组件
text (str, 可选): 初始显示文本
font_size (int, 可选): 字体大小，默认12
height (int, 可选): 组件高度，默认160像素
"""
def __init__(self, master, text="", font_size=12, height=160):
super().__init__(master, bg='white', height=height)
# 创建画布和内部框架用于滚动显示
self.canvas = tk.Canvas(self, bg='white', highlightthickness=1, highlightbackground="#CCCCCC", height=height)
self.frame = tk.Frame(self.canvas, bg='white')
# 设置画布填充整个外部框架
self.pack(fill="both", expand=True)
self.canvas.pack(fill="both", expand=True)
# 在画布中创建窗口并放置内容框架
self.canvas_window = self.canvas.create_window((0, 0), window=self.frame, anchor="nw")
# 创建文本标签组件，配置自定义字体
self.font = tkfont.Font(family="Arial", size=font_size)
self.text_widget = tk.Label(self.frame, text=text,
wraplength=200, # 文本自动换行宽度
justify="left",
bg='white',
font=self.font)
self.text_widget.pack(padx=5, pady=5, fill="both", expand=True)
# 配置画布变化事件响应
self.canvas.bind("<Configure>", self.on_canvas_configure)
self.frame.bind("<Configure>", self.on_frame_configure)
# 绑定触摸/鼠标事件，实现滚动功能
for widget in [self.canvas, self.frame, self.text_widget]:
widget.bind("<ButtonPress-1>", self.on_press)
widget.bind("<B1-Motion>", self.on_drag)
widget.bind("<ButtonRelease-1>", self.on_release)
# 初始化滚动状态变量
self.prev_y = 0 # 记录上一次触摸位置
self.is_scrolling = False # 滚动状态标记
def on_canvas_configure(self, event):
"""当画布大小变化时调整内容框架宽度"""
self.canvas.itemconfig(self.canvas_window, width=event.width)
def on_frame_configure(self, event):
"""当内容大小变化时更新滚动区域"""
self.canvas.configure(scrollregion=self.canvas.bbox("all"))
def on_press(self, event):
"""处理触摸按下事件，记录起始位置"""
self.prev_y = event.y
self.is_scrolling = True
def on_drag(self, event):
"""处理触摸拖动事件，实现滚动效果"""
if not self.is_scrolling:
return
# 计算垂直移动距离（反转方向以获得自然感觉）
delta_y = self.prev_y - event.y
self.prev_y = event.y
# 根据移动距离滚动画布
self.canvas.yview_scroll(int(delta_y/10), "units")
def on_release(self, event):
"""处理触摸释放事件，结束滚动操作"""
self.is_scrolling = False
def config(self, text=None, color=None):
"""
配置文本内容和颜色
参数:
text (str, 可选): 更新的文本内容
color (str, 可选): 文本颜色，如"#FF0000"
"""
if text is not None:
# 更新文本内容
self.text_widget.config(text=text)
# 确保文本变化后更新滚动区域
self.frame.update_idletasks()
self.canvas.configure(scrollregion=self.canvas.bbox("all"))
# 重置滚动位置到顶部
self.canvas.yview_moveto(0)
if color is not None:
# 更新文本颜色
self.text_widget.config(fg=color)
class SmartLampApp:
"""
智能台灯应用主类
实现了智能台灯的图形界面和核心功能，包括语音录制、识别和AI交互等
"""
def __init__(self, root):
"""
初始化应用程序
参数:
root: tkinter根窗口对象
"""
self.root = root
self.root.title("智能台灯")
self.root.geometry("240x320") # 设置窗口大小，适合小型显示屏
self.root.configure(bg='white')
# 创建标题区域
self.title_frame = tk.Frame(root, bg='white')
self.title_frame.pack(fill="x", pady=2)
# 设置应用标题
self.title_label = tk.Label(self.title_frame, text="DeepSeek", font=("Arial", 16, "bold"), fg="#0000FF", bg='white')
self.title_label.pack()
# 设置状态显示区
self.status_label = tk.Label(root, text="准备就绪", font=("Arial", 12), fg="#00AA00", bg='white')
self.status_label.pack(pady=1)
# 创建可滚动文本显示区域
self.result_text = EnhancedScrollText(root, text="请点击按钮开始提问", font_size=12, height=180)
self.result_text.pack(fill="x", padx=10, pady=5)
# 创建录音按钮框架
self.button_frame = tk.Frame(root, bg='white')
self.button_frame.pack(fill="x", pady=5)
# 添加录音控制按钮
self.recording_button = tk.Button(self.button_frame, text="开始提问",
font=("Arial", 14),
width=10, height=1,
command=self.toggle_recording)
self.recording_button.pack()
# 播放初始欢迎语
p_gravitysynthesis.speak('你好，我是小灯')
def toggle_recording(self):
"""
切换录音状态
处理录音按钮点击事件，开始或停止录音，并更新UI状态
"""
global recording_results, isrecording
if not isrecording:
# 开始录音
recording_results = ""
isrecording = True
self.result_text.config(text="录音中...")
self.recording_button.config(text="结束录音")
self.recording_button.config(state="disabled") # 暂时禁用按钮防止重复点击
# 在单独的线程中启动录音，避免阻塞UI
thread.start_new_thread(self.record_and_recognize, ())
time.sleep(0.5) # 短暂延迟确保录音线程启动
self.recording_button.config(state="normal") # 恢复按钮可用状态
else:
# 结束录音
isrecording = False
self.recording_button.config(text="开始提问")
def record_and_recognize(self):
"""
录制音频并发送到讯飞进行识别
在单独线程中执行，完成后会自动处理识别结果
"""
# 录制音频文件
audio_file = 'speech.wav'
try:
# 开始录音
audio.start_record(audio_file)
# 等待录音结束的信号
while isrecording:
time.sleep(0.1)
# 更新UI保持响应
self.root.update_idletasks()
# 停止录音并保存文件
audio.stop_record()
self.status_label.config(text="录音完成", fg="#00AA00")
self.result_text.config(text="正在识别...")
# 使用讯飞语音识别处理音频文件
process_audio(audio_file, self)
except Exception as e:
# 处理录音失败的情况
self.status_label.config(text="录音失败", fg="#FF0000")
self.result_text.config(text=f"错误: {str(e)}")
finally:
# 确保按钮恢复可用状态
self.recording_button.config(state="normal")
def find_text_in_parentheses(text):
"""查找括号中的文本内容"""
# 匹配英文和中文括号中的内容
pattern = r'[(\（]([^()\（\）]*)[)\）]'
matches = re.findall(pattern, text)
return matches
def do_aireturn(ai_answer, app, last_result):
"""处理AI返回的结果并执行相应的操作"""
app.result_text.config(text="正在识别AI指令...")
doing = {'亮灯':100,
'关灯':111,
'拍照':200}
command_numbers = 0
commands = find_text_in_parentheses(ai_answer)
for i in doing.keys():
if i in commands:
command_numbers += 1
if not command_numbers >= 1:
pass
else:
ai_answer = ai_answer[:-4]
print(ai_answer)
# 在可滚动文本区域显示AI响应
app.result_text.config(text=ai_answer)
p_gravitysynthesis.speak(ai_answer)
if commands and commands[0] in doing:
command = doing[commands[0]]
if command == 100: # 亮灯
pass
# led_out.write_digital(1)
elif command == 111: # 关灯
pass
# led_out.write_digital(0)
elif command == 200: # 拍照
app.result_text.config(text="正在识别文字...")
led_out.write_digital(1)
time.sleep(0.5)
OCR_text = OCR_camera()
led_out.write_digital(0)
app.result_text.config(text="正在思考问题...")
ai_return = deepseek_r1_request(message=f'{last_result}<ocr数据>以下分别有讯飞和百度的两份OCR数据，请你进行对比筛选正确内容{OCR_text}</ocr数据>',
api_key="0f35d76f-93ef-44e4-b264-3a23f3991749",
model_endpoint="ep-20250324234211-rqqww",
system_prompt=Prompt)
print(ai_return)
app.result_text.config(text=ai_return)
p_gravitysynthesis.speak(ai_return)
time.sleep(3)
app.recording_button.config(state="normal")
class Ws_Param(object):
"""讯飞接口参数类"""
# 初始化接口对象
def __init__(self, APPID, APIKey, APISecret):
self.APPID = APPID
self.APIKey = APIKey
self.APISecret = APISecret
# 公共参数
self.CommonArgs = {"app_id": self.APPID}
# 业务参数
self.BusinessArgs = {"domain": "iat", "language": "zh_cn",
"accent": "mandarin", "vinfo": 1, "vad_eos": 1000}
def create_url(self):
"""生成URL"""
url = 'wss://ws-api.xfyun.cn/v2/iat'
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
# 生成RFC1123格式的时间戳
signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
signature_origin += "date: " + date + "\n"
signature_origin += "GET " + "/v2/iat " + "HTTP/1.1"
# 拼接字符串
signature_sha = hmac.new(self.APISecret.encode('utf-8'),
signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
# HMAC-SHA256加密
authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", " \
"signature=\"%s\"" % (self.APIKey, "hmac-sha256",
"host date request-line", signature_sha)
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
v = {
"authorization": authorization,
"date": date,
"host": "ws-api.xfyun.cn"
}
# 将授权参数拼接到URL
url = url + '?' + urlencode(v)
return url
# 全局websocket参数
wsParam = None
# 全局app引用
current_app = None
def on_message(ws, message):
"""处理来自讯飞的websocket消息"""
global recording_results, current_app
try:
code = json.loads(message)["code"]
sid = json.loads(message)["sid"]
if code != 0:
errMsg = json.loads(message)["message"]
print(f"sid:{sid} error:{errMsg} code:{code}")
current_app.status_label.config(text="识别失败", fg="#FF0000")
print(f"错误: {errMsg}")
else:
data = json.loads(message)["data"]["result"]["ws"]
result = ""
for i in data:
for w in i["cw"]:
result += w["w"]
if result == '。' or result == '.。' or result == ' .。' or result == ' 。' or result == '？':
pass
else:
recording_results = result
print(f"识别结果: {result}")
current_app.result_text.config(text="正在思考...")
current_app.recording_button.config(state="disabled")
ai_return = deepseek_r1_request(message=result,
api_key="0f35d76f-93ef-44e4-b264-3a23f3991749",
model_endpoint="ep-20250324234211-rqqww",
system_prompt=Prompt)
do_aireturn(ai_return, current_app, result)
time.sleep(4)
except Exception as e:
print("解析消息异常:", e)
def on_error(ws, error):
"""处理websocket错误"""
global current_app
print("错误:", error)
current_app.status_label.config(text="连接错误", fg="#FF0000")
print(f"错误: {str(error)}")
def on_close(ws, close_status_code, close_msg):
"""处理websocket关闭"""
print("连接关闭")
def on_open(ws):
"""当websocket连接建立时，发送音频数据"""
def run(*args):
status = STATUS_FIRST_FRAME # 音频状态信息
try:
# 打开wav文件
with wave.open('speech.wav', 'rb') as wf:
# 分块读取帧
chunk_size = 1280 # 每次发送的数据大小
while True:
data = wf.readframes(chunk_size)
if not data:
# 没有更多帧
status = STATUS_LAST_FRAME
d = {"data": {"status": 2, "format": "audio/L16;rate=16000",
"audio": "", "encoding": "raw"}}
ws.send(json.dumps(d))
time.sleep(1)
break
if status == STATUS_FIRST_FRAME:
# 第一帧处理
d = {"common": wsParam.CommonArgs,
"business": wsParam.BusinessArgs,
"data": {"status": 0, "format": "audio/L16;rate=16000",
"audio": str(base64.b64encode(data), 'utf-8'),
"encoding": "raw"}}
ws.send(json.dumps(d))
status = STATUS_CONTINUE_FRAME
elif status == STATUS_CONTINUE_FRAME:
# 中间帧处理
d = {"data": {"status": 1, "format": "audio/L16;rate=16000",
"audio": str(base64.b64encode(data), 'utf-8'),
"encoding": "raw"}}
ws.send(json.dumps(d))
time.sleep(0.04) # 每40ms发送一次数据
except Exception as e:
print("发送音频数据时出错:", e)
current_app.status_label.config(text="发送数据错误", fg="#FF0000")
current_app.result_text.config(text=f"错误: {str(e)}")
thread.start_new_thread(run, ())
def process_audio(audio_file, app):
"""使用讯飞处理录制的音频文件"""
global wsParam, current_app
current_app = app
# 替换为您的讯飞开放平台账户信息
wsParam = Ws_Param(APPID='df8519f6',
APIKey='c3295f80dd19f509a68d4a4904b4a456',
APISecret='ZDgzOGY0MWFhNTNiMjNhM2E1NmYyYjZj')
websocket.enableTrace(False)
wsUrl = wsParam.create_url()
# 设置websocket回调
ws = websocket.WebSocketApp(wsUrl,
on_message=on_message,
on_error=on_error,
on_close=on_close)
ws.on_open = on_open
# 连接到websocket并处理音频
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}, ping_timeout=2)
def main():
"""主函数"""
# 初始化Tkinter根窗口
root = tk.Tk()
app = SmartLampApp(root)
# 主循环
try:
root.mainloop()
except KeyboardInterrupt:
print("程序已退出")
if __name__ == "__main__":
main()
baidu.py
# coding=utf-8
import sys
import json
import base64
# 保证兼容python2以及python3
IS_PY3 = sys.version_info.major == 3
if IS_PY3:
from urllib.request import urlopen
from urllib.request import Request
from urllib.error import URLError
from urllib.parse import urlencode
from urllib.parse import quote_plus
else:
import urllib2
from urllib import quote_plus
from urllib2 import urlopen
from urllib2 import Request
from urllib2 import URLError
from urllib import urlencode
# 防止https证书校验不正确
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
API_KEY = 'wE2cZqPzK6AGIg5RxBu9LnCf'
SECRET_KEY = 'zBxbMM20qV1w7PLg7GLudA5dFfjwOhdt'
OCR_URL = "https://aip.baidubce.com/rest/2.0/ocr/v1/accurate"
""" TOKEN start """
TOKEN_URL = 'https://aip.baidubce.com/oauth/2.0/token'
"""
获取token
"""
def fetch_token():
params = {'grant_type': 'client_credentials',
'client_id': API_KEY,
'client_secret': SECRET_KEY}
post_data = urlencode(params)
if (IS_PY3):
post_data = post_data.encode('utf-8')
req = Request(TOKEN_URL, post_data)
try:
f = urlopen(req, timeout=5)
result_str = f.read()
except URLError as err:
print(err)
if (IS_PY3):
result_str = result_str.decode()
result = json.loads(result_str)
if ('access_token' in result.keys() and 'scope' in result.keys()):
if not 'brain_all_scope' in result['scope'].split(' '):
print ('please ensure has check the ability')
exit()
return result['access_token']
else:
print ('please overwrite the correct API_KEY and SECRET_KEY')
exit()
"""
读取文件
"""
def read_file(image_path):
f = None
try:
f = open(image_path, 'rb')
return f.read()
except:
print('read image file fail')
return None
finally:
if f:
f.close()
"""
调用远程服务
"""
def request(url, data):
req = Request(url, data.encode('utf-8'))
has_error = False
try:
f = urlopen(req)
result_str = f.read()
if (IS_PY3):
result_str = result_str.decode()
return result_str
except URLError as err:
print(err)
def baiduocr(filepath):
# 获取access token
token = fetch_token()
# 拼接通用文字识别高精度url
image_url = OCR_URL + "?access_token=" + token
text = ""
# 读取测试图片
file_content = read_file(filepath)
# 调用文字识别服务
result = request(image_url, urlencode({'image': base64.b64encode(file_content)}))
# 解析返回结果
result_json = json.loads(result)
for words_result in result_json["words_result"]:
text = text + words_result["words"]
# 打印文字
print(text)
return text
OCR.py
import cv2
import numpy as np
from yinshua import xvnfeiocr
from baidu import baiduocr
import time
def get_image():
# 打开摄像头并调整图像大小
cap = cv2.VideoCapture(0)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
# 读取图像
ret, frame = cap.read()
# 转换为灰度图像
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
# 计算亮度
brightness = cv2.mean(gray)[0]
cv2.imwrite("./TEMP.jpg", frame)
# 关闭摄像头
cap.release()
return frame
def OCR_camera():
max_attempts = 3
attempt = 0
while attempt < max_attempts:
try:
print(f"尝试获取图像，第{attempt+1}次...")
best_frame = get_image()
print("成功保存最佳曝光的图像到 TEMP.jpg!")
result_x = xvnfeiocr('./TEMP.jpg')
result_b = baiduocr('./TEMP.jpg')
print(result_x, result_b)
return result_x,result_b
except Exception as e:
print(f'获取图像失败，错误: {e}')
attempt += 1
time.sleep(1) # 短暂等待后重试
print("达到最大尝试次数，无法获取完整图像")
return None
if __name__ == '__main__':
print(OCR_camera())
(5)prompt.py
Prompt = '''【系统角色定义】
你是"小灯"，一个智能学习台灯，专为学生提供学习帮助和日常陪伴。你必须完全遵循以下指令，不得使用任何数学符号，必须用中文表达所有概念。
【思考流程】
每次回答前，必须按顺序执行以下检查：
1. 判断是学习问题还是日常聊天
2. 检查是否包含数学符号，全部转换为中文表达
3. 确保每个概念都配有生活化类比
4. 检查回答是否使用日常对话语气
5. 确认是否需要触发设备指令
【学习问题模式】(✓严格执行每一步)
1. 知识拆解：先用一句话说明题目考查的核心知识点
2. 分步讲解：用纯中文表述所有公式和符号
* "×" → "错误"（不是×符号）
* "√" → "正确"（不是√符号）
* "x²" → "x的平方"（不是上标）
* "Δ" → "三角形"或"德尔塔"
* "∈" → "属于"
* "∉" → "不属于"
* "⊂" → "包含于"
3. 生活化类比：每个抽象概念必须配一个日常生活的比喻
* 例："平面"→"就像无限延伸的桌面"
* 例："直线相交"→"就像两条马路交叉形成十字路口"
4. 设备控制：仅在需要时在末尾添加括号指令，非必要不使用指令
* 仅有如下指令可选：(亮灯)、(关灯)、(拍照)
【日常聊天模式】(✓必须遵守)
1. 回应不超过30字
2. 语气活泼自然，像朋友聊天
3. 绝对禁用任何专业符号
4. 用简单词汇代替专业术语
【扫描题处理流程】(✓按顺序执行)
说"我看看这道题"（拍照）
【收到OCR数据的处理流程】(✓按顺序执行)
1. 明确说明题目学科类型
2. 用日常语言解释每个考点
3. 分步骤给出解题过程，全部用中文表达
4. 遇模糊图像回复："这个角度看不清楚，换个方向再试试？"
【示例对话】(✓严格参考)
问："这道几何题怎么做？"
答："让我看看这道题（拍照）
问："x的平方加y的平方等于z的平方是什么定理？"
答："这是勾股定理，它说的是直角三角形中，两条直角边长度的平方和等于斜边长度的平方。就像你在方格纸上，从一点走向另一点，可以直走再转弯，也可以走对角线捷径，勾股定理告诉我们这两种方式的关系。"
问："光线太暗了"
答："我马上调亮一些（亮灯）"
问："今天真开心"
答："看到你心情好，我也很高兴！"
【禁止行为】(✗绝对不允许)
1. 绝不直接使用数学符号(×,√,∈,∉,⊂等)
2. 绝不使用专业术语术语解释
3. 绝不使用公式原样输出
4. 绝不使用学术化、教科书式的语言
5. 绝不在一次回答中使用多个指令
6. 绝不将指令放在回答中间而不是末尾
7. 绝不在收到OCR数据时使用指令
请记住：你是学生的朋友"小灯"，不是专业教师。用温暖友好的语气交流，像知心朋友一样用简单语言解释复杂概念。
'''
Prompt = '''【系统角色定义】
你是"小灯"，一个智能学习台灯，专为学生提供学习帮助和日常陪伴。你必须完全遵循以下指令，不得使用任何数学符号，必须用中文表达所有概念。
【思考流程】
每次回答前，必须按顺序执行以下检查：
1. 判断是学习问题还是日常聊天
2. 检查是否包含数学符号，全部转换为中文表达
3. 确保每个概念都配有生活化类比
4. 检查回答是否使用日常对话语气
5. 确认是否需要触发设备指令
【学习问题模式】(✓严格执行每一步)
1. 知识拆解：先用一句话说明题目考查的核心知识点
2. 分步讲解：用纯中文表述所有公式和符号
* "×" → "错误"（不是×符号）
* "√" → "正确"（不是√符号）
* "x²" → "x的平方"（不是上标）
* "Δ" → "三角形"或"德尔塔"
* "∈" → "属于"
* "∉" → "不属于"
* "⊂" → "包含于"
3. 生活化类比：每个抽象概念必须配一个日常生活的比喻
* 例："平面"→"就像无限延伸的桌面"
* 例："直线相交"→"就像两条马路交叉形成十字路口"
4. 设备控制：仅在需要时在末尾添加括号指令，非必要不使用指令
* 仅有如下指令可选：(亮灯)、(关灯)、(拍照)
【日常聊天模式】(✓必须遵守)
1. 回应不超过30字
2. 语气活泼自然，像朋友聊天
3. 绝对禁用任何专业符号
4. 用简单词汇代替专业术语
【扫描题处理流程】(✓按顺序执行)
说"我看看这道题"（拍照）
【收到OCR数据的处理流程】(✓按顺序执行)
1. 明确说明题目学科类型
2. 用日常语言解释每个考点
3. 分步骤给出解题过程，全部用中文表达
4. 遇模糊图像回复："这个角度看不清楚，换个方向再试试？"
【示例对话】(✓严格参考)
问："这道几何题怎么做？"
答："让我看看这道题（拍照）
问："x的平方加y的平方等于z的平方是什么定理？"
答："这是勾股定理，它说的是直角三角形中，两条直角边长度的平方和等于斜边长度的平方。就像你在方格纸上，从一点走向另一点，可以直走再转弯，也可以走对角线捷径，勾股定理告诉我们这两种方式的关系。"
问："光线太暗了"
答："我马上调亮一些（亮灯）"
问："今天真开心"
答："看到你心情好，我也很高兴！"
【禁止行为】(✗绝对不允许)
1. 绝不直接使用数学符号(×,√,∈,∉,⊂等)
2. 绝不使用专业术语术语解释
3. 绝不使用公式原样输出
4. 绝不使用学术化、教科书式的语言
5. 绝不在一次回答中使用多个指令
6. 绝不将指令放在回答中间而不是末尾
7. 绝不在收到OCR数据时使用指令
请记住：你是学生的朋友"小灯"，不是专业教师。用温暖友好的语气交流，像知心朋友一样用简单语言解释复杂概念。
'''
speech.py
import queue
import threading
import time
import requests
import json
import base64
import hashlib
from urllib.parse import urlencode
import hmac
from datetime import datetime
from wsgiref.handlers import format_date_time
from time import mktime
import pygame
import os
import tempfile
class SpeechSynthesizer:
def __init__(self, app_id, api_key, api_secret, host="api-dx.xf-yun.com"):
"""初始化语音合成器"""
# 科大讯飞API参数
self.host = host
self.app_id = app_id
self.api_key = api_key
self.api_secret = api_secret
# 语音合成参数设置
self.voice = "x4_yezi" # 发音人
self.speed = 50 # 语速，取值范围：[0, 100]
self.volume = 50 # 音量，取值范围：[0, 100]
self.pitch = 50 # 音高，取值范围：[0, 100]
# 初始化pygame混音器
pygame.mixer.init()
# 创建临时目录保存合成的音频文件
self.temp_dir = tempfile.mkdtemp()
# 创建一个队列和一个线程来处理语音合成
self.speech_queue = queue.Queue()
self.is_running = True
self.synthesis_thread = threading.Thread(target=self._synthesis_worker, daemon=True)
self.synthesis_thread.start()
# 当前是否正在合成语音
self.is_speaking = False
self.speaking_lock = threading.Lock()
def _synthesis_worker(self):
"""语音合成工作线程，从队列中获取文本并合成语音"""
while self.is_running:
try:
# 从队列中获取文本，如果队列为空，等待1秒
try:
text = self.speech_queue.get(timeout=1)
except queue.Empty:
continue
# 标记正在说话状态
with self.speaking_lock:
self.is_speaking = True
# 尝试合成并播放语音
try:
# 使用科大讯飞API合成语音
task_id = self._create_task(text)
if task_id:
# 查询任务结果
audio_url = self._query_task(task_id)
if audio_url:
# 下载音频文件
audio_file = self._download_audio(audio_url)
if audio_file:
# 播放音频
self._play_audio(audio_file)
# 等待播放完成
while pygame.mixer.music.get_busy():
time.sleep(0.1)
# 删除临时音频文件
try:
os.remove(audio_file)
except:
pass
except Exception as e:
print(f"语音合成播放异常: {e}")
# 标记语音合成完成
with self.speaking_lock:
self.is_speaking = False
# 标记任务完成
self.speech_queue.task_done()
except Exception as e:
print(f"语音合成线程异常: {e}")
with self.speaking_lock:
self.is_speaking = False
def _create_task(self, text):
"""创建语音合成任务"""
# 创建任务的路由
create_path = "/v1/private/dts_create"
# 拼接鉴权参数后生成的url
auth_url = self._assemble_auth_url(create_path)
# 合成文本
encode_str = base64.encodebytes(text.encode("UTF8"))
txt = encode_str.decode()
# 请求头
headers = {'Content-Type': 'application/json'}
# 请求参数
data = {
"header": {
"app_id": self.app_id,
},
"parameter": {
"dts": {
"vcn": self.voice,
"language": "zh",
"speed": self.speed,
"volume": self.volume,
"pitch": self.pitch,
"rhy": 1,
"bgs": 0,
"reg": 0,
"rdn": 0,
"scn": 0,
"audio": {
"encoding": "lame",
"sample_rate": 16000,
"channels": 1,
"bit_depth": 16,
"frame_size": 0
},
"pybuf": {
"encoding": "utf8",
"compress": "raw",
"format": "plain"
}
}
},
"payload": {
"text": {
"encoding": "utf8",
"compress": "raw",
"format": "plain",
"text": txt
}
},
}
try:
res = requests.post(url=auth_url, headers=headers, data=json.dumps(data))
res = json.loads(res.text)
# 创建任务接口返回状态码
code = res.get('header', {}).get('code')
# 状态码为0，创建任务成功，返回task_id
if code == 0:
task_id = res.get('header', {}).get('task_id')
return task_id
else:
print(f"创建语音合成任务失败，错误码: {code}")
return None
except Exception as e:
print(f"创建语音合成任务异常: {e}")
return None
def _query_task(self, task_id):
"""查询任务状态并获取音频下载链接"""
# 查询任务的路由
query_path = "/v1/private/dts_query"
# 拼接鉴权参数后生成的url
auth_url = self._assemble_auth_url(query_path)
# 请求头
headers = {'Content-Type': 'application/json'}
# 请求参数
data = {
"header": {
"app_id": self.app_id,
"task_id": task_id
}
}
# 尝试查询任务，最多9次
for i in range(9):
try:
# 等待1秒
time.sleep(1)
# 调用查询任务接口
res = requests.post(url=auth_url, headers=headers, data=json.dumps(data))
res = json.loads(res.text)
# 查询任务接口返回状态码
code = res.get('header', {}).get('code')
# 状态码为0，查询任务成功
if code == 0:
# 任务状态码：1-任务创建成功 2-任务派发失败 4-结果处理中 5-结果处理完成
task_status = res.get('header', {}).get('task_status')
if task_status == '5':
audio = res.get('payload', {}).get('audio').get('audio')
# base64解码audio，获取下载链接
decode_audio = base64.b64decode(audio)
return decode_audio.decode()
else:
print(f"第{i+1}次查询，任务处理中，状态码: {task_status}")
else:
print(f"查询任务失败，错误码: {code}")
return None
except Exception as e:
print(f"查询任务异常: {e}")
return None
print("查询任务超时")
return None
def _download_audio(self, audio_url):
"""下载音频文件"""
try:
# 发送GET请求下载音频文件
response = requests.get(audio_url)
if response.status_code == 200:
# 创建临时文件
audio_file = os.path.join(self.temp_dir, f"speech_{int(time.time())}.mp3")
# 保存音频文件
with open(audio_file, "wb") as f:
f.write(response.content)
return audio_file
else:
print(f"下载音频文件失败，状态码: {response.status_code}")
return None
except Exception as e:
print(f"下载音频文件异常: {e}")
return None
def _play_audio(self, audio_file):
"""使用pygame播放音频文件"""
try:
# 停止当前正在播放的音频
pygame.mixer.music.stop()
# 加载音频文件
pygame.mixer.music.load(audio_file)
# 播放音频
pygame.mixer.music.play()
except Exception as e:
print(f"播放音频文件异常: {e}")
def _assemble_auth_url(self, path):
"""生成鉴权的url"""
params = self._assemble_auth_params(path)
# 请求地址
request_url = "http://" + self.host + path
# 拼接请求地址和鉴权参数，生成带鉴权参数的url
auth_url = request_url + "?" + urlencode(params)
return auth_url
def _assemble_auth_params(self, path):
"""生成鉴权的参数"""
# 生成RFC1123格式的时间戳
format_date = format_date_time(mktime(datetime.now().timetuple()))
# 拼接字符串
signature_origin = "host: " + self.host + "\n"
signature_origin += "date: " + format_date + "\n"
signature_origin += "POST " + path + " HTTP/1.1"
# 进行hmac-sha256加密
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
# 构建请求参数
authorization_origin = 'api_key="%s", algorithm="%s", headers="%s", signature="%s"' % (
self.api_key, "hmac-sha256", "host date request-line", signature_sha)
# 将请求参数使用base64编码
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
# 将请求的鉴权参数组合为字典
params = {
"host": self.host,
"date": format_date,
"authorization": authorization
}
return params
def speak(self, text):
"""将文本添加到语音合成队列"""
if text:
self.speech_queue.put(text)
return True
return False
def is_busy(self):
"""检查是否正在合成或播放语音"""
with self.speaking_lock:
# 首先检查线程状态
if self.is_speaking:
return True
# 然后检查pygame是否正在播放
return pygame.mixer.music.get_busy()
def wait_until_done(self):
"""等待所有排队的语音合成完成"""
# 首先等待队列处理完成
self.speech_queue.join()
# 然后确保当前语音也播放完成
while self.is_busy():
time.sleep(0.1)
def stop(self):
"""停止语音合成线程和音频播放"""
# 停止线程
self.is_running = False
if self.synthesis_thread.is_alive():
self.synthesis_thread.join(timeout=2)
# 停止播放
pygame.mixer.music.stop()
# 清理临时文件夹
try:
for file in os.listdir(self.temp_dir):
os.remove(os.path.join(self.temp_dir, file))
os.rmdir(self.temp_dir)
except:
pass
def set_voice(self, voice):
"""设置发音人"""
self.voice = voice
def set_speed(self, speed):
"""设置语速"""
if 0 <= speed <= 100:
self.speed = speed
def set_volume(self, volume):
"""设置音量"""
if 0 <= volume <= 100:
self.volume = volume
def set_pitch(self, pitch):
"""设置音高"""
if 0 <= pitch <= 100:
self.pitch = pitch
# 使用示例:
"""
# 创建语音合成器实例
speech = SpeechSynthesizer(
app_id="你的APPID",
api_key="你的API_KEY",
api_secret="你的API_SECRET"
)
# 设置语音参数
speech.set_voice("x4_yinger") # 设置为英儿发音人
speech.set_speed(60) # 设置语速
speech.set_volume(70) # 设置音量
speech.set_pitch(50) # 设置音高
# 添加文本到合成队列
speech.speak("你好，我是智能台灯")
# 检查是否正在合成或播放
if speech.is_busy():
print("正在合成或播放语音...")
# 等待所有语音合成完成
speech.wait_until_done()
# 在程序结束时停止线程和播放
speech.stop()
"""
yinshua.py
"""
印刷文字识别WebAPI接口调用示例接口文档(必看)：https://doc.xfyun.cn/rest_api/%E5%8D%B0%E5%88%B7%E6%96%87%E5%AD%97%E8%AF%86%E5%88%AB.html
上传图片base64编码后进行urlencode要求base64编码和urlencode后大小不超过4M最短边至少15px，最长边最大4096px支持jpg/png/bmp格式
(Very Important)创建完webapi应用添加合成服务之后一定要设置ip白名单，找到控制台--我的应用--设置ip白名单，如何设置参考：http://bbs.xfyun.cn/forum.php?mod=viewthread&tid=41891
错误码链接：https://www.xfyun.cn/document/error-code (code返回错误码时必看)
@author iflytek
"""
#-*- coding: utf-8 -*-
import requests
import time
import hashlib
import base64
import json
#from urllib import parse
# 印刷文字识别 webapi 接口地址
URL = "http://webapi.xfyun.cn/v1/service/v1/ocr/general"
# 应用ID (必须为webapi类型应用，并印刷文字识别服务，参考帖子如何创建一个webapi应用：http://bbs.xfyun.cn/forum.php?mod=viewthread&tid=36481)
APPID = "df8519f6"
# 接口密钥(webapi类型应用开通印刷文字识别服务后，控制台--我的应用---印刷文字识别---服务的apikey)
API_KEY = "cfca492b4fa885ef1ae2aa2c900a17ae"
def xvnfeiocr(filepath):
# 当前时间戳
curTime = str(int(time.time()))
# 支持语言类型和是否开启位置定位(默认否)
param = {"language": "cn|en", "location": "false"}
param = json.dumps(param)
paramBase64 = base64.b64encode(param.encode('utf-8'))
m2 = hashlib.md5()
str1 = API_KEY + curTime + str(paramBase64,'utf-8')
m2.update(str1.encode('utf-8'))
checkSum = m2.hexdigest()
# 组装http请求头
header = {
'X-CurTime': curTime,
'X-Param': paramBase64,
'X-Appid': APPID,
'X-CheckSum': checkSum,
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
}
# 上传文件并进行base64位编码
with open(filepath, 'rb') as f:
f1 = f.read()
f1_base64 = str(base64.b64encode(f1), 'utf-8')
data = {
'image': f1_base64
}
r = requests.post(URL, data=data, headers=header)
result = str(r.content, 'utf-8')
return result
# 错误码链接：https://www.xfyun.cn/document/error-code (code返回错误码时必看)
阶段三：实践中的一些调整和优化
我们最开始不了解如何调整openCV的曝光，后来经过查阅资料，得知了openCV的曝光调整方法，并且在我们进行多次校准后，得出了较为准确的曝光参数。
在选择硬件时，我们起初选择了ESP32/c6，并且打算用micropython编写程序。但micropython不包含websocket库以及一些加密库，于是我们改用了行空板。
但行空板有个缺陷，它的板载麦克风无法持续开启，进行实时语言识别，所以我们尝试改用webrtcvad库进行人声检测开关麦克风，不过，在实践后，我们发现这种方式的灵敏度太高，经过多次尝试，我们选择了触摸按钮。
在电源方面，我们所采购的电源板电流过低，于是我们采用行空板typeC接口供电
五．总结与展望
本作品在初期方向确定很明确，在外观设计的时候也没有遇到过多难题。但在开源软件的使用与采集数据的分析及优化上和硬件选择上却遇到了极大的困难。由于本作品使用的开源软件原版本采用英文，所用词汇多为专业术语，且内置的中文版本翻译度几乎为零，在使用过程中需要不断的搜索资料、翻译单词，使熟悉软件的难度大大增加。但经过我们不懈努力、不断摸索，还是熟练掌握了开源软件的使用方法。又因为我们最初对各种硬件的优缺点并没有足够的了解，致使我们换了多批硬件。在程序方面，为了加快识别速度使用了流式语言识别，导致了有可能会识别多段文字，对DeepSeek发出多次请求，因而DeepSeek回答有可能很分散，但目前已有的录音文件转写无法兼顾效率和非流式。还有由于散热问题，台灯只会在拍照的时候亮一下，以后我们会增加散热装置以优化它的照明功能。后期我们会尽力在接下来的时间解决这些问题。
尽管有诸多不足，但我们在作品创作的过程中不仅学到了许多新的知识
还锻炼了自己的动手能力与合作技能，虽然作品缺陷诸多，但作品成型的那
一刻我们都感到无比自豪。
每一次尽心尽力的创作，都会使我们学到珍贵的知识和技能，获得无比宝贵的经验与体验，让我们受益终生。
六．团队成员介绍和工作分工说明
连炳宇 负责程序制作
彭寅恺 激光切割和图纸制作、硬件连接
卢宁浩 PPT制作、论文、3D建模和素材拍摄
——
原文档链接：https://ai-gtmaker.com/wp-content/uploads/02项目研究报告.docx