林间

使用 Python 把树莓派改造成一个语音助手

2018/03/26 python raspberrypi linux 共 7234 字，约 21 分钟

语音助手已经不是什么新事物了。就在两三年前，语音助手的使用体验还不是那么好，尝尝鲜后也就没用过了。但最近发现不管是微软的 Cortana、苹果的 Siri，还是一些不怎么有名气的，例如 MIUI 的小爱同学等，使用体验真的改善了很多，确确实实能带来一些方便了。

随着各种云服务、API 的面世，语音方面的云服务可以说是十分健全了。你是否也想过自己动手搭建一个语音助手系统呢？本文将总结使用 Python 把树莓派（3 代 b 型）改造成一个简易语音助手的基本流程。

概述

这次要做的说白了，就是把各种云服务、API 串起来，并不涉及任何核心技术、算法的实现，望知悉。

这次将要使用到的服务包括：

为了实现这个语音助手系统，需要完成的工作每一个都不难，但数量稍多了些。以下是涉及到的一些博客：

后文在介绍各部分的具体实现时，只附上代码和进行一些必要的说明，详细内容还需要参考相应博客。

各部分的实现

由于整个项目用到的服务比较多，而且各部分的分工很明显，所以选择各部分分别用一个 python 程序来实现，最后再用一个程序整合在一起的方式。

录音

参考：树莓派学习手记——使用 Python 录音

笔者采用了 “按住按钮进行录音” 的操作方式，如下图所示接线。如果你手头上没有按钮或觉得这么做不方便，可以修改代码改成 “按回车键开始 / 结束录音” 之类的操作方式。

另外，树莓派的板载 3.5mm 耳机接口是不带语音输入功能的，所以你需要另外购买 USB 声卡。

  • 文件 rec.py
import RPi.GPIO as GPIO
import pyaudio
import wave
import os
import sys

def rec_fun():
	# 隐藏错误消息，因为会有一堆 ALSA 和 JACK 错误消息，但其实能正常录音
	os.close(sys.stderr.fileno())

	BUTT = 26	# 开始录音的按钮：一边接 GPIO26，一边接地
	GPIO.setmode(GPIO.BCM)
	# 设 GPIO26 脚为输入脚，电平拉高，也就是说 26 脚一旦读到低电平，说明按了按钮
	GPIO.setup(BUTT, GPIO.IN, pull_up_down = GPIO.PUD_UP)

	# wav 文件是由若干个 CHUNK 组成的，CHUNK 我们就理解成数据包或者数据片段。
	CHUNK = 512
	FORMAT = pyaudio.paInt16  # pyaudio.paInt16 表示我们使用量化位数 16 位来进行录音
	RATE = 44100  # 采样率 44.1k，每秒采样 44100 个点。
	WAVE_OUTPUT_FILENAME = "/home/pi/chat/command.wav"
	print('请按住按钮开始录音...')
	GPIO.wait_for_edge(BUTT, GPIO.FALLING)

	# To use PyAudio, first instantiate PyAudio using pyaudio.PyAudio(), which sets up the portaudio system.
	p = pyaudio.PyAudio()
	stream = p.open(format = FORMAT,
					channels = 1,	# cloud speecAPI 只支持单声道
					rate = RATE,
					input = True,
					frames_per_buffer = CHUNK)
	print("录音中...")

	frames = []
	# 按住按钮录音，放开时结束
	while GPIO.input(BUTT) == 0:
		data = stream.read(CHUNK)
		frames.append(data)
	print("录音完成，输出文件：" + WAVE_OUTPUT_FILENAME + '\n')
	stream.stop_stream()
	stream.close()
	p.terminate()

	wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
	wf.setnchannels(1)
	wf.setsampwidth(p.get_sample_size(FORMAT))	# Returns the size (in bytes) for the specified sample format.
	wf.setframerate(RATE)
	wf.writeframes(b''.join(frames))
	wf.close()

	return

# 可以直接运行 rec.py 进行测试，同时保证该文件 import 时不会自动运行
if __name__ == '__main__':
	rec_fun()

语音识别

参考：

使用 Google 云计算引擎实现科学上网

在 Windows 命令行、Linux 终端使用代理

