1
import torch
2
import torchaudio
3
import torch.nn as nn
4
import torch.nn.functional as F
5
import torch.optim as optim
6
from torch.utils.data import DataLoader
7
from torchaudio.datasets import SPEECHCOMMANDS
8
import os
9

10
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
11
#电脑并不是总有支持CUDA的GPU的，因此在没有CUDA时使用CPU来计算
12
print(f"Using device {device}")
13

14
LABELS = ['yes', 'no', 'up', 'down', 'left', 'right', 'on', 'off', 'stop', 'go']
15
label_to_index = {label: i for i, label in enumerate(LABELS)}
16
#enumerate()传入一个可迭代的对象作为参数，然后返回元组（序号, 可迭代对象中的一项）组成的列表
17
index_to_label = {i: label for i, label in enumerate(LABELS)}

1
class SubsetSC(SPEECHCOMMANDS):
2
    def __init__(self, subset: str = None):
3
        super().__init__("./", download=True)
4
        #在这里传入数据集所需要放置的路径，并打开自动下载
5
        def load_list(filename):
6
            filepath = os.path.join(self._path, filename)
7
            #传入的起始路径与我们得到的_path有关，它是在SPEECHCOMMANDS中定义的
8
            with open(filepath) as fileobj:
9
                return [os.path.normpath(os.path.join(self._path, line.strip())) for line in fileobj]
10

11
        if subset == "validation":
12
        #SPEECHCOMMANDS中下载的文件夹里已经包含了训练集，验证集的区分，并且以txt形式放在数据集的根目录下，直接使用即可
13
            self._walker = load_list("validation_list.txt")
14
        elif subset == "testing":
15
            self._walker = load_list("testing_list.txt")
16
        elif subset == "training":
17
            excludes = load_list("validation_list.txt") + load_list("testing_list.txt")
18
            excludes = set(excludes)
19
            self._walker = [w for w in self._walker if w not in excludes]
20

21
        #过滤出我们需要的标签
22
        self._walker = [w for w in self._walker if w.split(os.sep)[-2] in LABELS]
23

24
train_set = SubsetSC("training")
25
val_set = SubsetSC("validation")
26
test_set = SubsetSC("testing")
27

28
print(f"Train data size: {len(train_set)}")
29
#可以简单查看一下数据集的长度

1
criterion = nn.CrossEntropyLoss()
2
#这里传入了学习率
3
optimizer = optim.Adam(model.parameters(), lr=0.001)
4

5
def train_epoch(model, loader, optimizer, criterion):
6
    model.train()
7
    running_loss = 0.0
8
    correct = 0
9
    total = 0
10

11
    for inputs, labels in loader:
12
    #优化器的梯度本身是会累加的，也就是每一批新的梯度会和上一次的梯度累加在一起，我们不希望这样，因此可以用这个函数来清零梯度
13
        optimizer.zero_grad()
14

15
        outputs = model(inputs)
16
        loss = criterion(outputs, labels)
17
        loss.backward()
18
        #这里的backward()的作用就是反向传播，然后计算梯度，这时候pytorch会自动求导，实际上类似于通过两个点来求一个近似的斜率
19
        optimizer.step()
20
        #这一步真正地修改了模型的参数，它参考loss.backward()给出的结果，以及学习率，去向可能使loss更小的方向更新每一个参数
21
        running_loss += loss.item()
22
        _, predicted = outputs.max(1)
23
        total += labels.size(0)
24
        correct += predicted.eq(labels).sum().item()
25
        #predicted.eq(labels)是一个将预测结果与正确答案进行对比，从而得到的布尔值组成的列表，再加入.sum()就会将布尔值，也就是0和1求和，得到一个一维张量，再进行.item()就将数字从张量里拿出来变为一个数，也就是正确预测的个数
26

27
    #在这里，我们刚刚将每个Batch的Loss累加起来，现在除以数据的数量，就是平均Loss了
28
    return running_loss / len(loader), 100. * correct / total
29

30
def eval_epoch(model, loader, criterion):
31
    model.eval()
32
    running_loss = 0.0
33
    correct = 0
34
    total = 0
35

36
    #我们之前说过pytorch会自动求导，但是在验证的时候，我们不需要更新参数，也就不需要求导，因此可以通过这一行来使pytorch不自动求导，就可以节省算力了
37
    with torch.no_grad():
38
        for inputs, labels in loader:
39
            outputs = model(inputs)
40
            loss = criterion(outputs, labels)
41

42
            running_loss += loss.item()
43
            _, predicted = outputs.max(1)
44
            total += labels.size(0)
45
            correct += predicted.eq(labels).sum().item()
46

47
    return running_loss / len(loader), 100. * correct / total
48

