使用torchaudio的语音命令识别

原文：https://pytorch.org/tutorials/intermediate/speech_command_recognition_with_torchaudio.html

本教程将向您展示如何正确设置音频数据集的格式，然后在数据集上训练/测试音频分类器网络。

Colab 提供了 GPU 选项。 在菜单选项卡中，选择“运行系统”，然后选择“更改运行系统类型”。 在随后的弹出窗口中，您可以选择 GPU。 更改之后，运行时应自动重新启动（这意味着来自已执行单元的信息会消失）。

首先，让我们导入常见的 Torch 包，例如torchaudio，可以按照网站上的说明进行安装。

# Uncomment the following line to run in Google Colab

# CPU:
# !pip install torch==1.7.0+cpu torchvision==0.8.1+cpu torchaudio==0.7.0 -f https://download.pytorch.org/whl/torch_stable.html

# GPU:
# !pip install torch==1.7.0+cu101 torchvision==0.8.1+cu101 torchaudio==0.7.0 -f https://download.pytorch.org/whl/torch_stable.html

# For interactive demo at the end:
# !pip install pydub

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import torchaudio

import matplotlib.pyplot as plt
import IPython.display as ipd
from tqdm.notebook import tqdm

让我们检查一下 CUDA GPU 是否可用，然后选择我们的设备。 在 GPU 上运行网络将大大减少训练/测试时间。

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(device)

导入数据集

我们使用torchaudio下载并表示数据集。 在这里，我们使用 SpeechCommands，它是由不同人员说出的 35 个命令的数据集。 数据集SPEECHCOMMANDS是数据集的torch.utils.data.Dataset版本。 在此数据集中，所有音频文件的长度约为 1 秒（因此约为 16000 个时间帧）。

实际的加载和格式化步骤是在访问数据点时发生的，torchaudio负责将音频文件转换为张量。 如果想直接加载音频文件，可以使用torchaudio.load()。 它返回一个包含新创建的张量的元组以及音频文件的采样频率（SpeechCommands为 16kHz）。

回到数据集，这里我们创建一个子类，将其分为标准训练，验证和测试子集。

from torchaudio.datasets import SPEECHCOMMANDS
import os

class SubsetSC(SPEECHCOMMANDS):
    def __init__(self, subset: str = None):
        super().__init__("./", download=True)

        def load_list(filename):
            filepath = os.path.join(self._path, filename)
            with open(filepath) as fileobj:
                return [os.path.join(self._path, line.strip()) for line in fileobj]

        if subset == "validation":
            self._walker = load_list("validation_list.txt")
        elif subset == "testing":
            self._walker = load_list("testing_list.txt")
        elif subset == "training":
            excludes = load_list("validation_list.txt") + load_list("testing_list.txt")
            excludes = set(excludes)
            self._walker = [w for w in self._walker if w not in excludes]

# Create training and testing split of the data. We do not use validation in this tutorial.
train_set = SubsetSC("training")
test_set = SubsetSC("testing")

waveform, sample_rate, label, speaker_id, utterance_number = train_set[0]

SPEECHCOMMANDS数据集中的数据点是一个由波形（音频信号），采样率，发声（标签），讲话者的 ID，发声数组成的元组。

print("Shape of waveform: {}".format(waveform.size()))
print("Sample rate of waveform: {}".format(sample_rate))

plt.plot(waveform.t().numpy());

让我们找到数据集中可用的标签列表。

labels = sorted(list(set(datapoint[2] for datapoint in train_set)))
labels

35 个音频标签是用户说的命令。 前几个文件是人们所说的marvin。

waveform_first, *_ = train_set[0]
ipd.Audio(waveform_first.numpy(), rate=sample_rate)

waveform_second, *_ = train_set[1]
ipd.Audio(waveform_second.numpy(), rate=sample_rate)

最后一个文件是有人说“视觉”。

waveform_last, *_ = train_set[-1]
ipd.Audio(waveform_last.numpy(), rate=sample_rate)