在 Python 中使用谷歌 Cloud Speech API 将语音转换为文字（另一种方案）

由于某些原因，笔者选择了使用谷歌 Cloud Speech API 进行语音识别。既然要用谷歌的服务，自然就涉及到了科学上网、代理、谷歌云平台的使用，如果不想这么折腾，完全可以用国内的讯飞、百度来实现。

另外，API KEY 之类的字符串在这里删除了，还请麻烦修改代码加上你自己申请的 API KEY。

  • 文件 speech_api.py
import json
import urllib.request
import base64

def wav_to_text():
	api_url = "https://speech.googleapis.com/v1beta1/speech:syncrecognize?key = 替换成你的 API 密钥"
	print('语音文件编码中...')
	audio_file = open('/home/pi/chat/command.wav', 'rb')
	audio_b64str = (base64.b64encode(audio_file.read())).decode()
	audio_file.close()

	voice = {
	  "config":
	  {
		"languageCode": "cmn-Hans-CN"
	  },

	  "audio":
	  {
		"content": audio_b64str
	  }
	}
	voice = json.dumps(voice).encode('utf8')
	print('编码完成。正在上传语音...')
	req = urllib.request.Request(api_url, data=voice, headers={'content-type': 'application/json'})
	response = urllib.request.urlopen(req)
	response_str = response.read().decode('utf8')
	response_dic = json.loads(response_str)

	if ('results' not in response_dic.keys()):
		print('您录制的文件似乎没有声音，请检查麦克风。')
		return

	transcript = response_dic['results'][0]['alternatives'][0]['transcript']
	confidence = response_dic['results'][0]['alternatives'][0]['confidence']
	result_dic = {'text':transcript ,'confidence':confidence}
	print('识别完成。以字典格式输出：')
	print(result_dic)

	return result_dic

if __name__ == '__main__':
	wav_to_text()

获取文字回答

参考：使用 Python 与图灵机器人聊天

这个获取回答的程序有些粗糙，只能获得普通的文字回答。实际上图灵机器人回复的内容中包括了文字、问题类型甚至情感等信息，还有很多修改的空间。

  • 文件 turing.py
import json
import urllib.request

def chat(question):
	api_url = "http://openapi.tuling123.com/openapi/api/v2"
	text_input = question['text']
	req = {
		"perception":
		{
			"inputText":
			{
				"text": text_input
			},

			"selfInfo":
			{
				"location":
				{
					"city": "上海",
					"province": "上海",
					"street": "文汇路"
				}
			}
		},

		"userInfo":
		{
			"apiKey": "替换成你的 APIKEY",
			"userId": "用户参数"
		}
	}
	# 将字典格式的 req 转为 utf8 编码的字符串
	req = json.dumps(req).encode('utf8')

	print('\n' + '正在调用图灵机器人 API...')
	http_post = urllib.request.Request(api_url, data=req, headers={'content-type': 'application/json'})
	response = urllib.request.urlopen(http_post)

	print('得到回答，输出为字典格式：')
	response_str = response.read().decode('utf8')
	response_dic = json.loads(response_str)
	intent_code = response_dic['intent']['code']

	# 返回网页类的输出方式
	if(intent_code == 10023):
		results_url = response_dic['results'][0]['values']['url']
		results_text = response_dic['results'][1]['values']['text']
		answer = {"code": intent_code, "text": results_text, "url":results_url}
		print(answer)
		return(answer)
	# 一般的输出方式
	else:
		results_text = response_dic['results'][0]['values']['text']
		answer = {"code": intent_code, "text": results_text}
		print(answer)
		return(answer)

if __name__ == '__main__':
	eg_question = {'text': '今天是几号', 'confidence': 0.9}
	chat(eg_question)

读出回答（语音合成）

参考：在 Python 中使用科大讯飞 Web API 进行语音合成

笔者在使用讯飞 Web API 时，该服务才开放不到一周，难免以后该 API 会有所变动，如有问题建议查阅官方文档。

  • 文件 tts.py
import base64
import json
import time
import hashlib
import urllib.request
import urllib.parse
import os

