python调用麦克风和扬声器，并调用阿里云实时语音转文字

import time
import queue
import sounddevice as sd
import numpy as np
import nls
import sys# 阿里云配置信息
URL = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
TOKEN = "XXXX"  # 参考https://help.aliyun.com/document_detail/450255.html获取token
APPKEY = "XXXX"  # 获取Appkey请前往控制台：https://nls-portal.console.aliyun.com/applist# Queue to hold the recorded audio data
audio_queue = queue.Queue()
speaker_queue = queue.Queue()# Callback function to capture audio data from microphone
def audio_callback(indata, frames, time, status):if status:print(status, file=sys.stderr)audio_queue.put(indata.copy())# Callback function to capture audio data from speaker
def speaker_callback(indata, frames, time, status):if status:print(status, file=sys.stderr)speaker_queue.put(indata.copy())class RealTimeSpeechRecognizer:def __init__(self, url, token, appkey, name):self.url = urlself.token = tokenself.appkey = appkeyself.name = nameself.transcriber = Noneself.__initialize_transcriber()def __initialize_transcriber(self):self.transcriber = nls.NlsSpeechTranscriber(url=self.url,token=self.token,appkey=self.appkey,on_sentence_begin=self.on_sentence_begin,on_sentence_end=self.on_sentence_end,on_start=self.on_start,on_result_changed=self.on_result_changed,on_completed=self.on_completed,on_error=self.on_error,on_close=self.on_close,callback_args=[self.name])self.transcriber.start(aformat="pcm", enable_intermediate_result=True,enable_punctuation_prediction=True, enable_inverse_text_normalization=True)def send_audio(self, audio_data):if self.transcriber:self.transcriber.send_audio(audio_data)def stop_transcription(self):if self.transcriber:self.transcriber.stop()def on_sentence_begin(self, message, *args):print(f"{self.name} 过程中的对话: {message}")def on_sentence_end(self, message, *args):print(f"{self.name} 完整的对话: {message}")def on_start(self, message, *args):print(f"{self.name} Start: {message}")def on_result_changed(self, message, *args):print(f"{self.name} Result changed: {message}")def on_completed(self, message, *args):print(f"{self.name} Completed: {message}")def on_error(self, message, *args):print(f"{self.name} Error: {message}")def on_close(self, *args):print(f"{self.name} Closed: {args}")# 调用阿里云的语音转文字的接口
def recognize_speech(audio_data, recognizer):audio_data = np.concatenate(audio_data)recognizer.send_audio(audio_data.tobytes())# Start the audio stream and process audio data
def start_audio_stream(mic_recognizer, speaker_recognizer, speaker_device_index):with sd.InputStream(callback=audio_callback, channels=1, samplerate=16000, dtype='int16') as mic_stream, \sd.InputStream(callback=speaker_callback, channels=1, samplerate=16000, dtype='int16',device=speaker_device_index) as spk_stream:print("Recording audio... Press Ctrl+C to stop.")mic_audio_buffer = []speaker_audio_buffer = []try:while True:while not audio_queue.empty():mic_audio_buffer.append(audio_queue.get())while not speaker_queue.empty():speaker_audio_buffer.append(speaker_queue.get())if len(mic_audio_buffer) >= 10:recognize_speech(mic_audio_buffer, mic_recognizer)mic_audio_buffer = []  # Clear buffer after sendingif len(speaker_audio_buffer) >= 10:recognize_speech(speaker_audio_buffer, speaker_recognizer)speaker_audio_buffer = []  # Clear buffer after sendingtime.sleep(0.1)except KeyboardInterrupt:print("Stopping audio recording.")mic_recognizer.stop_transcription()speaker_recognizer.stop_transcription()if __name__ == "__main__":speaker_device_index = 8  # 使用 pulse 设备（索引 8）来捕获扬声器输出mic_recognizer = RealTimeSpeechRecognizer(URL, TOKEN, APPKEY, "调解人员说：")speaker_recognizer = RealTimeSpeechRecognizer(URL, TOKEN, APPKEY, "当事人说：")start_audio_stream(mic_recognizer, speaker_recognizer, speaker_device_index)

这段代码的主要功能是从麦克风和扬声器捕获音频数据，并实时将这些音频数据发送到阿里云进行语音转文字处理。以下是代码的详细解释：

导入模块

import time
import queue
import sounddevice as sd
import numpy as np
import nls
import sys

• time 和 queue：用于处理时间和线程安全的队列。
• sounddevice (sd)：用于从麦克风和扬声器捕获音频数据。
• numpy (np)：用于处理音频数据数组。
• nls：阿里云的语音服务库。
• sys：用于处理系统相关的操作，如错误输出。

阿里云配置信息

URL = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
TOKEN = "016ca1620aff421da8fac81b9fb52dc5"
APPKEY = "ahS8ZDaimkpWALHi"

