# -*- coding: utf-8 -*-
import tensorflow as tf
import scipy.io.wavfile as wav
from python_speech_features import mfcc,delta
import os
import numpy as np
import sklearn.preprocessing
path_film = os.path.abspath('.')
path = path_film + "/data/xunlian/"
test_path = path_film + "/data/test_data/"
isnot_test_path = path_film + "/data/isnot_test_path/"
#使用one-hot编码，将离散特征的取值扩展到了欧式空间
#全局one-hot编码空间
label_binarizer = ""
def def_one_hot(x):
if label_binarizer == "":
binarizer = sklearn.preprocessing.LabelBinarizer()
else:
binarizer = label_binarizer
binarizer.fit(range(max(x)+1))
y= binarizer.transform(x)
return y
"""读取文件位置"""
def read_wav_path(path):
map_path, map_relative = [str(path) + str(x) for x in os.listdir(path) if os.path.isfile(str(path) + str(x))], [y for y in os.listdir(path)]
return map_path, map_relative
"""获得mfcc系数"""
def def_wav_read_mfcc(file_name):
fs, audio = wav.read(file_name)
processed_audio = mfcc(audio, samplerate=fs, nfft=512)
return processed_audio
"""获取输入的矩阵形状（大小）"""
def find_matrix_max_shape(audio):
h, l = 0, 0
for a in audio:
a, b = np.array(a).shape
if a > h:
h=a
if b > l:
l=b
return 700,l
def matrix_make_up(audio):
h, l = find_matrix_max_shape(audio)
new_audio = []
for aa in audio:
zeros_matrix = np.zeros([h, l],np.int8)
a, b = np.array(aa).shape
for i in range(a):
for j in range(b):
zeros_matrix[i, j]=zeros_matrix[i,j]+aa[i,j]
new_audio.append(zeros_matrix)
return new_audio,h,l
def read_wav_matrix(path):
map_path, map_relative = read_wav_path(path)
audio=[]
labels=[]
for idx, folder in enumerate(map_path):
processed_audio_delta = def_wav_read_mfcc(folder)
audio.append(processed_audio_delta)
labels.append(int(map_relative[idx].split(".")[0].split("_")[0]))
x_data,h,l = matrix_make_up(audio)
x_data = np.array(x_data)
# 得到文件夹内每种语音的one-hot编码
x_label = np.array(def_one_hot(labels))
return x_data, x_label, h, l
"""初始化权值"""
def weight_variable(shape,name):
initial = tf.truncated_normal(shape,stddev=0.01)#生成一个截断的正态分布
return tf.Variable(initial,name=name)
"""初始化偏置"""
def bias_variable(shape,name):
initial = tf.constant(0.01,shape=shape)
return tf.Variable(initial,name=name)
"""卷积层"""
def conv2d(x,W):
# x input tensor of shape `[batch, in_height, in_width, in_channels]`[训练时一个batch的图片数量, 图片高度, 图片宽度, 图像通道数]
# W filter / kernel tensor of shape [filter_height, filter_width, in_channels, out_channels][卷积核的高度，卷积核的宽度，图像通道数，卷积核个数]
#`strides[0] = strides[3] = 1`. strides[1]代表x方向的步长，strides[2]代表y方向的步长
# padding: A `string` from: `"SAME", "VALID"`
return tf.nn.conv2d(x,W,strides=[1,1,1,1],padding='SAME')
"""池化层"""
def max_pool_2x2(x):
#[池化的输入,池化窗口的大小[1, height, width, 1],步长[1, stride,stride, 1],卷积类型]
#ksize [1,x,y,1]
return tf.nn.max_pool(x,ksize=[1,2,2,1],strides=[1,2,2,1],padding='SAME')
"""训练模型函数"""
def Tranning(path,test_path):
x_train, y_train, h, l = read_wav_matrix(path)
x_test, y_test, h, l = read_wav_matrix(test_path)
# print(x_train.shape)
m,n = y_train.shape
# 命名空间
# 定义两个placeholder
x = tf.placeholder(tf.float32, [None, h, l], name='x-input')
y = tf.placeholder(tf.float32, [None, n], name='y-input')
# 改变x的格式转为4D的向量[batch, in_height, in_width, in_channels]`
x_image = tf.reshape(x, [-1, h, l, 1], name='x_image')
# 初始化第一个卷积层的权值和偏置
W_conv1 = weight_variable([5, 5, 1, 32], name='W_conv1') # 5*5的采样窗口，32个卷积核从3个平面抽取特征
b_conv1 = bias_variable([32], name='b_conv1') # 每一个卷积核一个偏置值
# 把x_image和权值向量进行卷积，再加上偏置值，然后应用于relu激活函数
conv2d_1 = conv2d(x_image, W_conv1) + b_conv1
h_conv1 = tf.nn.leaky_relu(conv2d_1)
print("第一次卷积后",h_conv1)
h_pool1 = max_pool_2x2(h_conv1) # 进行max-pooling
print("第一次池化后",h_pool1)
# 初始化第二个卷积层的权值和偏置
W_conv2 = weight_variable([5, 5, 32, 64], name='W_conv2') # 5*5的采样窗口，64个卷积核从32个平面抽取特征
b_conv2 = bias_variable([64], name='b_conv2') # 每一个卷积核一个偏置值
# 把h_pool1和权值向量进行卷积，再加上偏置值，然后应用于relu激活函数
conv2d_2 = conv2d(h_pool1, W_conv2) + b_conv2
h_conv2 = tf.nn.leaky_relu(conv2d_2)
print("第二次卷积后", h_conv2)
h_pool2 = max_pool_2x2(h_conv2) # 进行max-pooling
print("第二次池化后", h_pool2)
# 300*300的图片第一次卷积后还是300*300，第一次池化后变为150*150
# 第二次卷积后为150*150，第二次池化后变为了75*75
# 进过上面操作后得到64张7*7的平面
# 初始化第一个全连接层的权值
W_fc1 = weight_variable([175 * 4 * 64, 2048], name='W_fc1') # 上一场有75*75*64个神经元，全连接层有1024个神经元
b_fc1 = bias_variable([2048], name='b_fc1') # 1024个节点
# 把池化层2的输出扁平化为1维
h_pool2_flat = tf.reshape(h_pool2, [-1, 175 * 4 * 64], name='h_pool2_flat')
# 求第一个全连接层的输出
wx_plus_b1 = tf.matmul(h_pool2_flat, W_fc1) + b_fc1
h_fc1 = tf.nn.leaky_relu(wx_plus_b1)
# keep_prob用来表示神经元的输出概率
keep_prob = tf.placeholder(tf.float32, name='keep_prob')
h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob, name='h_fc1_drop')
# 初始化第二个全连接层
W_fc2 = weight_variable([2048, n], name='W_fc2')
b_fc2 = bias_variable([n], name='b_fc2')
wx_plus_b2 = tf.matmul(h_fc1_drop, W_fc2) + b_fc2
# 计算输出
prediction = tf.nn.leaky_relu(wx_plus_b2)
tf.add_to_collection('predictions', prediction)
#向量归一化为（0，1）之间的值,即各个类的概率
p = tf.nn.softmax(wx_plus_b2)
tf.add_to_collection('p', p)
# 交叉熵代价函数
cross_entropy = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(labels=y, logits=prediction),
name='cross_entropy')
# 使用AdamOptimizer进行优化
train_step = tf.train.AdamOptimizer(epsilon=1e-5).minimize(cross_entropy)
# train_step = tf.train.GradientDescentOptimizer(5).minimize(cross_entropy)
# 求准确率
# 结果存放在一个布尔列表中
correct_prediction = tf.equal(tf.argmax(prediction, 1), tf.argmax(y, 1)) # argmax返回一维张量中最大的值所在的位置
# 求准确率
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
#保存模型使用环境
saver = tf.train.Saver()
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
# 创建一个协调器，管理线程
coord = tf.train.Coordinator()
# 启动QueueRunner, 此时文件名队列已经进队
threads = tf.train.start_queue_runners(sess=sess, coord=coord)
for i in range(100):
# 训练模型
sess.run(train_step, feed_dict={x: x_train, y: y_train, keep_prob: 1.0})
test_acc = sess.run(accuracy, feed_dict={x: x_test, y: y_test, keep_prob: 1.0})
train_acc = sess.run(accuracy, feed_d