49
#开始训练，这时候我们定义一下想要训练的epoch数，这里10就够了
50
epochs = 10
51
print("Starting training...")
52
for epoch in range(epochs):
53
    train_loss, train_acc = train_epoch(model, train_loader, optimizer, criterion)
54
    val_loss, val_acc = eval_epoch(model, val_loader, criterion)
55

56
    print(f"Epoch {epoch+1}/{epochs} | "
57
          f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}% | "
58
          f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%")
59
    #返回一下各种数据总是好的，可以让我们看到模型到底有没有长进，实际上也有tensorboard之类更高级的监测模块，不过我们没有使用

1
#定义保存路径
2
model_save_path = "audio_cnn_weights.pth"
3

4
#提取模型中的所有可学习参数（权重和偏置）
5
#model.state_dict() 是一个包含所有参数的 Python 字典
6
torch.save(model.state_dict(), model_save_path)
7

8
print(f"Saved weight to: {model_save_path}")

1
# 在测试集上测试
2
test_loss, test_acc = eval_epoch(model, test_loader, criterion)
3
print(f"Final Test Accuracy: {test_acc:.2f}%")
4

5
def predict_audio(model, waveform, sr):
6
    model.eval()
7
    with torch.no_grad():
8
        target_sr = 16000
9
        #在采样率不同的时候重采样
10
        if sr != target_sr:
11
            resampler = torchaudio.transforms.Resample(sr, target_sr)
12
            waveform = resampler(waveform)
13

14
    #在非单声道的时候将声道合并
15
        if waveform.shape[0] != 1:
16
            waveform = waveform.mean(dim=0, keepdim=True)
17
        #修补长度
18
        if waveform.shape[1] < 16000:
19
            waveform = F.pad(waveform, (0, 16000 - waveform.shape[1]))
20
        elif waveform.shape[1] > 16000:
21
            waveform = waveform[:, :16000]
22

23
        #添加Batch维度并移动到设备
24
        #unsqueeze()是一个对张量的操作，可以增加维度，其参数是增加维度的位置，这一张量原本的尺寸是(1, 16000)，因此一共有三个地方可以增加维度，从前到后分别用0，1，2表示，这样就能保证输入的波形形状是正确的
25
        waveform = waveform.unsqueeze(0).to(device) # (1, 1, 16000)
26

27
        #提取频谱图
28
        melspec = transform(waveform)
29

30
        #模型预测
31
        output = model(melspec)
32
        #之前在说CrossEntropyLoss的时候说过了Softmax，现在我们没有标准答案，也就不需要Loss函数了，不过它也带走了Softmax函数，因此我们也得加上一个Softmax来使得输出变为概率
33
        #使用softmax的时候需要指定对哪个维度进行操作
34
        #还记得模型的输出吗，是(Batch, Number of classes)，在这里是(1, 10)，第二个维度才是需要softmax的，又由于计算机从0开始数数，所以是维度1
35
        probabilities = F.softmax(output, dim=1)
36

37
        predicted_idx = torch.argmax(probabilities, dim=1).item()
38
        confidence = probabilities[0][predicted_idx].item()
39

40
        return index_to_label[predicted_idx], confidence

1
sample_waveform, sample_rate, true_label, *_ = test_set[0]
2
predicted_label, conf = predict_audio(model, sample_waveform, sample_rate)
3

4
print(f"True Label: {true_label}")
5
print(f"Predicted: {predicted_label} (Confidence: {conf*100:.2f}%)")

1
#显然你的电脑上不可能有这一条音频，因此要把路径换成你的，并且记得改掉单反斜杠
2
wav, sr = torchaudio.load("I:\\AudioResources\\yes.wav")
3
predicted_label, conf = predict_audio(model, wav, sr)
4

5
print(f"Predicted: {predicted_label} (Confidence: {conf*100:.2f}%)")

1
#重新实例化一模一样的模型结构
2
#你需要确保 AudioCNN 类的定义在这段代码块之前已经运行过，否则这个类是不存在的
3
loaded_model = AudioCNN(num_classes=len(LABELS)).to(device)
4

5
#从硬盘读取权重字典，并注入到模型中
6
#map_location非常重要，它能确保如果你的模型是在 GPU 上训练的，但现在你想在没有 GPU 的普通电脑上推理，它会自动将其映射到 CPU。
7
#同样的，可能你的权重路径不长这样，那么就需要改一下下面的路径参数
8
weights = torch.load("audio_cnn_weights.pth", map_location=device)
9
loaded_model.load_state_dict(weights)
10

11
#切换为评估模式，我们之前已经介绍过，这会关闭Dropout并冻结BatchNorm等层
12
loaded_model.eval()
13

14
print("Loaded weight")