-
Notifications
You must be signed in to change notification settings - Fork 24
/
infer_record.py
82 lines (67 loc) · 1.94 KB
/
infer_record.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import wave
import librosa
import numpy as np
import pyaudio
import tensorflow as tf
# 获取网络模型
model = tf.keras.models.load_model('models/resnet50.h5')
# 录音参数
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
RECORD_SECONDS = 3
WAVE_OUTPUT_FILENAME = "infer_audio.wav"
# 打开录音
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK)
# 读取音频数据
def load_data(data_path):
wav, sr = librosa.load(data_path, sr=16000)
intervals = librosa.effects.split(wav, top_db=20)
wav_output = []
for sliced in intervals:
wav_output.extend(wav[sliced[0]:sliced[1]])
if len(wav_output) < 8000:
raise Exception("有效音频小于0.5s")
wav_output = np.array(wav_output)
ps = librosa.feature.melspectrogram(y=wav_output, sr=sr, hop_length=256).astype(np.float32)
ps = ps[np.newaxis, ..., np.newaxis]
return ps
# 获取录音数据
def record_audio():
print("开始录音......")
frames = []
for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
data = stream.read(CHUNK)
frames.append(data)
print("录音已结束!")
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()
return WAVE_OUTPUT_FILENAME
# 预测
def infer(audio_data):
result = model.predict(audio_data)
lab = tf.argmax(result, 1)
return lab
if __name__ == '__main__':
try:
while True:
# 加载数据
data = load_data(record_audio())
# 获取预测结果
label = infer(data)
print('预测的标签为:%d' % label)
except Exception as e:
print(e)
stream.stop_stream()
stream.close()
p.terminate()