基于Tensorflow的环境声音分类
生活随笔
收集整理的這篇文章主要介紹了
基于Tensorflow的环境声音分类
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
博客轉自于:https://yeyupiaoling.blog.csdn.net/article/details/105708251#t2
前言
本章我們來介紹如何使用Tensorflow訓練一個區分不同音頻的分類模型，例如你有這樣一個需求，需要根據不同的鳥叫聲識別是什么種類的鳥,或者識別環境中的聲音類型(空調聲、汽車鳴笛聲、兒童玩耍聲、狗叫聲、鉆孔聲、引擎空轉聲)等，這時你就可以使用這個方法來實現你的需求了。
環境準備
主要介紹libsora，PyAudio，pydub的安裝，其他的依賴包根據需要自行安裝。
- Python 3.7
- Tensorflow 2.1
安裝依賴包
pip install pytest-runner pip install librosa pip install pyaudio pip install pydub訓練分類模型
把音頻轉換成訓練數據最重要的是使用了librosa
，使用librosa可以很方便得到音頻的梅爾頻譜（Mel Spectrogram），使用的API為librosa.feature.melspectrogram()，輸出的是numpy值，可以直接用tensorflow訓練和預測。關于梅爾頻譜具體信息讀者可以自行了解，跟梅爾頻譜同樣很重要的梅爾倒譜（MFCCs）更多用于y1, sr1 = librosa.load(data_path, duration=2.97)
ps = librosa.feature.melspectrogram(y=y1, sr=sr1)
語音識別中，對應的API為librosa.feature.mfcc()。同樣以下的代碼，就可以獲取到音頻的梅爾頻譜，其中duration參數指定的是截取音頻的長度。
創建訓練數據
根據上面的方法，我們創建Tensorflow訓練數據，因為分類音頻數據小而多，最好的方法就是把這些音頻文件生成TFRecord，加快訓練速度。創建create_data.py用于生成TFRecord文件。
首先需要生成數據列表，用于下一步的讀取需要，audio_path為音頻文件路徑，用戶需要提前把音頻數據集存放在dataset/audio目錄下，每個文件夾存放一個類別的音頻數據，如dataset/audio/狗叫聲/······。每條音頻數據長度大于2.1秒，當然可以可以只其他的音頻長度，這個可以根據讀取的需要修改，如有需要的參數筆者都使用注釋標注了。audio是數據列表存放的位置，生成的數據類別的格式為音頻路徑\t音頻對應的類別標簽。讀者也可以根據自己存放數據的方式修改以下函數。
def get_data_list(audio_path, list_path):sound_sum = 0audios = os.listdir(audio_path)f_train = open(os.path.join(list_path, 'train_list.txt'), 'w')f_test = open(os.path.join(list_path, 'test_list.txt'), 'w')for i in range(len(audios)):sounds = os.listdir(os.path.join(audio_path, audios[i]))for sound in sounds:sound_path = os.path.join(audio_path, audios[i], sound)t = librosa.get_duration(filename=sound_path)# [可能需要修改參數] 過濾小于2.1秒的音頻if t >= 2.1:if sound_sum % 100 == 0:f_test.write('%s\t%d\n' % (sound_path, i))else:f_train.write('%s\t%d\n' % (sound_path, i))sound_sum += 1print("Audio：%d/%d" % (i + 1, len(audios)))f_test.close()f_train.close()if __name__ == '__main__':get_data_list('dataset/audio', 'dataset')有了以上的數據列表，就可開始生成TFRecord文件了。最終會生成train.tfrecord和test.tfrecord。筆者設置的音頻長度為2.04秒，不足長度會補0，如果需要使用不同的音頻長度時，需要修改wav_len參數值和len(ps)過濾值，wav_len參數值為音頻長度 16000 * 秒數，len(ps)過濾值為梅爾頻譜shape相乘。
# 獲取浮點數組 def _float_feature(value):if not isinstance(value, list):value = [value]return tf.train.Feature(float_list=tf.train.FloatList(value=value))# 獲取整型數據 def _int64_feature(value):if not isinstance(value, list):value = [value]return tf.train.Feature(int64_list=tf.train.Int64List(value=value))# 把數據添加到TFRecord中 def data_example(data, label):feature = {'data': _float_feature(data),'label': _int64_feature(label),}return tf.train.Example(features=tf.train.Features(feature=feature))# 開始創建tfrecord數據 def create_data_tfrecord(data_list_path, save_path):with open(data_list_path, 'r') as f:data = f.readlines()with tf.io.TFRecordWriter(save_path) as writer:for d in tqdm(data):try:path, label = d.replace('\n', '').split('\t')wav, sr = librosa.load(path, sr=16000)intervals = librosa.effects.split(wav, top_db=20)wav_output = []# [可能需要修改參數] 音頻長度 16000 * 秒數wav_len = int(16000 * 2.04)for sliced in intervals:wav_output.extend(wav[sliced[0]:sliced[1]])for i in range(5):# 裁剪過長的音頻，過短的補0if len(wav_output) > wav_len:l = len(wav_output) - wav_lenr = random.randint(0, l)wav_output = wav_output[r:wav_len + r]else:wav_output.extend(np.zeros(shape=[wav_len - len(wav_output)], dtype=np.float32))wav_output = np.array(wav_output)# 轉成梅爾頻譜ps = librosa.feature.melspectrogram(y=wav_output, sr=sr, hop_length=256).reshape(-1).tolist()# [可能需要修改參數] 梅爾頻譜shape ，librosa.feature.melspectrogram(y=wav_output, sr=sr, hop_length=256).shapeif len(ps) != 128 * 128: continuetf_example = data_example(ps, int(label))writer.write(tf_example.SerializeToString())if len(wav_output) <= wav_len:breakexcept Exception as e:print(e)if __name__ == '__main__':create_data_tfrecord('dataset/train_list.txt', 'dataset/train.tfrecord')create_data_tfrecord('dataset/test_list.txt', 'dataset/test.tfrecord')Urbansound8K是目前應用較為廣泛的用于自動城市環境聲分類研究的公共數據集，包含10個分類：空調聲、汽車鳴笛聲、兒童玩耍聲、狗叫聲、鉆孔聲、引擎空轉聲、槍聲、手提鉆、警笛聲和街道音樂聲。數據集下載地址
。以下是針對Urbansound8K生成數據列表的函數。數據下載并解壓到dataset目錄下，生成數據列表代碼如下。
創建reader.py用于在訓練時讀取TFRecord文件數據。如果讀者使用了其他的音頻長度，需要修改一下tf.io.FixedLenFeature參數的值，為梅爾頻譜的shape相乘的值。
import tensorflow as tfdef _parse_data_function(example):# [可能需要修改參數】 設置的梅爾頻譜的shape相乘的值data_feature_description = {'data': tf.io.FixedLenFeature([16384], tf.float32),'label': tf.io.FixedLenFeature([], tf.int64),}return tf.io.parse_single_example(example, data_feature_description)def train_reader_tfrecord(data_path, num_epochs, batch_size):raw_dataset = tf.data.TFRecordDataset(data_path)train_dataset = raw_dataset.map(_parse_data_function)train_dataset = train_dataset.shuffle(buffer_size=1000) \.repeat(count=num_epochs) \.batch(batch_size=batch_size) \.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)return train_datasetdef test_reader_tfrecord(data_path, batch_size):raw_dataset = tf.data.TFRecordDataset(data_path)test_dataset = raw_dataset.map(_parse_data_function)test_dataset = test_dataset.batch(batch_size=batch_size)return test_dataset訓練
接著就可以開始訓練模型了，創建train.py。我們搭建簡單的卷積神經網絡，通過把音頻數據轉換成梅爾頻譜，數據的shape也相當于灰度圖，所以我們可以當作圖像的輸入創建一個深度神經網絡。然后定義優化方法和獲取訓練和測試數據。input_shape設置為(128, None, 1))主要是為了適配其他音頻長度的輸入和預測是任意大小的輸入。class_dim為分類的總數。
import tensorflow as tf import reader import numpy as npclass_dim = 10 EPOCHS = 100 BATCH_SIZE=32model = tf.keras.models.Sequential([tf.keras.applications.ResNet50V2(include_top=False, weights=None, input_shape=(128, None, 1)),tf.keras.layers.ActivityRegularization(l2=0.5),tf.keras.layers.Dropout(rate=0.5),tf.keras.layers.GlobalMaxPooling2D(),tf.keras.layers.Dense(units=class_dim, activation=tf.nn.softmax) ])model.summary()# 定義優化方法 optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)train_dataset = reader.train_reader_tfrecord('dataset/train.tfrecord', EPOCHS, batch_size=BATCH_SIZE) test_dataset = reader.test_reader_tfrecord('dataset/test.tfrecord', batch_size=BATCH_SIZE)最后執行訓練，每200個batch執行一次測試和保存模型。要注意的是在創建TFRecord文件時，已經把音頻數據的梅爾頻譜轉換為一維list了，所以在數據輸入到模型前，需要把數據reshape為之前的shape，操作方式為reshape((-1, 128, 128, 1))。要注意的是如果讀者使用了其他長度的音頻，需要根據梅爾頻譜的shape修改。
for batch_id, data in enumerate(train_dataset):# [可能需要修改參數】 設置的梅爾頻譜的shapesounds = data['data'].numpy().reshape((-1, 128, 128, 1))labels = data['label']# 執行訓練with tf.GradientTape() as tape:predictions = model(sounds)# 獲取損失值train_loss = tf.keras.losses.sparse_categorical_crossentropy(labels, predictions)train_loss = tf.reduce_mean(train_loss)# 獲取準確率train_accuracy = tf.keras.metrics.sparse_categorical_accuracy(labels, predictions)train_accuracy = np.sum(train_accuracy.numpy()) / len(train_accuracy.numpy())# 更新梯度gradients = tape.gradient(train_loss, model.trainable_variables)optimizer.apply_gradients(zip(gradients, model.trainable_variables))if batch_id % 20 == 0:print("Batch %d, Loss %f, Accuracy %f" % (batch_id, train_loss.numpy(), train_accuracy))if batch_id % 200 == 0 and batch_id != 0:test_losses = list()test_accuracies = list()for d in test_dataset:# [可能需要修改參數】 設置的梅爾頻譜的shapetest_sounds = d['data'].numpy().reshape((-1, 128, 128, 1))test_labels = d['label']test_result = model(test_sounds)# 獲取損失值test_loss = tf.keras.losses.sparse_categorical_crossentropy(test_labels, test_result)test_loss = tf.reduce_mean(test_loss)test_losses.append(test_loss)# 獲取準確率test_accuracy = tf.keras.metrics.sparse_categorical_accuracy(test_labels, test_result)test_accuracy = np.sum(test_accuracy.numpy()) / len(test_accuracy.numpy())test_accuracies.append(test_accuracy)print('=================================================')print("Test, Loss %f, Accuracy %f" % (sum(test_losses) / len(test_losses), sum(test_accuracies) / len(test_accuracies)))print('=================================================')# 保存模型model.save(filepath='models/resnet50.h5')預測
在訓練結束之后，我們得到了一個預測模型，有了預測模型，執行預測非常方便。我們使用這個模型預測音頻，輸入的音頻會裁剪靜音部分，所以非靜音部分不能小于 0.5 秒，避免特征數量太少，當然這也不是一定的，可以任意修改。在執行預測之前，需要把音頻裁剪掉靜音部分，并且把裁剪后的音頻轉換為梅爾頻譜數據。預測的數據shape第一個為輸入數據的 batch 大小，如果想多個音頻一起數據，可以把他們存放在 list 中一起預測。最后輸出的結果即為預測概率最大的標簽。
import librosa import numpy as np import tensorflow as tfmodel = tf.keras.models.load_model('models/resnet50.h5')# 讀取音頻數據 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]])assert len(wav_output) >= 8000, "有效音頻小于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 psdef infer(audio_path):data = load_data(audio_path)result = model.predict(data)lab = tf.argmax(result, 1)return labif __name__ == '__main__':# 要預測的音頻文件path = ''label = infer(path)print('音頻：%s 的預測結果標簽為：%d' % (path, label))其他
為了方便讀取錄制數據和制作數據集，這里提供了兩個程序，首先是record_audio.py，這個用于錄制音頻，錄制的音頻幀率為44100，通道為1，16bit。
import pyaudio import wave import uuid from tqdm import tqdm import oss = input('請輸入你計劃錄音多少秒：')CHUNK = 1024 FORMAT = pyaudio.paInt16 CHANNELS = 1 RATE = 44100 RECORD_SECONDS = int(s) WAVE_OUTPUT_FILENAME = "save_audio/%s.wav" % str(uuid.uuid1()).replace('-', '')p = pyaudio.PyAudio()stream = p.open(format=FORMAT,channels=CHANNELS,rate=RATE,input=True,frames_per_buffer=CHUNK)print("開始錄音, 請說話......")frames = []for i in tqdm(range(0, int(RATE / CHUNK * RECORD_SECONDS))):data = stream.read(CHUNK)frames.append(data)print("錄音已結束!")stream.stop_stream() stream.close() p.terminate()if not os.path.exists('save_audio'):os.makedirs('save_audio')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('文件保存在：%s' % WAVE_OUTPUT_FILENAME) os.system('pau創建crop_audio.py，筆者在訓練默認訓練2.04秒的音頻，所以我們要把錄制的硬盤安裝每3秒裁剪一段，把裁剪后音頻存放在音頻名稱命名的文件夾中。最后把這些文件按照訓練數據的要求創建數據列表，和生成TFRecord文件。
import os import uuid import wave from pydub import AudioSegment# 按秒截取音頻 def get_part_wav(sound, start_time, end_time, part_wav_path):save_path = os.path.dirname(part_wav_path)if not os.path.exists(save_path):os.makedirs(save_path)start_time = int(start_time) * 1000end_time = int(end_time) * 1000word = sound[start_time:end_time]word.export(part_wav_path, format="wav")def crop_wav(path, crop_len):for src_wav_path in os.listdir(path):wave_path = os.path.join(path, src_wav_path)print(wave_path[-4:])if wave_path[-4:] != '.wav':continuefile = wave.open(wave_path)# 幀總數a = file.getparams().nframes# 采樣頻率f = file.getparams().framerate# 獲取音頻時間長度t = int(a / f)print('總時長為 %d s' % t)# 讀取語音sound = AudioSegment.from_wav(wave_path)for start_time in range(0, t, crop_len):save_path = os.path.join(path, os.path.basename(wave_path)[:-4], str(uuid.uuid1()) + '.wav')get_part_wav(sound, start_time, start_time + crop_len, save_path)if __name__ == '__main__':crop_len = 3crop_wav('save_audio', crop_len)創建infer_record.py，這個程序是用來不斷進行錄音識別，錄音時間之所以設置為 3 秒，保證裁剪靜音部分后有足夠的音頻長度用于預測，當然也可以修改成其他的長度值。因為識別的時間比較短，所以我們可以大致理解為這個程序在實時錄音識別。
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 labif __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()Github地址：https://github.com/yeyupiaoling/AudioClassification_Tensorflow
總結
以上是生活随笔為你收集整理的基于Tensorflow的环境声音分类的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： KTV项目总结
- 下一篇： 串口连接-console口