1.環(huán)境搭建

#環(huán)境說明：(盡量在ubuntu下使用, 本次代碼均在該環(huán)境下實(shí)現(xiàn))
sudo apt-get install sox  # 安裝語音播放軟件
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simple

2.去訊飛官網(wǎng)申請(qǐng)密鑰

https://www.xfyun.cn/

3.語音識(shí)別（sst）

語音轉(zhuǎn)文字

#! /usr/bin/env python
# -*- coding:utf-8 -*-
'''
1.環(huán)境說明：(盡量在ubuntu下使用, 本次代碼均在該環(huán)境下實(shí)現(xiàn))
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simplepyaudio的安裝（注意區(qū)分不同平臺(tái)）windows pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simpleLinux   sudo apt install python3-pyaudiomac     brew install portaudiopip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple2.密鑰說明：
每一個(gè)賬號(hào)的用額只有500次數(shù)的請(qǐng)求，平時(shí)用足夠使用了。
自行前往去注冊(cè)：https://www.xfyun.cn/'''
import wave
import pyaudio
import websocket
import datetime
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as threadSTATUS_FIRST_FRAME = 0  # 第一幀的標(biāo)識(shí)
STATUS_CONTINUE_FRAME = 1  # 中間幀標(biāo)識(shí)
STATUS_LAST_FRAME = 2  # 最后一幀的標(biāo)識(shí)class Ws_Param(object):# 初始化def __init__(self, APPID, APIKey, APISecret, AudioFile):self.APPID = APPIDself.APIKey = APIKeyself.APISecret = APISecretself.AudioFile = AudioFile# 公共參數(shù)(common)self.CommonArgs = {"app_id": self.APPID}# 業(yè)務(wù)參數(shù)(business)，更多個(gè)性化參數(shù)可在官網(wǎng)查看self.BusinessArgs = {"domain": "iat", "language": "zh_cn", "accent": "mandarin", "vinfo":1,"vad_eos":10000}# 生成urldef create_url(self):url = 'wss://ws-api.xfyun.cn/v2/iat'# 生成RFC1123格式的時(shí)間戳now = datetime.now()date = format_date_time(mktime(now.timetuple()))# 拼接字符串signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"signature_origin += "date: " + date + "\n"signature_origin += "GET " + "/v2/iat " + "HTTP/1.1"# 進(jìn)行hmac-sha256進(jìn)行加密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')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')# 將請(qǐng)求的鑒權(quán)參數(shù)組合為字典v = {"authorization": authorization,"date": date,"host": "ws-api.xfyun.cn"}# 拼接鑒權(quán)參數(shù)，生成urlurl = url + '?' + urlencode(v)# print("date: ",date)# print("v: ",v)# 此處打印出建立連接時(shí)候的url,參考本demo的時(shí)候可取消上方打印的注釋，比對(duì)相同參數(shù)時(shí)生成的url與自己代碼生成的url是否一致# print('websocket url :', url)return url# 收到websocket消息的處理
def on_message(ws, message):try:code = json.loads(message)["code"]sid = json.loads(message)["sid"]if code != 0:errMsg = json.loads(message)["message"]print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))else:data = json.loads(message)["data"]["result"]["ws"]# print(json.loads(message))result = ""for i in data:for w in i["cw"]:result += w["w"]print(result)# print(json.dumps(data, ensure_ascii=False))except Exception as e:print("receive msg,but parse exception:", e)def on_error(ws, error):passdef on_close(ws):print("### closed ###")def on_open(ws):def run(*args):frameSize = 8000  # 每一幀的音頻大小intervel = 0.04  # 發(fā)送音頻間隔(單位:s)status = STATUS_FIRST_FRAME with open(wsParam.AudioFile, "rb") as fp:while True:buf = fp.read(frameSize)# 文件結(jié)束if not buf:status = STATUS_LAST_FRAME# 第一幀處理# 發(fā)送第一幀音頻，帶business 參數(shù)# appid 必須帶上，只需第一幀發(fā)送if status == STATUS_FIRST_FRAME:d = {"common": wsParam.CommonArgs,"business": wsParam.BusinessArgs,"data": {"status": 0, "format": "audio/L16;rate=16000","audio": str(base64.b64encode(buf), 'utf-8'),"encoding": "raw"}}d = json.dumps(d)ws.send(d)status = STATUS_CONTINUE_FRAME# 中間幀處理elif status == STATUS_CONTINUE_FRAME:d = {"data": {"status": 1, "format": "audio/L16;rate=16000","audio": str(base64.b64encode(buf), 'utf-8'),"encoding": "raw"}}ws.send(json.dumps(d))# 最后一幀處理elif status == STATUS_LAST_FRAME:d = {"data": {"status": 2, "format": "audio/L16;rate=16000","audio": str(base64.b64encode(buf), 'utf-8'),"encoding": "raw"}}ws.send(json.dumps(d))time.sleep(1)break# 模擬音頻采樣間隔time.sleep(intervel)ws.close()thread.start_new_thread(run, ())
def record(time):    #錄音程序CHUNK = 1024FORMAT = pyaudio.paInt16CHANNELS = 1RATE = 16000RECORD_SECONDS =timeWAVE_OUTPUT_FILENAME = "./output.pcm"p = pyaudio.PyAudio()stream = p.open(format=FORMAT,channels=CHANNELS,rate=RATE,input=True,frames_per_buffer=CHUNK)print("* recording")frames = []for i in range(0,int(RATE / CHUNK * RECORD_SECONDS)):data = stream.read(CHUNK)frames.append(data)print("* done recording")stream.stop_stream()stream.close()p.terminate()wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')wf.setnchannels(CHANNELS)wf.setsampwidth(p.get_sample_size(FORMAT))wf.setframerate(RATE)wf.writeframes(b''.join(frames))wf.close()if __name__ == "__main__":record(5)time1 = datetime.now()# 這里改成自己的key秘鑰， 每個(gè)人每天只有500的訪問量。wsParam = Ws_Param(APPID='d69356bc', APIKey='c3f938a4da84f7449bd2f958d461e7e1',APISecret='ZjE4ZGE4ZGU4YzViZmFhNTI0ZmYyNTE0',AudioFile=r'./output.pcm')websocket.enableTrace(False)wsUrl = wsParam.create_url()ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)ws.on_open = on_openws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})time2 = datetime.now()

