物联网项目记录（2）树莓派语音与视觉基础

需要：

树莓派4B*1，4g内存至少（2G就不能推广到边缘计算了，8G其实最好，不过给学校省经费用的4G），32GB存储卡。
语音部分需要树莓派免驱扬声器，那种USB直连的。
视觉部分需要USB摄像头，最好也是免驱的。

语音部分#

扬声器测试#

在树莓派中新建项目文件夹，因为树莓派主要是下位机功能，所以我这里叫client_rsp。

1
mkdir client_rsp
2
cd client_rsp

这里推荐从终端换成vscode的remote ssh连接，写代码比较方便。

配置环境：

1
sudo apt-get update
2
sudo apt-get install python3-pyaudio

创建虚拟环境：

1
python3 -m venv --system-site-packages venv
2
source venv/bin/activate

简单将一个audio.wav传到这个目录下，然后运行下面代码，可以测试扬声器是否正常工作：

1
import pyaudio
2
import wave
3

4
def play_audio(filename):
5
    with wave.open(filename, 'rb') as wf:
6
        p = pyaudio.PyAudio()
7
        stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
8
                        channels=wf.getnchannels(),
9
                        rate=wf.getframerate(),
10
                        output=True)
11

12
        chunk_size = 1024
13

14
        data = wf.readframes(chunk_size)
15
        while data:
16
            stream.write(data)
17
            data = wf.readframes(chunk_size)
18

19
        stream.stop_stream()
20
        stream.close()
21
        p.terminate()
22

23
play_audio('audio.wav')

有声音就是正常。

树莓派音量调节#

1
# 查询音量
2
amixer get Master
3

4
# 调节音量
5
amixer set Master 50%

自定义语音输出#

这里希望实现一个输入任意文字，能输出语音的接口，也就是语音合成。这个语音合成可以部署在树莓派或者服务器上，还是比较好部署的，但是我这里直接用了阿里云dashscope的api，一是比较方便，二是合成效果更好（模型好，更自然）。

获取api key在这里：API-KEY的获取

配置阿里云DashScope的SDK：

1
pip install dashscope

同时配置一下环境：

1
pip install paho-mqtt
2
pip install pyyaml

和前面一样，我们在项目文件夹下新建一个config.yaml，用来填写配置：

1
dashscope:
2
  api-key: "<api-key>"

下面代码实现了一个自定义语音输出：

1
import time
2
import dashscope
3
from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
4
from dashscope.audio.tts import ResultCallback, SpeechSynthesizer, SpeechSynthesisResult
5
import pyaudio
6

7
import os
8
import sys
9
import yaml
10

11
with open('config.yaml', 'r', encoding='utf-8') as file:
12
    config = yaml.safe_load(file)
13
dashscope.api_key = config['dashscope']['api-key']
14

15
sport_active = False
16

17
class Callback(ResultCallback):
18
    _player = None
19
    _stream = None
20

21
    def on_open(self):
22
        print('Speech synthesizer is opened.')
23
        current_timestamp = int(time.time())
24

25
        self._player = pyaudio.PyAudio()
26
        self._stream = self._player.open(
27
            format=pyaudio.paInt16,
28
            channels=1,
29
            rate=48000,
30
            output=True)
31

32
    def on_complete(self):
33
        print('Speech synthesizer is completed.')
34

35
    def on_error(self, response: SpeechSynthesisResponse):
36
        print('Speech synthesizer failed, response is %s' % (str(response)))
37

38
    def on_close(self):
39
        print('Speech synthesizer is closed.')
40
        self._stream.stop_stream()
41
        self._stream.close()
42
        self._player.terminate()
43

44
    def on_event(self, result: SpeechSynthesisResult):
45
        if result.get_audio_frame() is not None:
46
            print('audio result length:', sys.getsizeof(result.get_audio_frame()))
47
            self._stream.write(result.get_audio_frame())
48

49
        if result.get_timestamp() is not None:
50
            print('timestamp result:', str(result.get_timestamp()))
51

52
callback = Callback()
53

54

55
while True:
56
    message = input("请输入要合成的文本：")
57
    if message:
58
        SpeechSynthesizer.call(model='sambert-zhiya-v1', text=message, sample_rate=48000, format='pcm', callback=callback)