def speak(text_content):
	# API 请求地址、API KEY、APP ID 等参数，提前填好备用
	api_url = "http://api.xfyun.cn/v1/service/v1/tts"
	API_KEY = "替换成你的 APIKEY"
	APP_ID = "替换成你的 APPID"
	AUE = "lame"

	# 构造输出音频配置参数
	Param = {
		"auf": "audio/L16;rate=16000",	# 音频采样率
		"aue": AUE,	# 音频编码，raw(生成 wav) 或 lame(生成 mp3)
		"voice_name": "xiaoyan",
		"speed": "50",	# 语速 [0,100]
		"volume": "10",	# 音量 [0,100]
		"pitch": "50",	# 音高 [0,100]
		"engine_type": "aisound"	# 引擎类型。aisound（普通效果），intp65（中文），intp65_en（英文）
	}
	# 配置参数编码为 base64 字符串，过程：字典→明文字符串→utf8 编码→base64(bytes)→base64 字符串
	Param_str = json.dumps(Param)	# 得到明文字符串
	Param_utf8 = Param_str.encode('utf8')	# 得到 utf8 编码 (bytes 类型)
	Param_b64 = base64.b64encode(Param_utf8)	# 得到 base64 编码 (bytes 类型)
	Param_b64str = Param_b64.decode('utf8')	# 得到 base64 字符串

	# 构造 HTTP 请求的头部
	time_now = str(int(time.time()))
	checksum = (API_KEY + time_now + Param_b64str).encode('utf8')
	checksum_md5 = hashlib.md5(checksum).hexdigest()
	header = {
		"X-Appid": APP_ID,
		"X-CurTime": time_now,
		"X-Param": Param_b64str,
		"X-CheckSum": checksum_md5
	}

	# 构造 HTTP 请求 Body
	body = {
		"text": text_content
	}
	body_urlencode = urllib.parse.urlencode(body)
	body_utf8 = body_urlencode.encode('utf8')

	# 发送 HTTP POST 请求
	print('\n' + "正在调用科大讯飞语音合成 API...")
	req = urllib.request.Request(api_url, data=body_utf8, headers=header)
	response = urllib.request.urlopen(req)

	# 读取结果
	response_head = response.headers['Content-Type']
	if(response_head == "audio/mpeg"):
		out_file = open('/home/pi/chat/answer.mp3', 'wb')
		data = response.read() # a `bytes` object
		out_file.write(data)
		out_file.close()
		print('得到结果，输出文件: /home/pi/chat/answer.mp3')
	else:
		print(response.read().decode('utf8'))

	# 播放音频
	print("播放音频中...")
	print("以下均为 mplayer 的输出内容 \ n")
	os.system("mplayer -ao alsa:device=hw=1.0 /home/pi/chat/answer.mp3")

	return

if __name__ == '__main__':
	eg_text_content = "苟利国家生死以，岂因祸福避趋之"
	speak(eg_text_content)

整合 & 测试

现在，你的项目文件夹中应该有这些 python 代码文件：

接下来我们只需要将他们整合在一起运行。

  • 文件 combine.py
# 这些 import 进来的模块是同目录下的 py 文件
import rec	# rec.py 负责录制 wav 音频
import speech_api	# speech_api.py 负责 wav 转文字
import turing	# turing.py 负责获得图灵机器人的文字回答
import tts	# tts.py 负责读出回答

rec.rec_fun()	# 录制音频
recognize_result = speech_api.wav_to_text()	# 识别语音，返回值是字典格式，包含文字结果和信心
turing_answer = turing.chat(recognize_result)	# 得到图灵的回答，返回值仍是字典格式
tts.speak(turing_answer['text'])

如果一切顺利的话，实际运行效果如下： 树莓派_语音助手_youku

小结

语音助手这边的工作算是告一段落了，结果小结却不知道怎么写了。不管怎么说，很开心最后能得到实际的结果，做的过程中也有一些脑洞想要继续扩展，过段时间应该还会继续！

做这个项目的过程中，项目外的收获或许比这个项目本身还要多。这段时间从很多博客、论坛得到了数不尽的帮助，国内的、国外的、中文的、英文的、日文的都有，深深地感受到了互联网共享精神的力量，这也是促使我开始写这些文章的原因。那么，最后还是说一句：感谢你阅读文章！

文档信息

Search

    Table of Contents