4.語音合成（tts）

文字轉(zhuǎn)語音

#! /usr/bin/env python
# -*- coding:utf-8 -*-
'''
1.環(huán)境說明：(盡量在ubuntu下使用, 本次代碼均在該環(huán)境下實(shí)現(xiàn))
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pygame -i https://pypi.tuna.tsinghua.edu.cn/simplepyaudio的安裝（注意區(qū)分不同平臺(tái)）windows pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simpleLinux   sudo apt install python3-pyaudiomac     brew install portaudiopip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple2.密鑰說明：
每一個(gè)賬號(hào)的用額只有500次數(shù)的請(qǐng)求，平時(shí)用足夠使用了。
自行前往去注冊(cè)：https://www.xfyun.cn/'''
import websocket
import datetime
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as thread
import os
from pydub import AudioSegment
import time
import pygameSTATUS_FIRST_FRAME = 0  # 第一幀的標(biāo)識(shí)
STATUS_CONTINUE_FRAME = 1  # 中間幀標(biāo)識(shí)
STATUS_LAST_FRAME = 2  # 最后一幀的標(biāo)識(shí)class Ws_Param(object):# 初始化def __init__(self, APPID, APIKey, APISecret, Text):self.APPID = APPIDself.APIKey = APIKeyself.APISecret = APISecretself.Text = Text# 公共參數(shù)(common)self.CommonArgs = {"app_id": self.APPID}# 業(yè)務(wù)參數(shù)(business)，更多個(gè)性化參數(shù)可在官網(wǎng)查看self.BusinessArgs = {"aue": "raw", "auf": "audio/L16;rate=16000", "vcn": "xiaofeng", "tte": "utf8"}self.Data = {"status": 2, "text": str(base64.b64encode(self.Text.encode('utf-8')), "UTF8")}# 生成urldef create_url(self):url = 'wss://tts-api.xfyun.cn/v2/tts'# 生成RFC1123格式的時(shí)間戳now = datetime.now()date = format_date_time(mktime(now.timetuple()))# 拼接字符串signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"signature_origin += "date: " + date + "\n"signature_origin += "GET " + "/v2/tts " + "HTTP/1.1"# 進(jìn)行hmac-sha256進(jìn)行加密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')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')# 將請(qǐng)求的鑒權(quán)參數(shù)組合為字典v = {"authorization": authorization,"date": date,"host": "ws-api.xfyun.cn"}# 拼接鑒權(quán)參數(shù)，生成urlurl = url + '?' + urlencode(v)# print("date: ",date)# print("v: ",v)# 此處打印出建立連接時(shí)候的url,參考本demo的時(shí)候可取消上方打印的注釋，比對(duì)相同參數(shù)時(shí)生成的url與自己代碼生成的url是否一致# print('websocket url :', url)return urldef on_message(ws, message):try:message =json.loads(message)code = message["code"]sid = message["sid"]audio = message["data"]["audio"]audio = base64.b64decode(audio)status = message["data"]["status"]# print(message)if status == 2:# print("ws is closed")ws.close()if code != 0:errMsg = message["message"]print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))else:with open('./demo.pcm', 'ab') as f:f.write(audio)except Exception as e:print("receive msg,but parse exception:", e)# 收到websocket錯(cuò)誤的處理
def on_error(ws, error):print("### error:", error)# 收到websocket關(guān)閉的處理
def on_close(ws,arg1,arg2):# print("### closed ###")passdef play_fun():# 加載PCM音頻文件audio = AudioSegment.from_file("demo.pcm", format="raw", sample_width=2, frame_rate=16000, channels=1)# 將音頻轉(zhuǎn)換為MP3格式并播放audio.export("file.mp3", format="mp3")time.sleep(1)# # ==== windows 平臺(tái)使用下面代碼來播放音頻，并將下面的代碼注釋# # 初始化pygame# pygame.mixer.init()# # 加載MP3文件# pygame.mixer.music.load("file.mp3")# # 播放MP3文件# pygame.mixer.music.play()# # 等待音頻播放完畢# while pygame.mixer.music.get_busy():#     pygame.time.Clock().tick(10)# print("------ 播放完畢 ------")# ==== linux 平臺(tái)使用下面代碼來播放音頻，并將上面的代碼注釋file = "file.mp3"os.system('mpg123 '+ file)def tts_fun(string_txr):# 收到websocket連接建立的處理def on_open(ws):def run(*args):d = {"common": wsParam.CommonArgs,"business": wsParam.BusinessArgs,"data": wsParam.Data,}d = json.dumps(d)print("------開始發(fā)送文本數(shù)據(jù)------")ws.send(d)if os.path.exists('./demo.pcm'):os.remove('./demo.pcm')thread.start_new_thread(run, ())# 測(cè)試時(shí)候在此處正確填寫相關(guān)信息即可運(yùn)行# 這里改成自己的key ,每一個(gè)人的用額都是有限的。https://console.xfyun.cn/services/iatwsParam = Ws_Param(APPID='d69356bc', APISecret='c3f938a4da84f7449bd2f958d461e7e1',APIKey='ZjE4ZGE4ZGU4YzViZmFhNTI0ZmYyNTE0',Text=string_txr)websocket.enableTrace(False)wsUrl = wsParam.create_url()ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)ws.on_open = on_openws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})def tts_main(input_txt):tts_fun(input_txt)play_fun()class tts_clss():def __init__(self, input_txt):tts_fun(input_txt)play_fun()if __name__ == "__main__":tts_main("你好，我是機(jī)器人！")

