基於Pytorch實現的聲音分類範例程式碼

2022-06-21 14:00:18

前言

本章我們來介紹如何使用Pytorch訓練一個區分不同音訊的分類模型，例如你有這樣一個需求，需要根據不同的鳥叫聲識別是什麼種類的鳥，這時你就可以使用這個方法來實現你的需求了。

原始碼地址：https://github.com/yeyupiaoling/AudioClassification-Pytorch

環境準備

主要介紹libsora，PyAudio，pydub的安裝，其他的依賴包根據需要自行安裝。

Python 3.7
Pytorch 1.10.0

安裝libsora

最簡單的方式就是使用pip命令安裝，如下：

pip install pytest-runner
pip install librosa==0.9.1

注意： 如果pip命令安裝不成功，那就使用原始碼安裝，下載原始碼：https://github.com/librosa/librosa/releases/， windows的可以下載zip壓縮包，方便解壓。

pip install pytest-runner
tar xzf librosa-<版本號>.tar.gz 或者 unzip librosa-<版本號>.tar.gz
cd librosa-<版本號>/
python setup.py install

如果出現 libsndfile64bit.dll': error 0x7e錯誤，請指定安裝版本0.6.3，如 pip install librosa==0.6.3

安裝ffmpeg，下載地址：http://blog.gregzaal.com/how-to-install-ffmpeg-on-windows/，筆者下載的是64位元，static版。
然後到C槽，筆者解壓，修改檔名為 ffmpeg，存放在 C:Program Files目錄下，並新增環境變數 C:Program Filesffmpegbin

最後修改原始碼，路徑為 C:Python3.7Libsite-packagesaudioreadffdec.py，修改32行程式碼，如下：

COMMANDS = ('C:\Program Files\ffmpeg\bin\ffmpeg.exe', 'avconv')

安裝PyAudio

使用pip安裝命令，如下：

pip install pyaudio

在安裝的時候需要使用到C++庫進行編譯，如果讀者的系統是windows，Python是3.7，可以在這裡下載whl安裝包，下載地址：https://github.com/intxcc/pyaudio_portaudio/releases

安裝pydub

使用pip命令安裝，如下：

pip install pydub

訓練分類模型

把音訊轉換成訓練資料最重要的是使用了librosa，使用librosa可以很方便得到音訊的梅爾頻譜（Mel Spectrogram），使用的API為 librosa.feature.melspectrogram()，輸出的是numpy值。關於梅爾頻譜具體資訊讀者可以自行了解，跟梅爾頻譜同樣很重要的梅爾倒譜（MFCCs）更多用於語音識別中，對應的API為 librosa.feature.mfcc()。同樣以下的程式碼，就可以獲取到音訊的梅爾頻譜。

wav, sr = librosa.load(data_path, sr=16000)
features = librosa.feature.melspectrogram(y=wav, sr=sr, n_fft=400, n_mels=80, hop_length=160, win_length=400)
features = librosa.power_to_db(features, ref=1.0, amin=1e-10, top_db=None)

生成資料列表

生成資料列表，用於下一步的讀取需要，audio_path為音訊檔路徑，使用者需要提前把音訊資料集存放在dataset/audio目錄下，每個資料夾存放一個類別的音訊資料，每條音訊資料長度在3秒以上，如 dataset/audio/鳥叫聲/······。audio是資料列表存放的位置，生成的資料類別的格式為 音訊路徑t音訊對應的類別標籤，音訊路徑和標籤用製表符 t分開。讀者也可以根據自己存放資料的方式修改以下函數。

Urbansound8K 是目前應用較為廣泛的用於自動城市環境聲分類研究的公共資料集，包含10個分類：空調聲、汽車鳴笛聲、兒童玩耍聲、狗叫聲、鑽孔聲、引擎空轉聲、槍聲、手提鑽、警笛聲和街道音樂聲。資料集下載地址：https://zenodo.org/record/1203745/files/UrbanSound8K.tar.gz。以下是針對Urbansound8K生成資料列表的函數。如果讀者想使用該資料集，請下載並解壓到 dataset目錄下，把生成資料列表程式碼改為以下程式碼。

# 生成資料列表
def get_data_list(audio_path, list_path):
    sound_sum = 0
    audios = 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:
            if '.wav' not in sound:continue
            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('%st%dn' % (sound_path, i))
                else:
                    f_train.write('%st%dn' % (sound_path, i))
                sound_sum += 1
        print("Audio：%d/%d" % (i + 1, len(audios)))

    f_test.close()
    f_train.close()


if __name__ == '__main__':
    get_data_list('dataset/UrbanSound8K/audio', 'dataset')

建立 reader.py用於在訓練時讀取資料。編寫一個 CustomDataset類，用讀取上一步生成的資料列表。