格式化数据

这是将转换应用于数据的好地方。 对于波形，我们对音频进行下采样以进行更快的处理，而不会损失太多的分类能力。

我们无需在此应用其他转换。 对于某些数据集，通常必须通过沿通道维度取平均值或仅保留其中一个通道来减少通道数量（例如，从立体声到单声道）。 由于SpeechCommands使用单个通道进行音频，因此此处不需要。

new_sample_rate = 8000
transform = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=new_sample_rate)
transformed = transform(waveform)

ipd.Audio(transformed.numpy(), rate=new_sample_rate)

我们使用标签列表中的每个索引对每个单词进行编码。

def label_to_index(word):
    # Return the position of the word in labels
    return torch.tensor(labels.index(word))

def index_to_label(index):
    # Return the word corresponding to the index in labels
    # This is the inverse of label_to_index
    return labels[index]

word_start = "yes"
index = label_to_index(word_start)
word_recovered = index_to_label(index)

print(word_start, "-->", index, "-->", word_recovered)

为了将由录音和语音构成的数据点列表转换为该模型的两个成批张量，我们实现了整理函数，PyTorch DataLoader使用了该函数，允许我们分批迭代数据集。 有关使用整理函数的更多信息，请参见文档。

在整理函数中，我们还应用了重采样和文本编码。

def pad_sequence(batch):
    # Make all tensor in a batch the same length by padding with zeros
    batch = [item.t() for item in batch]
    batch = torch.nn.utils.rnn.pad_sequence(batch, batch_first=True, padding_value=0.)
    return batch.permute(0, 2, 1)

def collate_fn(batch):

    # A data tuple has the form:
    # waveform, sample_rate, label, speaker_id, utterance_number

    tensors, targets = [], []

    # Gather in lists, and encode labels as indices
    for waveform, _, label, *_ in batch:
        tensors += [waveform]
        targets += [label_to_index(label)]

    # Group the list of tensors into a batched tensor
    tensors = pad_sequence(tensors)
    targets = torch.stack(targets)

    return tensors, targets

batch_size = 256

if device == "cuda":
    num_workers = 1
    pin_memory = True
else:
    num_workers = 0
    pin_memory = False

train_loader = torch.utils.data.DataLoader(
    train_set,
    batch_size=batch_size,
    shuffle=True,
    collate_fn=collate_fn,
    num_workers=num_workers,
    pin_memory=pin_memory,
)
test_loader = torch.utils.data.DataLoader(
    test_set,
    batch_size=batch_size,
    shuffle=False,
    drop_last=False,
    collate_fn=collate_fn,
    num_workers=num_workers,
    pin_memory=pin_memory,
)

定义网络

在本教程中，我们将使用卷积神经网络来处理原始音频数据。 通常，更高级的转换将应用于音频数据，但是 CNN 可以用于准确处理原始数据。 具体架构是根据本文中描述的 M5 网络架构建模的。 模型处理原始音频数据的一个重要方面是其第一层过滤器的接收范围。 我们模型的第一个过滤器长度为 80，因此在处理以 8kHz 采样的音频时，接收场约为 10ms（而在 4kHz 时约为 20ms）。 此大小类似于语音处理应用，该应用通常使用 20ms 到 40ms 的接收域。