5.USB聲卡可能報(bào)錯(cuò)

設(shè)置完默認(rèn)聲卡后，用Python錄音和播放會(huì)出現(xiàn)一些錯(cuò)誤提示，但發(fā)現(xiàn)錄音和播放都正常，錯(cuò)誤顯示比如這樣：

ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.rear
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.center_lfe
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.side
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.hdmi
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.hdmi
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.modem
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.modem
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.phoneline
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.phoneline
ALSA lib confmisc.c:1281:(snd_func_refer) Unable to find definition ‘defaults.bluealsa.device’
ALSA lib conf.c:4528:(_snd_config_evaluate) function snd_func_refer returned error: No such file or directory
ALSA lib conf.c:4996:(snd_config_expand) Args evaluate error: No such file or directory
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM bluealsa
ALSA lib confmisc.c:1281:(snd_func_refer) Unable to find definition ‘defaults.bluealsa.device’
ALSA lib conf.c:4528:(_snd_config_evaluate) function snd_func_refer returned error: No such file or directory
ALSA lib conf.c:4996:(snd_config_expand) Args evaluate error: No such file or directory
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM bluealsa
connect(2) call to /tmp/jack-1000/default/jack_0 failed (err=No such file or directory)
attempt to connect to server failed

為了不顯示這些錯(cuò)誤可以在Python代碼里p = pyaudio.PyAudio() 之前加上：

os.close(sys.stderr.fileno())