使用 Raspberry Pi 驱动的旋转电话热线进行 ChatGPT 复古

快速阅读: 《XDA-开发人员》消息，基于MIT许可证的开源项目PolluxLabs推出一款基于ChatGPT的转盘电话系统，由弗雷德里克·库姆巴尔茨基开发。该系统通过模拟传统拨号方式与AI交互，支持语音输入和输出，适用于复古科技爱好者。

基于ChatGPT的转盘电话系统
—————————————————————-

https://en.polluxlabs.net
MIT许可证

特此免费授予任何获得该软件及其相关文档文件副本的人，允许他人同样使用该软件，无需任何限制，涵盖但不限于使用、复制、修改、合并、发布、分发、再许可或销售软件副本的权利。前提是以下条件得到满足：

“`python
import time
import threading
from queue import Queue
import os
import sys

class 转盘拨号器:
def __init__(self, openai_client):
self.openai_client = openai_client
self.last_pulse_time = 0
self.running = True
self.pulse_count = 0
self.switch = Switch()
self.audio_manager = AudioManager()
self.speech_recognizer = SpeechRecognizer()

def 启动(self):
# 设置回调函数
self.switch.when_pressed(self._脉冲检测)
self.switch.when_released(self._处理开关释放)
self.switch.when_pressed(self._处理开关按下)

# 初始化为就绪状态
if not self.switch.is_pressed():
# 听筒被拿起
self.audio_manager.开始连续音调()
else:
# 听筒在挂钩上
print(“电话处于空闲状态。请拿起听筒开始。”)
print(“转盘已准备好，请拨号。”)

try:
self._主循环()
except KeyboardInterrupt:
print(“程序终止。”)
self._清理资源()

def _主循环(self):
while self.running:
self._检查号码()
time.sleep(0.1)

def _脉冲检测(self):
if not self.switch.is_pressed():
current_time = time.time()
if current_time – self.last_pulse_time > 抖动时间:
self.pulse_count += 1
self.last_pulse_time = current_time

def _检查号码(self):
if not self.switch.is_pressed() and self.pulse_count > 0:
self.audio_manager.停止连续音调()
time.sleep(1.5) # 等待数字间间隔

if self.pulse_count == 10:
self.pulse_count = 0 # “0” 以 10 次脉冲发送
print(“拨号的服务号码为:”, self.pulse_count)

if self.pulse_count == 1:
self._调用GPT服务()

# 对话结束后恢复拨号音
if not self.switch.is_pressed(): # 只有在听筒未挂断的情况下
self._重置状态()
self.pulse_count = 0

def _调用GPT服务(self):
# 对话上下文历史记录
对话历史记录 = []
第一次交互 = True

# 用于更快的切换
语音识别器 = self.speech_recognizer
回答生成器 = self.response_generator

# 下次录音的准备
下次录音线程 = None
下次录音队列 = Queue()

# 对话循环 – 直到听筒被挂断
while not self.switch.is_pressed():
# 如果有准备好的下次录音线程，使用其结果
if 下次录音线程:
下次录音线程.join()
识别文本 = 下次录音队列.get()
下次录音线程 = None
else:
# 仅在首次迭代或作为回退
print(“正在倾听…” + “（请立即讲话）” if 第一次交互 else “”)
第一次交互 = False

# 开始音频处理
识别文本 = 语音识别器.捕获并转录()
if not 识别文本:
print(“未能识别您的语音内容”)
self.audio_manager.播放文件(AUDIO_FILES[“try_again”])
continue

print(“已理解:”, 识别文本)

# 更新对话上下文历史记录
对话历史记录.append({“角色”: “用户”, “内容”: 识别文本})

# 并行启动下一个录音线程任务
下次录音线程 = threading.Thread(
target=self._后台捕获,
args=(语音识别器, 下次录音队列)
)
下次录音线程.daemon = True
下次录音线程.start()

# 生成回答
回答 = 回答生成器.生成流式响应(识别文本, 对话历史记录)

# 将回答添加到历史记录
对话历史记录.append({“角色”: “助手”, “内容”: 回答})

# 检查听筒是否在此期间被挂断
if self.switch.is_pressed():
break

# 如果到这里，听筒已被挂断
if 下次录音线程 and 下次录音线程.is_alive():
下次录音线程.join(timeout=0.5)

def _后台捕获(self, 识别器, 结果队列):
try:
结果 = 识别器.捕获并转录()
结果队列.put(结果)
except Exception as e:
print(f”后台录音发生错误: {e}”)
结果队列.put(None)

def _重置状态(self):
self.pulse_count = 0
self.audio_manager.停止连续音调()
self.audio_manager.开始连续音调()
print(“转盘已准备好，请拨号。”)

def _处理开关释放(self):
print(“听筒被拿起，系统将重启。”)
self._重启脚本()

def _处理开关按下(self):
print(“听筒被挂断，系统将终止运行。”)
self._清理资源()
self.running = False
# 完整终止前延迟
threading.Timer(1.0, self._重启脚本).start()
return

def _重启脚本(self):
print(“正在重新启动脚本…”)
self.audio_manager.停止连续音调()
os.execv(sys.executable, [‘python’] + sys.argv)

def _清理资源(self):
# 终止音频管理器
self.audio_manager.停止连续音调()
# 如果存在，终止语音识别器
if hasattr(self, ‘speech_recognizer’) and self.speech_recognizer:
self.speech_recognizer.清理()
print(“资源已释放完毕。”)

def 主程序():
# 初始化OpenAI客户端
客户端 = OpenAI(api_key=OPENAI_API_KEY)

# 创建并启动转盘
转盘 = 转盘拨号器(客户端)
转盘.启动()
print(“程序已终止。”)

if __name__ == “__main__”:
主程序()
“`

(以上内容均由Ai生成)