class M5(nn.Module):
    def __init__(self, n_input=1, n_output=35, stride=16, n_channel=32):
        super().__init__()
        self.conv1 = nn.Conv1d(n_input, n_channel, kernel_size=80, stride=stride)
        self.bn1 = nn.BatchNorm1d(n_channel)
        self.pool1 = nn.MaxPool1d(4)
        self.conv2 = nn.Conv1d(n_channel, n_channel, kernel_size=3)
        self.bn2 = nn.BatchNorm1d(n_channel)
        self.pool2 = nn.MaxPool1d(4)
        self.conv3 = nn.Conv1d(n_channel, 2 * n_channel, kernel_size=3)
        self.bn3 = nn.BatchNorm1d(2 * n_channel)
        self.pool3 = nn.MaxPool1d(4)
        self.conv4 = nn.Conv1d(2 * n_channel, 2 * n_channel, kernel_size=3)
        self.bn4 = nn.BatchNorm1d(2 * n_channel)
        self.pool4 = nn.MaxPool1d(4)
        self.fc1 = nn.Linear(2 * n_channel, n_output)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(self.bn1(x))
        x = self.pool1(x)
        x = self.conv2(x)
        x = F.relu(self.bn2(x))
        x = self.pool2(x)
        x = self.conv3(x)
        x = F.relu(self.bn3(x))
        x = self.pool3(x)
        x = self.conv4(x)
        x = F.relu(self.bn4(x))
        x = self.pool4(x)
        x = F.avg_pool1d(x, x.shape[-1])
        x = x.permute(0, 2, 1)
        x = self.fc1(x)
        return F.log_softmax(x, dim=2)

model = M5(n_input=transformed.shape[0], n_output=len(labels))
model.to(device)
print(model)

def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)

n = count_parameters(model)
print("Number of parameters: %s" % n)

我们将使用与本文相同的优化技术，将权重衰减设置为 0.0001 的 Adam 优化器。 首先，我们将以 0.01 的学习率进行训练，但是在 20 个周期后的训练过程中，我们将使用scheduler将其降低到 0.001。

optimizer = optim.Adam(model.parameters(), lr=0.01, weight_decay=0.0001)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=20, gamma=0.1)  # reduce the learning after 20 epochs by a factor of 10

训练和测试网络

现在，我们定义一个训练函数，它将训练数据输入模型中，并执行反向传播和优化步骤。 对于训练，我们将使用的损失是负对数可能性。 然后，在每个周期之后将对网络进行测试，以查看训练期间准确率如何变化。

def train(model, epoch, log_interval):
    model.train()
    for batch_idx, (data, target) in enumerate(train_loader):

        data = data.to(device)
        target = target.to(device)

        # apply transform and model on whole batch directly on device
        data = transform(data)
        output = model(data)

        # negative log-likelihood for a tensor of size (batch x 1 x n_output)
        loss = F.nll_loss(output.squeeze(), target)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        # print training stats
        if batch_idx % log_interval == 0:
            print(f"Train Epoch: {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)} ({100\. * batch_idx / len(train_loader):.0f}%)]\tLoss: {loss.item():.6f}")

        # update progress bar
        pbar.update(pbar_update)
        # record loss
        losses.append(loss.item())

现在我们有了训练函数，我们需要制作一个用于测试网络准确率的函数。 我们将模型设置为eval()模式，然后对测试数据集进行推断。 调用eval()将网络中所有模块中的训练变量设置为false。 某些层（例如批量归一化层和丢弃层）在训练期间的行为会有所不同，因此此步骤对于获取正确的结果至关重要。

def number_of_correct(pred, target):
    # count number of correct predictions
    return pred.squeeze().eq(target).sum().item()

def get_likely_index(tensor):
    # find most likely label index for each element in the batch
    return tensor.argmax(dim=-1)

def test(model, epoch):
    model.eval()
    correct = 0
    for data, target in test_loader:

        data = data.to(device)
        target = target.to(device)

        # apply transform and model on whole batch directly on device
        data = transform(data)
        output = model(data)

        pred = get_likely_index(output)
        correct += number_of_correct(pred, target)

        # update progress bar
        pbar.update(pbar_update)

    print(f"\nTest Epoch: {epoch}\tAccuracy: {correct}/{len(test_loader.dataset)} ({100\. * correct / len(test_loader.dataset):.0f}%)\n")

最后，我们可以训练和测试网络。 我们将训练网络十个周期，然后降低学习率，再训练十个周期。 在每个周期之后将对网络进行测试，以查看训练过程中准确率如何变化。

log_interval = 20
n_epoch = 2

pbar_update = 1 / (len(train_loader) + len(test_loader))
losses = []

