Python人工智能：使用Keras库实现基于1维卷积神经网络的噪声分类算法_专栏

本文实现基于1DCNN的10种不同噪声类型的分类算法，精度高达99%。本文的操作系统为Ubuntu 22.04，大部分内容与Windows系统相同，唯一不同的文件路径的表示方式不同。

一、噪声数据的获取与预处理

1.1 噪声数据集的获取

(1) 本文使用SPIB开源噪声数据集NoiseX-92中的15种噪声数据进行基于神经网络的噪声分类算法，官方下载下载地址为：Signal Processing Information Base (SPIB)，下载的时候需要一个个下载且下载速度比较慢。
(2) 也可以通过百度网盘下载

链接: https://pan.baidu.com/s/1FDZ3tMHyLbDPj275hEiuqQ，提取码: ayrr。

1.2 噪声数据的预处理

使用Python对NoiseX-92噪声数据集进行预处理使用了如下四个python库：

将下面的代码存放到如下图所示的noise_data_pro.py文件中，并将下载的NOISEX-92噪声数据聚集放入到同样的目录下。

易百纳社区

噪声数据预处理函数noise_data_pro.py函数主要输入参数与返回值如下所示：

(1) 输入参数：

length：每个样本的数据长度，本文选取1024；
number：每种噪声类型的样本个数，本文选取1000，10种噪声类型则共具有10000个样本；

(2) 返回值：

X_train：训练噪声数据（占总体的60%）
y_train：训练噪声目标值（占总体的60%）
X_valid：验证噪声数据（占总体的20%）
y_valid：验证噪声目标值（占总体的20%）
X_test：测试噪声数据（占总体的20%）
y_test：测试噪声目标值（占总体的20%）

噪声数据预处理代码：

from scipy.io import loadmat
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.model_selection import train_test_split, StratifiedShuffleSplit
import numpy as np

def noise_data_pro(
    length=1024,        # 每个样本的信号长度
    number=100,         # 每种信号的样本数
    enc_step=28         # 每次增强时候的步长
):
    # 读取噪声数据，并存储在files字典中
    files = {}

    # 依次读取NOISEX-92_mat文件夹中的.mat格式的噪声数据
    # (1) 白噪声数据的获取
    white = loadmat('./NOISEX-92_mat/white.mat')
    files['white'] = white['white'].ravel()

    # (2) 粉噪声数据的获取
    pink = loadmat('./NOISEX-92_mat/pink.mat')
    files['pink'] = pink['pink'].ravel()

    # (3) babble噪声数据的获取
    babble = loadmat('./NOISEX-92_mat/babble.mat')
    files['babble'] = babble['babble'].ravel()

    # (4) factory噪声数据的获取
    factory = loadmat('./NOISEX-92_mat/factory1.mat')
    files['factory'] = factory['factory1'].ravel()

    # (5) volvo噪声数据的获取
    volvo = loadmat('./NOISEX-92_mat/volvo.mat')
    files['volov'] = volvo['volvo'].ravel()

    # (6) leopard噪声数据的获取
    leopard = loadmat('./NOISEX-92_mat/leopard.mat')
    files['leopard'] = leopard['leopard'].ravel()

    # (7) f16噪声数据的获取
    f16 = loadmat('./NOISEX-92_mat/f16.mat')
    files['f16'] = f16['f16'].ravel()

    # (8) buccaneer噪声数据的获取
    buccaneer = loadmat('./NOISEX-92_mat/buccaneer1.mat')
    files['buccaneer'] = buccaneer['buccaneer1'].ravel()

    # (9) machinegun噪声数据的获取
    machinegun = loadmat('./NOISEX-92_mat/machinegun.mat')
    files['machinegun'] = machinegun['machinegun'].ravel()

    # (10) hfchannel噪声数据的获取
    hfchannel = loadmat('./NOISEX-92_mat/hfchannel.mat')
    files['hfchannel'] = hfchannel['hfchannel'].ravel()

    keys = files.keys()     # 得到files字典的键值
    Train_Samples = {}      # 所有训练噪声数据样本的暂存字典

    # 使用随机滑动的方法分别将10种噪声切分为number个样本，
    # 且每个样本的长度为length,滑动的步长为enc_step。
    for i in keys:
        slice_data = files[i]           # 读取每种噪声数据的序列
        all_lenght = len(slice_data)    # 获取每种噪声数据的长度
        end_index = int(all_lenght)     # 获得每种噪声数据的结束位置
        samp_train = int(number)        # 每个样本的噪声信号长度

        Train_sample = []       # 存放每种噪声数据的列表

        # 随机滑动获得噪声数据切片
        enc_time = length // enc_step
        samp_step = 0
        for j in range(samp_train):
            random_start = np.random.randint(
                low=0, high=(end_index - 2*length)
            )
            label = 0
            for h in range(enc_time):
                samp_step += 1
                random_start += enc_step
                sample = slice_data[random_start:random_start+length]
                Train_sample.append(sample)
                if samp_step == samp_train:
                    label = 1
                    break
            if label:
                break

        Train_Samples[i] = Train_sample

    x_train = []    # 训练数据
    y_train = []    # 训练数据标签
    label = 0

    # 以0-9来表示10种噪声数据的标签
    for i in Train_Samples.keys():
        x = Train_Samples[i]
        x_train += x
        lenx = len(x)
        y_train += [label] * lenx
        label += 1

    # 使用StandardScaler()方法对噪声数据进行正则化处理
    # 注意：此时x_train由list格式数据转变为ndarray格式
    x_train = StandardScaler().fit_transform(x_train)
    # 将y_train列表格式数据也转换为one_hot格式
    y_train = np.array(y_train).reshape([-1,1])
    Encoder = OneHotEncoder()
    Encoder.fit(y_train)
    y_train = Encoder.transform(y_train).toarray()
    y_train = np.asarray(y_train, dtype=np.int32)

    # 下面将数据集划分训练数据集与测试数据集
    X_train, X_test_temp, y_train, y_test_temp = train_test_split(
        x_train, y_train,
        test_size=0.4       # 训练数据集与测试数据集的比例为6:4
    )

    # 然后使用Stratifiedshuffle方法将测试数据划分为验证数据集与测试数据集
    ss = StratifiedShuffleSplit(n_splits=1, test_size=0.5)
    for train_index, test_index in ss.split(X_test_temp, y_test_temp):
        X_valid, X_test = X_test_temp[train_index], X_test_temp[test_index]
        y_valid, y_test = y_test_temp[train_index], y_test_temp[test_index]

    # 返回值：
    # (1) 训练数据集: X_train, y_train
    # (2) 测试数据集: X_test, y_test
    # (3) 噪声数据标签：keys
    return X_train, y_train, X_valid, y_valid,  X_test, y_test, keys