59
    else:
60
        time.sleep(1)

MQTT远程语音控制#

自定义输入文字还不够，我们的项目是服务器得到运动评价后，将评语传递给树莓派，让它说话。因此要实现一个mqtt的远程语音控制。

配置文件config.yaml中加上MQTT的信息：

1
dashscope:
2
  api-key: "<api-key>"
3
  model: "<model>"
4

5
server:
6
  address: "<ip>"         # MQTT服务器地址
7
  port: <port>                        # MQTT服务器端口

然后是运行的文件voice.py，实现了通过mqtt远程发送后输出声音：

1
import time
2
import os
3
import sys
4
import yaml
5

6
with open('config.yaml', 'r', encoding='utf-8') as file:
7
    config = yaml.safe_load(file)
8

9
# -----------------------------------------------------------------------------
10

11
'''
12
    语音合成
13
'''
14

15
import dashscope
16
from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
17
from dashscope.audio.tts import ResultCallback, SpeechSynthesizer, SpeechSynthesisResult
18
import pyaudio
19

20
dashscope.api_key = config['dashscope']['api-key']
21

22
class Callback(ResultCallback):
23
    _player = None
24
    _stream = None
25

26
    def on_open(self):
27
        print('Speech synthesizer is opened.')
28
        current_timestamp = int(time.time())
29

30
        self._player = pyaudio.PyAudio()
31
        self._stream = self._player.open(
32
            format=pyaudio.paInt16,
33
            channels=1,
34
            rate=48000,
35
            output=True)
36

37
    def on_complete(self):
38
        print('Speech synthesizer is completed.')
39

40
    def on_error(self, response: SpeechSynthesisResponse):
41
        print('Speech synthesizer failed, response is %s' % (str(response)))
42

43
    def on_close(self):
44
        print('Speech synthesizer is closed.')
45
        self._stream.stop_stream()
46
        self._stream.close()
47
        self._player.terminate()
48

49
    def on_event(self, result: SpeechSynthesisResult):
50
        if result.get_audio_frame() is not None:
51
            print('audio result length:', sys.getsizeof(result.get_audio_frame()))
52
            self._stream.write(result.get_audio_frame())
53

54
        if result.get_timestamp() is not None:
55
            print('timestamp result:', str(result.get_timestamp()))
56

57
callback = Callback()
58

59
# -----------------------------------------------------------------------------
60

61
'''
62
    MQTT部分
63
'''
64

65
import paho.mqtt.client as mqtt
66

67
mqtt_server = config['server']['address']
68
mqtt_port = config['server']['port']
69
mqtt_user = "rsp_voicer"
70

71
mqtt_topic = "rsp/voice"
72

73
def on_connect(client, userdata, flags, rc):
74
    print("Connected with result code "+str(rc))
75
    client.subscribe(mqtt_topic)
76
    client.subscribe("sport_cmd")
77

78
def on_message(client, userdata, msg):
79
#    global sport_active
80
    message = msg.payload.decode('utf-8')
81
    print(f"Message arrived on topic: {msg.topic}. Message: {message}")
82

83
    if msg.topic == mqtt_topic:
84
        SpeechSynthesizer.call(model=config['dashscope']['model'], text=message, sample_rate=48000, format='pcm', callback=callback)
85

86
client = mqtt.Client()
87
client.on_connect = on_connect
88
client.on_message = on_message
89

90
client.username_pw_set(mqtt_user)
91
client.connect(mqtt_server, mqtt_port, 60)
92

93
client.loop_forever()

基本上实现了这部分功能。

视觉#

因为初始版本是openmv，并且为了适应低带宽像素调的很低，因此树莓派先实现了一个比较低级的版本。也就是像素进行很大程度的压缩，并且控制了采样率。

配置opencv：

1
sudo apt-get install python3-opencv

略了。

树莓派脚本开机自启动#

在随便哪个路径下创建一个要开机执行的命令，假设是/<路径>/start.sh。写入命令后，要记得给执行权限：

1
chmod +x /<路径>/start.sh

然后编辑这个文件：

1
sudo nano /etc/rc.local

在exit 0前，输入：

1
su pi -c "exec /<路径>/"

这里意思是在pi用户下执行命令，主要是为了避免权限问题导致失败。