class CustomDataset(Dataset):
    def __init__(self, data_list_path, model='train', sr=16000, chunk_duration=3):
        super(CustomDataset, self).__init__()
        with open(data_list_path, 'r') as f:
            self.lines = f.readlines()
        self.model = model
        self.sr = sr
        self.chunk_duration = chunk_duration

    def __getitem__(self, idx):
        try:
            audio_path, label = self.lines[idx].replace('n', '').split('t')
            spec_mag = load_audio(audio_path, mode=self.model, sr=self.sr, chunk_duration=self.chunk_duration)
            return spec_mag, np.array(int(label), dtype=np.int64)
        except Exception as ex:
            print(f"[{datetime.now()}] 資料: {self.lines[idx]} 出錯，錯誤資訊: {ex}", file=sys.stderr)
            rnd_idx = np.random.randint(self.__len__())
            return self.__getitem__(rnd_idx)

    def __len__(self):
        return len(self.lines)

下面是在訓練時或者測試時讀取音訊資料，訓練時對轉換的梅爾頻譜資料隨機裁剪，如果是測試，就取前面的，最好要執行歸一化。

def load_audio(audio_path, mode='train', sr=16000, chunk_duration=3):
    # 讀取音訊資料
    wav, sr_ret = librosa.load(audio_path, sr=sr)
    if mode == 'train':
        # 隨機裁剪
        num_wav_samples = wav.shape[0]
        # 資料太短不利於訓練
        if num_wav_samples < sr:
            raise Exception(f'音訊長度不能小於1s，實際長度為：{(num_wav_samples / sr):.2f}s')
        num_chunk_samples = int(chunk_duration * sr)
        if num_wav_samples > num_chunk_samples + 1:
            start = random.randint(0, num_wav_samples - num_chunk_samples - 1)
            stop = start + num_chunk_samples
            wav = wav[start:stop]
            # 對每次都滿長度的再次裁剪
            if random.random() > 0.5:
                wav[:random.randint(1, sr // 2)] = 0
                wav = wav[:-random.randint(1, sr // 2)]
    elif mode == 'eval':
        # 為避免視訊記憶體溢位，只裁剪指定長度
        num_wav_samples = wav.shape[0]
        num_chunk_samples = int(chunk_duration * sr)
        if num_wav_samples > num_chunk_samples + 1:
            wav = wav[:num_chunk_samples]
    features = librosa.feature.melspectrogram(y=wav, sr=sr, n_fft=400, n_mels=80, hop_length=160, win_length=400)
    features = librosa.power_to_db(features, ref=1.0, amin=1e-10, top_db=None)
    # 歸一化
    mean = np.mean(features, 0, keepdims=True)
    std = np.std(features, 0, keepdims=True)
    features = (features - mean) / (std + 1e-5)
    features = features.astype('float32')
    return features

訓練

接著就可以開始訓練模型了，建立 train.py。我們搭建簡單的折積神經網路，如果音訊種類非常多，可以適當使用更大的折積神經網路模型。通過把音訊資料轉換成梅爾頻譜。然後定義優化方法和獲取訓練和測試資料。要注意 args.num_classes引數的值，這個是類別的數量，要根據你資料集中的分類數量來修改。

def train(args):
    # 獲取資料
    train_dataset = CustomDataset(args.train_list_path, model='train')
    train_loader = DataLoader(dataset=train_dataset, batch_size=args.batch_size, shuffle=True, collate_fn=collate_fn, num_workers=args.num_workers)

    test_dataset = CustomDataset(args.test_list_path, model='eval')
    test_loader = DataLoader(dataset=test_dataset, batch_size=args.batch_size, collate_fn=collate_fn, num_workers=args.num_workers)
    # 獲取分類標籤
    with open(args.label_list_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()
        class_labels = [l.replace('n', '') for l in lines]
    # 獲取模型
    device = torch.device("cuda")
    model = EcapaTdnn(num_classes=args.num_classes)
    model.to(device)

    # 獲取優化方法
    optimizer = torch.optim.Adam(params=model.parameters(),
                                 lr=args.learning_rate,
                                 weight_decay=5e-4)
    # 獲取學習率衰減函數
    scheduler = CosineAnnealingLR(optimizer, T_max=args.num_epoch)

    # 恢復訓練
    if args.resume is not None:
        model.load_state_dict(torch.load(os.path.join(args.resume, 'model.pth')))
        state = torch.load(os.path.join(args.resume, 'model.state'))
        last_epoch = state['last_epoch']
        optimizer_state = torch.load(os.path.join(args.resume, 'optimizer.pth'))
        optimizer.load_state_dict(optimizer_state)
        print(f'成功載入第 {last_epoch} 輪的模型引數和優化方法引數')

    # 獲取損失函數
    loss = torch.nn.CrossEntropyLoss()

最後執行訓練，每100個batch列印一次訓練紀錄檔，訓練一輪之後執行測試和儲存模型，在測試時，把每個batch的輸出都統計，最後求平均值。

    for epoch in range(args.num_epoch):
        loss_sum = []
        accuracies = []
        for batch_id, (spec_mag, label) in enumerate(train_loader):
            spec_mag = spec_mag.to(device)
            label = label.to(device).long()
            output = model(spec_mag)
            # 計算損失值
            los = loss(output, label)
            optimizer.zero_grad()
            los.backward()
            optimizer.step()

            # 計算準確率
            output = torch.nn.functional.softmax(output, dim=-1)
            output = output.data.cpu().numpy()
            output = np.argmax(output, axis=1)
            label = label.data.cpu().numpy()
            acc = np.mean((output == label).astype(int))
            accuracies.append(acc)
            loss_sum.append(los)
            if batch_id % 100 == 0:
                print(f'[{datetime.now()}] Train epoch [{epoch}/{args.num_epoch}], batch: {batch_id}/{len(train_loader)}, '
                      f'lr: {scheduler.get_last_lr()[0]:.8f}, loss: {sum(loss_sum) / len(loss_sum):.8f}, '
                      f'accuracy: {sum(accuracies) / len(accuracies):.8f}')
        scheduler.step()

每輪訓練結束之後都會執行一次評估，和儲存模型。評估會出來輸出準確率，還儲存了混合矩陣圖片，如下。

預測

在訓練結束之後，我們得到了一個模型引數檔案，我們使用這個模型預測音訊，在執行預測之前，需要把音訊轉換為梅爾頻譜資料，最後輸出的結果即為預測概率最大的標籤。

parser = argparse.ArgumentParser(description=__doc__)
add_arg = functools.partial(add_arguments, argparser=parser)
add_arg('audio_path',       str,    'dataset/UrbanSound8K/audio/fold5/156634-5-2-5.wav', '圖片路徑')
add_arg('num_classes',      int,    10,                        '分類的類別數量')
add_arg('label_list_path',  str,    'dataset/label_list.txt',  '標籤列表路徑')
add_arg('model_path',       str,    'models/model.pth',        '模型儲存的路徑')
args = parser.parse_args()


# 獲取分類標籤
with open(args.label_list_path, 'r', encoding='utf-8') as f:
    lines = f.readlines()
class_labels = [l.replace('n', '') for l in lines]
# 獲取模型
device = torch.device("cuda")
model = EcapaTdnn(num_classes=args.num_classes)
model.to(device)
model.load_state_dict(torch.load(args.model_path))
model.eval()


def infer():
    data = load_audio(args.audio_path, mode='infer')
    data = data[np.newaxis, :]
    data = torch.tensor(data, dtype=torch.float32, device=device)
    # 執行預測
    output = model(data)
    result = torch.nn.functional.softmax(output, dim=-1)
    result = result.data.cpu().numpy()
    # 顯示圖片並輸出結果最大的label
    lab = np.argsort(result)[0][-1]
    print(f'音訊：{args.audio_path} 的預測結果標籤為：{class_labels[lab]}')


if __name__ == '__main__':
    infer()

其他

為了方便讀取錄製資料和製作資料集，這裡提供了兩個程式，首先是 record_audio.py，這個用於錄製音訊，錄製的音訊影格率為44100，通道為1，16bit。

import pyaudio
import wave
import uuid
from tqdm import tqdm
import os

s = 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('pause')

建立 crop_audio.py，在訓練是隻是裁剪前面的3秒的音訊，所以我們要把錄製的硬碟安裝每3秒裁剪一段，把裁剪後音訊存放在音訊名稱命名的資料夾中。最後把這些檔案按照訓練資料的要求建立資料列表和訓練資料。

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) * 1000
    end_time = int(end_time) * 1000
    word = 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':
            continue
        file = 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 = 3
    crop_wav('save_audio', crop_len)

建立 infer_record.py，這個程式是用來不斷進行錄音識別，錄音時間之所以設定為6秒，所以我們可以大致理解為這個程式在實時錄音識別。通過這個應該我們可以做一些比較有趣的事情，比如把麥克風放在小鳥經常來的地方，通過實時錄音識別，一旦識別到有鳥叫的聲音，如果你的資料集足夠強大，有每種鳥叫的聲音資料集，這樣你還能準確識別是那種鳥叫。如果識別到目標鳥類，就啟動程式，例如拍照等等。

# 錄音引數
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 6
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 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_path):
    data = load_audio(audio_path, mode='infer')
    data = data[np.newaxis, :]
    data = torch.tensor(data, dtype=torch.float32, device=device)
    # 執行預測
    output = model(data)
    result = torch.nn.functional.softmax(output, dim=-1)
    result = result.data.cpu().numpy()
    # 顯示圖片並輸出結果最大的label
    lab = np.argsort(result)[0][-1]
    return class_labels[lab]


if __name__ == '__main__':
    try:
        while True:
            # 載入資料
            audio_path = record_audio()
            # 獲取預測結果
            label = infer(audio_path)
            print(f'預測的標籤為：{label}')
    except Exception as e:
        print(e)
        stream.stop_stream()
        stream.close()
        p.terminate()