心音智能检测

PCG_Categrieos_Detection/
├── Dataset
│   ├── dataset_2_categories # 二分类数据（共4399个）│   └── dataset_4_categories # 四分类数据（共800个）├── dataset2.npy # 预处理后保存的二分类数据├── dataset4.npy # 预处理后保存的四分类数据├── model_weights # 模型训练完成后保存的权重文件│   ├── model_2.pdiparams
│   ├── model_2.pdiparams.info
│   ├── model_2.pdmodel
│   ├── model_4.pdiparams
│   ├── model_4.pdiparams.info
│   └── model_4.pdmodel
├── extract_bispectrum.py # 数据预处理├── PCG_MLP.py # 网络定义├── polycoherence.py # 二阶谱分析法├── rasbeery_inference.py # 在树莓派运行的脚本├── read_data.py # 读取数据└── train.py # 训练网络

# 读取音频import librosaimport matplotlib.pyplot as pltdef plot_signal(audio_data, title=None):
    plt.figure(figsize=(9, 3.0), dpi=300)
    plt.plot(audio_data, linewidth=1)
    plt.title(title,fontsize = 16)
    plt.tick_params(labelsize=12)
    plt.grid()
    plt.show()# 路径需根据实际情况修改，这里以‘work/dataset2/normal/b0031.wav’为例audio_path = 'work/PCG_Categrieos_Detection/Dataset/dataset_2_categories/normal/New_N_001.wav'audio_data, fs = librosa.load(audio_path, sr=None)
plot_signal(audio_data, title='Initial Audio')

# 音频降噪# 由于音频在制作时不可避免地会保存一部分噪声，我们需对音频文件进行数字滤波，旨在滤除高频噪声以及直流噪声，同时尽可能保留心音信号。我们把音频送入二阶25-400hz的巴特沃斯中值滤波器，并可视化音频。from scipy import signaldef band_pass_filter(original_signal, order, fc1,fc2, fs):
    '''
    中值滤波器
    :param original_signal: 音频数据
    :param order: 滤波器阶数
    :param fc1: 截止频率
    :param fc2: 截止频率
    :param fs: 音频采样率
    :return: 滤波后的音频数据
    '''
    b, a = signal.butter(N=order, Wn=[2*fc1/fs,2*fc2/fs], btype='bandpass')
    new_signal = signal.lfilter(b, a, original_signal)    return new_signal

audio_data = band_pass_filter(audio_data, 2, 25, 400, fs)
plot_signal(audio_data, title='After Filter')

# 音频下采样# 为了降低模型的计算量，我们对所有的音频信号进行下采样，考虑到我们已经对音频进行了25-400hz的中值滤波，根据奈奎斯特采样定律，我们把信号下采样到1000hz。这里我们调用samplerate包来完成这项工作，并可视化音频。import samplerate# AIStudio自带环境中并没有samplerate工具包，需要先下载，在终端中输入 pip install samplerate 即可完成安装# 下采样down_sample_audio_data = samplerate.resample(audio_data.T, 1000 / fs, converter_type='sinc_best').T
plot_signal(down_sample_audio_data, title='Down_sampled')

# 归一化# 由于不同数据集中的音频文件尺度差异较大，我们对所有的音频信号进行归一化，使其范围在[-1,1]区间内，并可视化音频import numpy as np
down_sample_audio_data = down_sample_audio_data / np.max(np.abs(down_sample_audio_data))
plot_signal(down_sample_audio_data, title='Normalized')

# 切割音频# 由于最终数据集数据来源于不同渠道，音频的时间不一致。为了尽可能多利用已有的数据集，我们对较长的音频进行切割。我们以2.5s为单位对音频进行切割。total_num = len(down_sample_audio_data)/(2500)# 计算切割次数fig = plt.figure(figsize=(12, 5), dpi=300)

ax1 = fig.add_subplot(2,1,1)
plt.plot(down_sample_audio_data,linewidth=1)
plt.title('Cut Audio(With Overlap)',fontsize=16)
plt.tick_params(labelsize=12)
plt.ylim([-1.2,1.2])
plt.grid()for j in range(int(total_num)):
    plt.vlines(j*2500, -1.2, 1.2, color="red",linestyle='--',linewidth=1.1)

