使用树莓派做一个智能音箱

功能

监控声音、识别语音、语音合成、音乐播放、聊天、物联开关

需要安装的库

Pyaudio + baidu-aip + MPG321

例如执行pip install baidu-aip即可

设备

设备自行换吧，我没有设备所以随便找了摄像头上的麦克风和耳机

麦克风：摄像头Playstation3 EYE （含麦克风阵列），只要插在USB口上

外放设备：普通耳机，只要插在3.5mm口上

语音唤醒

网上的案例基本都是需要搭配按钮才能监控，不够智能，所以这里实现了达到一定分贝才识别。

语音功能

使用ai.baidu.com提供的语音识别和语音合成

Python SDK：语音识别文档、语音合成文档

语音识别

举例，要对段保存有一段语音的语音文件进行识别：

# 读取文件
def get_file_content(filePath):
    with open(filePath, 'rb') as fp:
        return fp.read()

# 识别本地文件
client.asr(get_file_content('audio.pcm'), 'pcm', 16000, {
    'dev_pid': 1536,
})

语音合成

合成文本长度必须小于1024字节，如果本文长度较长，可以采用多次请求的方式。文本长度不可超过限制

举例，要把一段文字合成为语音文件：

result = client.synthesis('你好百度', 'zh', 1, {
    'vol': 5,
})

# 识别正确返回语音二进制 错误则返回dict 参照下面错误码
if not isinstance(result, dict):
    with open('auido.mp3', 'wb') as f:
        f.write(result)

音乐播放

我是先下载好一些歌曲，然后随机播放，使用mpg321播放。

聊天

智能机器人API接口说明
支持功能：天气、翻译、藏头诗、笑话、歌词、计算、域名信息/备案/收录查询、IP查询、手机号码归属、人工智能聊天。
接口地址：http://api.qingyunke.com/api.php?key=free&appid=0&msg=关键词
　　　　　key　固定参数free
　　　　　appid 设置为0，表示智能识别，可忽略此参数
　　　　　msg　关键词，请参考下方参数示例，该参数可智能识别，该值请经过 urlencode 处理后再提交
返回结果：{"result":0,"content":"内容"}
　　　　　result　状态，0表示正常，其它数字表示错误
　　　　　content　信息内容

物联开关

目前是控制GPIO输出来开关LED灯，可以自行编写外接别的东西。

最终代码

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import pyaudio
import wave
import numpy as np
from aip import AipSpeech
import json
import re
import RPi.GPIO as GPIO
import requests
from urllib import parse
import os
import random

LED_PIN_NUM = 21  # LED 的长脚接的GPIO

""" 你的 APPID AK SK """
APP_ID = 'appid'
API_KEY = 'apikey'
SECRET_KEY = 'secretkey'

client = AipSpeech(APP_ID, API_KEY, SECRET_KEY)

def get_file_content(filePath):
    with open(filePath, 'rb') as fp:
        return fp.read()

def asr(filePath):
    try:
        msg = client.asr(get_file_content(filePath), 'wav', 16000, {
            'dev_pid': 1536,
        })
        print(msg)
        if 'result' in msg:
            return msg['result']
        else:
            return ['']
    except Exception:
        return ['']
    

def tts(text):
    print('tts', text)
    result = client.synthesis(text, 'zh', 1, {
        'vol': 5,
        'per': 4
    })
    # 识别正确返回语音二进制 错误则返回dict 参照下面错误码
    if not isinstance(result, dict):
        with open('cacheTTS.mp3', 'wb') as f:
            f.write(result)
            f.close()
            os.system('mpg321 cacheTTS.mp3')
            print('tts ok')
            #os.system('rm -rf cacheTTS.mp3')
            return 0
    print('tts no')
    return -1

def Monitor():
    CHUNK = 512
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
    RECORD_SECONDS = 5
    WAVE_OUTPUT_FILENAME = "cache.wav"
    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK)
    print("开始缓存录音")
    frames = []
    flag = False
    while (True):
        try:
            print('begin ')
            if flag == False:
                frames.clear()
            for i in range(0, 50):
                data = stream.read(CHUNK)
                frames.append(data)
            audio_data = np.fromstring(data, dtype=np.short)
            large_sample_count = np.sum(audio_data > 800)
            temp = np.max(audio_data)
            print('max：', temp, ' sum:', large_sample_count)
            if temp > 2000 or large_sample_count > 300:
                print("检测到信号")
                flag = True
                continue
            elif flag:
                break
        except Exception:
            print('err')
            return ['']
        
    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()
    print('end')
    asrMsg = asr(WAVE_OUTPUT_FILENAME)
    return asrMsg

def checkRe(msgAsr, regex):
    matchObj = re.match(regex, msgAsr)
    if matchObj:
        return True
    else:
        return False

def AI_openled():
    print('led on')
    GPIO.output(LED_PIN_NUM, GPIO.HIGH)

def AI_offled():
    print('led off')
    GPIO.output(LED_PIN_NUM, GPIO.LOW)

def AI_music():
    print('music')
    rd = random.randint(0, 5)
    os.system('mpg321 ./music/'+ str(rd) +'.mp3')
    return 1

def talk(msg):
    global request
    values = {'key': 'free', 'appid': '0', 'msg': msg}
    data = parse.urlencode(values)

    response = requests.get("http://api.qingyunke.com/api.php?" + data)
    rdata = json.loads(response.text)

    if 'content' in rdata:
        return rdata['content'].replace(u'{br}', '\n').replace('提示：按分类看笑话请发送“笑话分类”', '')
    else:
        return ''

if __name__ == '__main__':
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(LED_PIN_NUM, GPIO.OUT)
    GPIO.output(LED_PIN_NUM, GPIO.LOW)
    while True:
        
        msgAsr = Monitor()
        print('result', msgAsr)

        if msgAsr[0] != '' :
            regex_OnpenLED = r'.*开.*灯.*'
            regex_OnpenLED = r'.*要有光.*'
            regex_OffLED = r'.*关.*灯.*'
            regex_music = r'.*歌|(音乐).*'
            if(checkRe(msgAsr[0], regex_OnpenLED)):
                AI_openled()
            elif(checkRe(msgAsr[0], regex_OnpenLED)):
                AI_openled()
            elif(checkRe(msgAsr[0], regex_OffLED)):
                AI_offled()
            elif(checkRe(msgAsr[0], regex_music)):
                AI_music()
            else:
                msgTalk = talk(msgAsr[0])
                if msgTalk != '':
                    tts(msgTalk)

        
        client = AipSpeech(APP_ID, API_KEY, SECRET_KEY)

    GPIO.cleanup()

演示：

自己用树莓派做一个智能音箱/语音助手