二、基于Keras的1维卷积神经网络噪声分类算法实现方法

from keras.layers import Dense, Conv1D, BatchNormalization
from keras.layers import MaxPooling1D, Activation, Flatten, Dropout
from keras.models import Sequential
from keras.utils import plot_model
from keras.regularizers import l2
from noise_data_pro import noise_data_pro
import numpy as np
import time

# 训练超参数
batch_size = 128        # 每次训练的批次
epochs = 15             # 训练的轮数
num_classes = 10        # 噪声类别
length = 1024           # 每个噪声样本的长度
number = 1000           # 每种噪声具有的样本数
mark = time.strftime("%Y%m%d_%H%M", time.localtime())

# 获取预处理后的噪声数据，其中，
# 训练数据集：(x_train, y_train)具有6000组数据；
# 验证数据集：(x_valid, y_valid)具有2000组数据；
# 测试数据集：(x_test, y_test)具有2000组数据。
x_train, y_train, x_valid, y_valid, x_test, y_test, keys = noise_data_pro(
    length=1024, number=1000
)

# 对数据集添加第三个轴，使其变成张量形式
x_train = x_train[:,:,np.newaxis]
x_valid = x_valid[:,:,np.newaxis]
x_test = x_test[:,:,np.newaxis]
# 下面获取每次送入1DCNN的数据形状
input_shape = x_train.shape[1:]

# 定义一个带有运行时间的模型结构图名的变量model_name
model_name = "cnn_diagnosis-{}".format(mark)

# 实例化一个Sequential
model = Sequential()

# 下面定义1DCNN的结构
# (1) 第一层卷积
model.add(
    Conv1D(
        filters=16,         # 1维卷积核数量
        kernel_size=64,     # 1维卷积核长度
        strides=16,         # 卷积核步长
        padding='same',     # 填充方法
        kernel_regularizer=l2(1e-4),
        input_shape=input_shape
    )
)                                       # 1D卷积层
model.add(BatchNormalization())         # 正则化层
model.add(Activation('relu'))           # 激活函数
model.add(MaxPooling1D(pool_size=2))    # 池化层

# (2) 第二层卷积
model.add(
    Conv1D(
        filters=32,
        kernel_size=3,
        strides=1,
        padding='same',
        kernel_regularizer=l2(1e-4),
        input_shape=input_shape
    )
)
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling1D(
    pool_size=2, strides=2, padding='valid'
))

# (3) 全连接层
model.add(Flatten())        # 展平
model.add(Dropout(0.2))     # 丢弃其中20%的权重，以防止过拟合
model.add(Dense(32))        # 添加全连接层
model.add(Activation('relu'))   # 激活函数

# (4) 输出层
model.add(
    Dense(
        units=num_classes,          # 噪声输出类型
        activation='softmax',       # 输出层激活函数
        kernel_regularizer=l2(1e-4)
    )
)

# 编译模型
model.compile(
    optimizer='Adam',                   # 优化其
    loss='categorical_crossentropy',    # 损失函数
    metrics=['accuracy']                # 算法衡量指标
)

# 模型训练
model.fit(
    x=x_train, y=y_train,
    batch_size=batch_size,
    epochs=epochs,
    verbose=1,
    validation_data=(x_valid, y_valid),
    shuffle=True
)

# 输出模型结构图
plot_model(
    model=model,            # 模型
    to_file='images/cnn-diagnosis.png', # 存储的位置及其名称
    show_shapes=True,       # 显示数据形状
    show_layer_names=True   # 显示每层的名字
)

# 模型评估
score = model.evaluate(x=x_test, y=y_test, verbose=0)
print("测试集上的损失率：", score[0])
print("测试集上的准确率：", score[1])

代码执行结果如下图所示：

易百纳社区