# The transform needs to live on the same device as the model and the data.
transform = transform.to(device)
with tqdm(total=n_epoch) as pbar:
    for epoch in range(1, n_epoch + 1):
        train(model, epoch, log_interval)
        test(model, epoch)
        scheduler.step()

# Let's plot the training loss versus the number of iteration.
# plt.plot(losses);
# plt.title("training loss");

2 个周期后，测试集的网络准确率应超过 65%，而 21 个周期后，网络应达到 85%。 让我们看一下训练集中的最后几句话，看看模型是如何做到的。

def predict(tensor):
    # Use the model to predict the label of the waveform
    tensor = tensor.to(device)
    tensor = transform(tensor)
    tensor = model(tensor.unsqueeze(0))
    tensor = get_likely_index(tensor)
    tensor = index_to_label(tensor.squeeze())
    return tensor

waveform, sample_rate, utterance, *_ = train_set[-1]
ipd.Audio(waveform.numpy(), rate=sample_rate)

print(f"Expected: {utterance}. Predicted: {predict(waveform)}.")

如果有一个示例，我们来寻找一个分类错误的示例。

for i, (waveform, sample_rate, utterance, *_) in enumerate(test_set):
    output = predict(waveform)
    if output != utterance:
        ipd.Audio(waveform.numpy(), rate=sample_rate)
        print(f"Data point #{i}. Expected: {utterance}. Predicted: {output}.")
        break
else:
    print("All examples in this dataset were correctly classified!")
    print("In this case, let's just look at the last data point")
    ipd.Audio(waveform.numpy(), rate=sample_rate)
    print(f"Data point #{i}. Expected: {utterance}. Predicted: {output}.")

随意尝试使用其中一个标签的自己的录音！ 例如，使用 Colab，在执行下面的单元格时说“ Go”。 这将录制一秒钟的音频并尝试对其进行分类。

from google.colab import output as colab_output
from base64 import b64decode
from io import BytesIO
from pydub import AudioSegment

RECORD = """
const sleep  = time => new Promise(resolve => setTimeout(resolve, time))
const b2text = blob => new Promise(resolve => {
  const reader = new FileReader()
  reader.onloadend = e => resolve(e.srcElement.result)
  reader.readAsDataURL(blob)
})
var record = time => new Promise(async resolve => {
  stream = await navigator.mediaDevices.getUserMedia({ audio: true })
  recorder = new MediaRecorder(stream)
  chunks = []
  recorder.ondataavailable = e => chunks.push(e.data)
  recorder.start()
  await sleep(time)
  recorder.onstop = async ()=>{
    blob = new Blob(chunks)
    text = await b2text(blob)
    resolve(text)
  }
  recorder.stop()
})
"""

def record(seconds=1):
    display(ipd.Javascript(RECORD))
    print(f"Recording started for {seconds} seconds.")
    s = colab_output.eval_js("record(%d)" % (seconds * 1000))
    print("Recording ended.")
    b = b64decode(s.split(",")[1])

    fileformat = "wav"
    filename = f"_audio.{fileformat}"
    AudioSegment.from_file(BytesIO(b)).export(filename, format=fileformat)
    return torchaudio.load(filename)

waveform, sample_rate = record()
print(f"Predicted: {predict(waveform)}.")
ipd.Audio(waveform.numpy(), rate=sample_rate)

总结

在本教程中，我们使用了torchaudio来加载数据集并对信号进行重新采样。 然后，我们定义了经过训练的神经网络，以识别给定命令。 还有其他数据预处理方法，例如找到梅尔频率倒谱系数（MFCC），可以减小数据集的大小。 此变换也可以在torchaudio中作为torchaudio.transforms.MFCC使用。

脚本的总运行时间：（0 分钟 0.000 秒）

下载 Python 源码：speech_command_recognition_with_torchaudio.py

下载 Jupyter 笔记本：speech_command_recognition_with_torchaudio.ipynb

由 Sphinx 画廊生成的画廊