ax2 = fig.add_subplot(2, 1, 2)
plt.plot(down_sample_audio_data, linewidth=1)for j in range(int(total_num)):
    plt.vlines(j*2500 + 1250, -1.2, 1.2, color="green",linestyle='--',linewidth=1.1)
plt.ylim([-1.2,1.2])
plt.grid()
plt.show()

from paddle.io import Dataset, DataLoaderclass MyDataset(Dataset):
    """
    步骤一：继承paddle.io.Dataset类
    """
    def __init__(self, data, label):
        """
        步骤二：实现构造函数，定义数据集大小
        """
        super(MyDataset, self).__init__()
        self.data = data
        self.label = label    def __getitem__(self, index):
        """
        步骤三：实现__getitem__方法，定义指定index时如何获取数据，并返回单条数据（训练数据，对应的标签）
        """
        # for img, lab in zip(self.data, self.label):
            
        return self.data[index], self.label[index]    def __len__(self):
        """
        步骤四：实现__len__方法，返回数据集总数目
        """
        return len(self.data)

import librosa # librosa是一个用于音乐和音频分析的 python 包。它提供了创建音乐信息检索系统所需的构建块。from scipy.fftpack import next_fast_len # fftpack用于进行傅立叶变换计算from scipy.signal import spectrogram # 处理声音信号def get_bi_spectrum(file_folder, class_list, data_num=2000):
    # dataset = np.zeros((1,1,256,256)) 
    dataset = []    for class_nam in class_list:
        path = os.path.join(file_folder, class_nam)
        all_files = get_all_filenames(path)        # index = 0
        for name in all_files[:data_num]:
            file_path = os.path.join(path, name)
            sig, sr = librosa.load(file_path, sr=1000) # load heart sound data
            freq1, freq2, bi_spectrum = polycoherence(sig,nfft=1024, fs = 1000, norm=None,noverlap = 100, nperseg=256)
            bi_spectrum = np.array(abs(bi_spectrum))  # calculate bi_spectrum
            bi_spectrum = 255 * (bi_spectrum - np.min(bi_spectrum)) / (np.max(bi_spectrum) - np.min(bi_spectrum))
            bi_spectrum = bi_spectrum.reshape((1, 256, 256))
            dataset.append(bi_spectrum)  # concat the dataset
    return dataset