这些变量存储了阿里云语音服务的配置信息，包括服务的 URL、令牌（TOKEN）和应用密钥（APPKEY）。

音频数据队列

audio_queue = queue.Queue()
speaker_queue = queue.Queue()

用于存储从麦克风和扬声器捕获的音频数据。

回调函数

音频数据回调函数

def audio_callback(indata, frames, time, status):if status:print(status, file=sys.stderr)audio_queue.put(indata.copy())

这个回调函数会在音频数据可用时被调用，将捕获到的麦克风音频数据复制到 audio_queue 队列中。

扬声器数据回调函数

def speaker_callback(indata, frames, time, status):if status:print(status, file=sys.stderr)speaker_queue.put(indata.copy())

这个回调函数会在音频数据可用时被调用，将捕获到的扬声器音频数据复制到 speaker_queue 队列中。

RealTimeSpeechRecognizer 类

class RealTimeSpeechRecognizer:def __init__(self, url, token, appkey, name):self.url = urlself.token = tokenself.appkey = appkeyself.name = nameself.transcriber = Noneself.__initialize_transcriber()

初始化函数，接收 URL、TOKEN 和 APPKEY，并调用内部函数 __initialize_transcriber 初始化语音转文字服务。

def __initialize_transcriber(self):self.transcriber = nls.NlsSpeechTranscriber(url=self.url,token=self.token,appkey=self.appkey,on_sentence_begin=self.on_sentence_begin,on_sentence_end=self.on_sentence_end,on_start=self.on_start,on_result_changed=self.on_result_changed,on_completed=self.on_completed,on_error=self.on_error,on_close=self.on_close,callback_args=[self.name])self.transcriber.start(aformat="pcm", enable_intermediate_result=True,enable_punctuation_prediction=True, enable_inverse_text_normalization=True)

初始化语音转文字服务并配置相关回调函数。

def send_audio(self, audio_data):if self.transcriber:self.transcriber.send_audio(audio_data)def stop_transcription(self):if self.transcriber:self.transcriber.stop()

用于发送音频数据到阿里云并停止转录。

回调函数

def on_sentence_begin(self, message, *args):print(f"{self.name} 过程中的对话: {message}")def on_sentence_end(self, message, *args):print(f"{self.name} 完整的对话: {message}")def on_start(self, message, *args):print(f"{self.name} Start: {message}")def on_result_changed(self, message, *args):print(f"{self.name} Result changed: {message}")def on_completed(self, message, *args):print(f"{self.name} Completed: {message}")def on_error(self, message, *args):print(f"{self.name} Error: {message}")def on_close(self, *args):print(f"{self.name} Closed: {args}")

这些函数在语音转文字服务的不同事件发生时被调用，打印相关信息。

处理音频数据

def recognize_speech(audio_data, recognizer):audio_data = np.concatenate(audio_data)recognizer.send_audio(audio_data.tobytes())

将音频数据连接成一个数组并发送给阿里云语音转文字服务。

开始音频流并处理音频数据

def start_audio_stream(mic_recognizer, speaker_recognizer, speaker_device_index):with sd.InputStream(callback=audio_callback, channels=1, samplerate=16000, dtype='int16') as mic_stream, \sd.InputStream(callback=speaker_callback, channels=1, samplerate=16000, dtype='int16',device=speaker_device_index) as spk_stream:print("Recording audio... Press Ctrl+C to stop.")mic_audio_buffer = []speaker_audio_buffer = []try:while True:while not audio_queue.empty():mic_audio_buffer.append(audio_queue.get())while not speaker_queue.empty():speaker_audio_buffer.append(speaker_queue.get())if len(mic_audio_buffer) >= 10:recognize_speech(mic_audio_buffer, mic_recognizer)mic_audio_buffer = []  # Clear buffer after sendingif len(speaker_audio_buffer) >= 10:recognize_speech(speaker_audio_buffer, speaker_recognizer)speaker_audio_buffer = []  # Clear buffer after sendingtime.sleep(0.1)except KeyboardInterrupt:print("Stopping audio recording.")mic_recognizer.stop_transcription()speaker_recognizer.stop_transcription()

这个函数打开音频输入流，开始录音并处理音频数据，将其发送到阿里云进行转录。当用户按下 Ctrl+C 时，停止录音并结束转录。

主程序入口

if __name__ == "__main__":speaker_device_index = 8  # 使用 pulse 设备（索引 8）来捕获扬声器输出mic_recognizer = RealTimeSpeechRecognizer(URL, TOKEN, APPKEY, "调解人员说：")speaker_recognizer = RealTimeSpeechRecognizer(URL, TOKEN, APPKEY, "当事人说：")start_audio_stream(mic_recognizer, speaker_recognizer, speaker_device_index)

创建两个 RealTimeSpeechRecognizer 实例，一个用于麦克风，一个用于扬声器。然后开始录音和处理音频数据。