基于Pytorch实现的声音分类实例代码_python_

pip install pytest-runner pip install librosa==0.9.1 

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

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

pip install pyaudio 

pip install pydub 

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) 

# 生成数据列表 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('%s\t%d\n' % (sound_path, i)) else: f_train.write('%s\t%d\n' % (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') 

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 

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() 

 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()