import paddleimport paddle.nn as nnimport paddle.nn.functional as Ffrom paddle.vision.transforms import ToTensorimport numpy as npimport matplotlib.pyplot as pltimport copyfrom paddle.static import InputSpecclass RepMLP(nn.Layer):

    def __init__(self, in_channels, out_channels,
                 H, W, h, w,
                 reparam_conv_k=None,
                 fc1_fc2_reduction=1,
                 fc3_groups=1,
                 classes = 2,
                 deploy=False,):
        super(RepMLP, self).__init__()

        self.C = in_channels
        self.O = out_channels
        self.fc3_groups = fc3_groups

        self.H, self.W, self.h, self.w = H, W, h, w

        self.h_parts = self.H // self.h
        self.w_parts = self.W // self.w        assert self.H % self.h == 0
        assert self.W % self.w == 0
        self.target_shape = [-1, self.O, self.H, self.W]

        self.deploy = deploy
        self.classes = classes

        self.need_global_perceptron = (H != h) or (W != w)        if self.need_global_perceptron:
            internal_neurons = int(self.C * self.h_parts * self.w_parts // fc1_fc2_reduction)
            self.fc1_fc2 = nn.Sequential(
            ('fc1', nn.Linear(self.C * self.h_parts * self.w_parts, internal_neurons)),
            ('relu', nn.ReLU()),
            ('fc2', nn.Linear(internal_neurons, self.C * self.h_parts * self.w_parts))
            )            if deploy:
                self.avg = nn.AvgPool2d(kernel_size=(self.h, self.w))            else:
                self.avg = nn.Sequential(
                ('avg', nn.AvgPool2D(kernel_size=(self.h, self.w))),                # ('bn', nn.BatchNorm2D(self.C))
                )
                self.avg_bn = nn.BatchNorm2D(self.C)

        self.fc3 = nn.Conv2D(self.C * self.h * self.w, self.O * self.h * self.w, 1, 1, 0, bias_attr=deploy, groups=fc3_groups)
        self.fc3_bn = nn.Identity() if deploy else nn.BatchNorm1D(num_features = self.O * self.h * self.w)

        self.reparam_conv_k = reparam_conv_k        if not deploy and reparam_conv_k is not None:            for k in reparam_conv_k:
                conv_branch = nn.Sequential(
                ('conv', nn.Conv2D(in_channels=self.C, out_channels=self.O, kernel_size=k, padding=k // 2, bias_attr=False, groups=fc3_groups))                #, ('bn', nn.BatchNorm(self.O))
                )
                conv_branch_bn = nn.Sequential(
                    ('bn', nn.BatchNorm(self.O))
                    )
                self.__setattr__('repconv{}'.format(k), conv_branch)
                self.__setattr__('repconv_bn{}'.format(k), conv_branch_bn)
        self.end_avg = nn.AvgPool2D(kernel_size=(self.h, self.w))
        self.fc4 = nn.Linear(self.h_parts*self.w_parts*self.O, self.classes)    
    @paddle.jit.to_static(input_spec=[InputSpec(shape=[None, 1, 256, 256], dtype='float32')]) 
    def forward(self, inputs):
        if self.need_global_perceptron:
            v = self.avg(inputs)
            v = paddle.reshape(v, [-1, self.C * self.h_parts * self.w_parts])
            v = self.fc1_fc2(v)
            v = paddle.reshape(v, [-1, self.C, self.h_parts, 1, self.w_parts, 1])
            inputs = paddle.reshape(inputs, [-1, self.C, self.h_parts, self.h, self.w_parts, self.w])
            inputs = inputs + v        else:
            inputs = paddle.reshape(inputs, [-1, self.C, self.h_parts, self.h, self.w_parts, self.w])

        partitions = paddle.transpose(inputs, perm=[0, 2, 4, 1, 3, 5])        #   Feed partition map into Partition Perceptron
        fc3_inputs = paddle.reshape(partitions, [-1, self.C * self.h * self.w, 1, 1])
        fc3_out = self.fc3(fc3_inputs)
        fc3_out = paddle.reshape(fc3_out, [-1, self.O * self.h * self.w])
        fc3_out = self.fc3_bn(fc3_out)
        fc3_out = paddle.reshape(fc3_out, [-1, self.h_parts, self.w_parts, self.O, self.h, self.w])        #   Feed partition map into Local Perceptron
        if self.reparam_conv_k is not None and not self.deploy:
            conv_inputs = paddle.reshape(partitions, [-1, self.C, self.h, self.w])
            conv_out = 0
            for k in self.reparam_conv_k:
                conv_branch = self.__getattr__('repconv{}'.format(k))
                conv_branch_bn = self.__getattr__('repconv_bn{}'.format(k))
                conv_out += conv_branch_bn(conv_branch(conv_inputs))
            conv_out = paddle.reshape(conv_out, [-1, self.h_parts, self.w_parts, self.O, self.h, self.w])
            fc3_out += conv_out

        fc3_out = paddle.transpose(fc3_out, perm=[0, 3, 1, 4, 2, 5])  # N, O, h_parts, out_h, w_parts, out_w 
        fc3_out = paddle.reshape(fc3_out, self.target_shape)
        out = self.end_avg(fc3_out)
        out = paddle.reshape(out, [-1, self.h_parts*self.w_parts*self.O])
        out = self.fc4(out)        return outif __name__ == '__main__':
    N = 1
    C = 1
    H = 256
    W = 256
    h = 32
    w = 32
    O = 1
    groups = 1
    net = RepMLP(C, O, H=H, W=W, h=h, w=w, reparam_conv_k=(1,3,5), fc1_fc2_reduction=1, fc3_groups=groups, classes = 4, deploy=False)
    net.eval()
    params_info = paddle.summary(net, (1, 1, 256, 256), ['float32'])    print(params_info)

def train(train_loader, valid_loader, model, optim, loss, epoch_num, use_gpu):
    print('start training ... ')
    paddle.set_device('gpu:0') if use_gpu else paddle.set_device('cpu')
    val_loss_history = []
    val_acc_history = []    for epoch in range(epoch_num):
        model.train()        for batch_id, (data, label) in enumerate(train_loader()):
            x_data = paddle.to_tensor(data)
            y_data = paddle.to_tensor(label)
            y_data = paddle.unsqueeze(y_data, 1)
            logits = model(x_data)            # print('logits:', logits.shape)
            # print('y_data:', y_data.shape)
            # pred = paddle.argmax(logits)
            loss = F.cross_entropy(logits, y_data)            if (batch_id+1) % 10 == 0:                print("epoch: {}, batch_id: {}, loss is: {}".format(epoch, batch_id, loss.numpy()))
            optim.clear_grad()
            loss.backward()
            optim.step()        # evaluate model after one epoch
        model.eval()
        accuracies = []
        losses = []        for batch_id, (data, label) in enumerate(valid_loader()):
            x_data = paddle.to_tensor(data)
            y_data = paddle.to_tensor(label)
            y_data = paddle.unsqueeze(y_data, 1)

            logits = model(x_data)
            loss = F.cross_entropy(logits, y_data)
            acc = paddle.metric.accuracy(logits, y_data)
            accuracies.append(acc.numpy())
            losses.append(loss.numpy())

        avg_acc, avg_loss = np.mean(accuracies), np.mean(losses)        print("[validation] accuracy/loss: {}/{}".format(avg_acc, avg_loss))
        val_acc_history.append(avg_acc)
        val_loss_history.append(avg_loss)
    path = "./model_weights/model_2"
    paddle.jit.save(
    layer=model,
    path=path)    return val_acc_history, val_loss_history

country=CN
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
update_config=1
 network={
ssid="WIFI Name" # WIFI名称psk="*********" # WIFI密码key_mgmt=WPA-PSK
priority=1}

--arch: (armv8|armv7|armv7hf)   arm版本，默认为armv8
--toolchain: (gcc|clang)        编译器类型，默认为gcc
--with_extra: (OFF|ON)          是否编译OCR/NLP模型相关kernel&OP，默认为OFF，只编译CV模型相关kernel&OP
--with_python: (OFF|ON)         是否编译python预测库, 默认为 OFF
--python_version: (2.7|3.5|3.7) 编译whl的Python版本，默认为 None--with_cv: (OFF|ON)             是否编译CV相关预处理库, 默认为 OFF
--with_log: (OFF|ON)            是否输出日志信息, 默认为 ON
--with_exception: (OFF|ON)      是否在错误发生时抛出异常，默认为 OFF

inference_lite_lib.armlinux.armv7hf
├── cxx                                           C++ 预测库和头文件
│   ├── include                                   C++ 头文件
│   │   ├── paddle_api.h
│   │   ├── paddle_image_preprocess.h
│   │   ├── paddle_lite_factory_helper.h
│   │   ├── paddle_place.h
│   │   ├── paddle_use_kernels.h
│   │   ├── paddle_use_ops.h
│   │   └── paddle_use_passes.h
│   └── lib                                       C++ 预测库
│       ├── libpaddle_api_light_bundled.a         C++ 静态库
│       └── libpaddle_light_api_shared.so         C++ 动态库
├── demo
│   └── python                                    Python 预测库demo
│
└── python                                        Python 预测库(需要打开with_python选项)
    ├── install
    │   └── dist 			          Python whl包
    │       └── paddlelite-51fcbb609-cp37-cp37m-linux_armv7l.whl          
    └── lib
        └── lite.so                               Python 预测库

import paddlelite.lite as lite

a=lite.Opt()# 非combined形式a.set_model_dir("Your model file path")# conmbined形式，具体模型和参数名称，请根据实际修改# a.set_model_file("Your model file path")# a.set_param_file("Your param file path")a.set_optimize_out("model") # 设置模型文件名a.set_valid_places("arm")   # 设置为arma.run()

from paddlelite.lite import *import numpy as np

config = MobileConfig()
config.set_model_from_file("./model.nb") # 这里使用自己的*.nb文件路径

predictor = create_paddle_predictor(config)

image_data = np.random.rand(10, 1, 256, 256) # 这里我们先使用一组随机样本测试一下，这里可以替换成自己的读取数据代码。

input_tensor = predictor.get_input(0)
input_tensor.from_numpy(image_data)

predictor.run()

output_tensor = predictor.get_output(0)print(output_tensor.numpy())print(np.argmax(output_tensor.numpy(), axis=-1))

RST_PIN = 25CS_PIN = 8DC_PIN = 24KEY_UP_PIN = 6KEY_DOWN_PIN = 19KEY_LEFT_PIN = 5KEY_RIGHT_PIN = 26KEY_PRESS_PIN = 13KEY1_PIN = 21KEY2_PIN = 20KEY3_PIN = 16# 设置树莓派输入输出GPIO.setmode(GPIO.BCM)
GPIO.setup(KEY_UP_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY_DOWN_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY_LEFT_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY_RIGHT_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY_PRESS_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY1_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY2_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(KEY3_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

im_init = Image.open('./tupian/init.png')
show_image(im_init)
im_AS = Image.open('./tupian/AS.png')
im_MR = Image.open('./tupian/MR.png')
im_MS = Image.open('./tupian/MS.png')
im_MVP = Image.open('./tupian/MVP.png')
im_wait = Image.open('./tupian/wait.png')
im_taking = Image.open('./tupian/taking.png')
im_normal = Image.open('./tupian/normal.png')
im_abnormal = Image.open('./tupian/abnormal.png')
im_welcome = Image.open('./tupian/welcome.png')
im_testing = Image.open('./tupian/testing.png')

# Raspberry Pi pin configuration:RST = 27DC = 25BL = 24bus = 0device = 0# 240x240 display with hardware SPI:disp = ST7789.ST7789(SPI.SpiDev(bus, device),RST, DC, BL)# Initialize library.disp.Init()# Clear display.disp.clear()# 展示图片def show_image(photo,t=3.0):
    disp.ShowImage(photo,0,0)
    time.sleep(t)

 if not GPIO.input(KEY1_PIN) :
        show_image(im_taking,10)
        os.system('arecord --format S32_LE --rate 2000 -c 2 -d 10 ./sounds/test.wav')
        audio_path = './sounds/test.wav'
        show_image(im_wait,1)
        audio_data, fs = sf.read(audio_path)
        audio_data = band_pass_filter(audio_data, 2, 25, 400, fs)
        down_sample_audio_data = samplerate.resample(
            audio_data.T, 1000 / fs, converter_type='sinc_best').T
        down_sample_audio_data = down_sample_audio_data / np.max(np.abs(down_sample_audio_data))
        plot_signal(down_sample_audio_data, title=audio_path)
        im = Image.open('./hs_data/hs_img.jpg') 
        # show_image(im)
		freq1, fre2, bi_spectrum = polycoherence(
        down_sample_audio_data[-2500:], nfft=1024, fs=1000, norm=None, nperseg=256)
    bi_spectrum = np.array(abs(bi_spectrum))  # calculate bi_spectrum
    bi_spectrum = bi_spectrum.reshape((1, 256, 256))
    bi_spectrum = 255 * (bi_spectrum - np.min(bi_spectrum)) / \
        (np.max(bi_spectrum) - np.min(bi_spectrum))
    dataset = np.array([bi_spectrum])    # dataset = np.delete(dataset, 0, 0)
    dataset = dataset.astype('float32')

elif not GPIO.input(KEY2_PIN) :
    show_image(im_testing)
    config=MobileConfig()
    config.set_model_from_file("./model_2.nb")
    predictor=create_paddle_predictor(config)

    image_data=dataset
    input_tensor = predictor.get_input(0)
    input_tensor.from_numpy(image_data)
    predictor.run()
    output_tensor = predictor.get_output(0)    print(output_tensor.numpy())    print(np.argmax(output_tensor.numpy(),axis=-1)[0])    if np.argmax(output_tensor.numpy(),axis=-1)[0] == 0:
        show_image(im_normal)    else:
        show_image(im_abnormal)

elif not GPIO.input(KEY3_PIN):
    config = MobileConfig()
    config.set_model_from_file("./model_4.nb")
    predictor = create_paddle_predictor(config)

    image_data = dataset
    input_tensor = predictor.get_input(0)
    input_tensor.from_numpy(image_data)
    predictor.run()
    output_tensor = predictor.get_output(0)    print(output_tensor.numpy())    print(np.argmax(output_tensor.numpy(), axis=-1))    if np.argmax(output_tensor.numpy(),axis=-1)[0] == 0:
        show_image(im_AS)    elif np.argmax(output_tensor.numpy(),axis=-1)[0] == 1:
        show_image(im_MS)    elif np.argmax(output_tensor.numpy(),axis=-1)[0] == 2:
        show_image(im_MVP)    elif np.argmax(output_tensor.numpy(),axis=-1)[0] == 3:
        show_image(im_MR)

else:
    show_image(im_welcome,0.1)