Tensorflow_Discuz验证码识别

1. 前言

  • 本文将使用深度学习框架Tensorflow训练出一个用于破解Discuz验证码的模型。
  • 环境:GTX1080的显卡,GPU,tensorflow_gpu版本,CUDA,CUDNN
  • GitHub地址:链接
  • Discuz验证码数据集下载地址:

2. Discuz验证码识别

  • 数据集大致如下所示,都是100*30大小的图片:

2.1 数据预处理

  • 首先,数据预处理分为两个部分,第一部分是读取图片,并划分训练集和测试集。因为整个数据集为6W张图片,所以我们可以让训练集为5W张,测试集为1W张。
  • 标签是文件名,我们认识,但是机器是不认识的,因此我们要使用text2vec,将标签进行向量化。

  • 读取数据

    • 通过定义rate,来划分数据集比例,例如:测试集1w,训练集5w,rate=1/5=0.2,划分代码如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def get_imgs(rate = 0.2):
"""
获取图片,并划分训练集和测试集
Parameters:
rate:测试集和训练集的比例,即测试集个数/训练集个数
Returns:
test_imgs:测试集
test_labels:测试集标签
train_imgs:训练集
test_labels:训练集标签
"""
data_path = './Discuz'
# 读取图片
imgs = os.listdir(data_path)
# 打乱图片顺序
random.shuffle(imgs)
# 数据集总共个数
imgs_num = len(imgs)
# 按照比例求出测试集个数
test_num = int(imgs_num * rate / (1 + rate))
# 测试集
test_imgs = imgs[:test_num]
# 根据文件名获取测试集标签:map和lamda的使用方法自行搜索
test_labels = list(map(lambda x: x.split('.')[0], test_imgs))
# 训练集
train_imgs = imgs[test_num:]
# 根据文件名获取训练集标签
train_labels = list(map(lambda x: x.split('.')[0], train_imgs))
return test_imgs, test_labels, train_imgs, train_labels
  • 标签向量化
    • 定义下面两个函数,一个是标签文本转向量,另一个是向量转文本
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import numpy as np
def text2vec(text):
"""
文本转向量
Parameters:
text:文本
Returns:
vector:向量
"""
if len(text) > 4:
raise ValueError('验证码最长4个字符')

vector = np.zeros(4 * 63)
def char2pos(c):
if c =='_':
k = 62
return k
k = ord(c) - 48
if k > 9:
k = ord(c) - 55
if k > 35:
k = ord(c) - 61
if k > 61:
raise ValueError('No Map')
return k
for i, c in enumerate(text):
idx = i * 63 + char2pos(c)
vector[idx] = 1
return vector

def vec2text(vec):
"""
向量转文本
Parameters:
vec:向量
Returns:
文本
"""
char_pos = vec.nonzero()[0]
text = []
for i, c in enumerate(char_pos):
char_at_pos = i #c/63
char_idx = c % 63
if char_idx < 10:
char_code = char_idx + ord('0')
elif char_idx < 36:
char_code = char_idx - 10 + ord('A')
elif char_idx < 62:
char_code = char_idx - 36 + ord('a')
elif char_idx == 62:
char_code = ord('_')
else:
raise ValueError('error')
text.append(chr(char_code))
return "".join(text)

print(text2vec('abcd'))
print(vec2text(text2vec('abcd')))
  • 运行结果:

  • 这里包括了63个字符的转化,0-9 a-z A-Z _(验证码如果小于4,用_补齐)。

2.2 根据batch_size获取数据

在训练模型的时候,需要根据不同的batch_size”喂”数据。这就需要我们写个函数,从整体数据集中获取指定batch_size大小的数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
"""
获得batch_size大小的数据集
Parameters:
batch_size:batch_size大小
train_flag:是否从训练集获取数据
Returns:
batch_x:大小为batch_size的数据x
batch_y:大小为batch_size的数据y
"""
def get_next_batch(self,train_flag = True, batch_size=100):
# 训练集中获取数据集
if train_flag == True:
# 如果还有剩余的batch_size个数据
if (batch_size + self.train_ptr) < self.train_size:
trains = train_imgs[self.train_ptr:(self.train_ptr + batch_size)]
labels = train_labels[self.train_ptr:(self.train_ptr + batch_size)]
self.train_ptr += batch_size
# 如果剩余不足batch_size个数据,则还要加上前面的某些个数据凑数
else:
new_ptr = (self.train_ptr + batch_size) % self.train_size
trains = train_imgs[self.train_ptr:] + train_imgs[:new_ptr]
labels = train_labels[self.train_ptr:] + train_labels[:new_ptr]
self.train_ptr = new_ptr
batch_x = np.zeros([batch_size,self.width * self.heigth])
batch_y = np.zeros([max_captcha * self.char_set_len])
# enumerate方法用来遍历trains数据,index是对应数据的下标,train是数据
for index, train in enumerate(trains):
# cv2.imread读取彩色图片返回加载的图像,mean方法是求均值,-1的作用目前不知道
img = np.mean(cv2.imread(self.data_path + train),-1)
# 将多维转为1维:batch_x从开始一个一个的数据,flatten方法按行的方向降维
batch_x[index,:] = img.flatten() / 255
for index, label in enumerate(labels):
batch_y[index,:] = self.text2vec(label)
# 测试集中获取数据
else:
if (batch_size + self.test_ptr) < self.test_size:
tests = self.test_imgs[self.test_ptr:(self.test_ptr + batch_size)]
labels = self.test_labels[self.test_ptr:(self.test_ptr + batch_size)]
self.test_ptr += batch_size
else:
new_ptr = (self.test_ptr + batch_size) % self.test_size
tests = self.test_imgs[self.test_ptr:] + self.test_imgs[:new_ptr]
labels = self.test_labels[self.test_ptr:] + self.test_labels[:new_ptr]
self.test_ptr = new_ptr
batch_x = np.zeros([batch_size, self.heigth*self.width])
batch_y = np.zeros([batch_size, self.max_captcha*self.char_set_len])
for index, test in enumerate(tests):
img = np.mean(cv2.imread(self.data_path + test), -1)
# 将多维降维1维
batch_x[index,:] = img.flatten() / 255
for index, label in enumerate(labels):
batch_y[index,:] = self.text2vec(label)
return batch_x,batch_y
  • 我们通过train_flag来确定是从训练集获取数据还是测试集获取数据,通过batch_size来获取指定大小的数据。获取数据之后,将batch_size大小的图片数据和经过向量化处理的标签存放到numpy数组中。

3. CNN网络模型

  • 网络模型由3层卷积层+1层全连接层组成,代码如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
# 定义cnn网络模型,w_alpha:权重系数,b_alpha:偏置系数
def crack_captcha_cnn(self, w_alpha=0.01, b_alpha=0.1):
# 卷积的input: 一个Tensor。数据维度是四维[batch, in_height, in_width, in_channels]
# 具体含义是[batch大小, 图像高度, 图像宽度, 图像通道数]
# 因为是灰度图,所以是单通道的[?, 100, 30, 1].-1表示该参数会由函数自动计算
x = tf.reshape(self.X, shape = [-1, 100, 30, 1])
# 卷积的filter:一个Tensor。数据维度是四维[filter_height, filter_width, in_channels, out_channels]
# 具体含义是[卷积核的高度, 卷积核的宽度, 图像通道数, 卷积核个数]
# random_normal用于从服从指定正太分布的数值中取出随机数
# b_alpha和w_alpha:给一些偏差防止死亡节点
w_c1 = tf.Variable(w_alpha * tf.random_normal([3, 3, 1, 32]))
# 偏置项bias
b_c1 = tf.Variable(b_alpha * tf.random_normal([32]))
# conv2d卷积层输入:
# strides: 一个长度是4的一维整数类型数组,每一维度对应的是 input 中每一维的对应移动步数
# padding:一个字符串,取值为 SAME 或者 VALID 前者使得卷积后图像尺寸不变, 后者尺寸变化
# conv2d卷积层输出:
# 一个四维的Tensor, 数据维度为 [batch, out_width, out_height, in_channels * out_channels]
# [?, 100, 30, 32]
# 输出计算公式H0 = (H - F + 2 * P) / S + 1
# 对于本卷积层而言,因为padding为SAME,所以P为1。
# 其中H为图像高度,F为卷积核高度,P为边填充,S为步长
# 学习参数:
# 32*(3*3+1)=320
# 连接个数:
# (输出图像宽度*输出图像高度)(卷积核高度*卷积核宽度+1)*卷积核数量(100*30)(3*3+1)*32=100*30*320=960000个
# bias_add:将偏差项bias加到value上。这个操作可以看做是tf.add的一个特例,其中bias是必须的一维。
# 该API支持广播形式,因此value可以是任何维度。但是,该API又不像tf.add,可以让bias的维度和value的最后一维不同
conv1 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(x,w_c1,strides=[1,1,1,1],padding = 'SAME'),b_c1))
# max_pool池化层输入:
# ksize:池化窗口的大小,取一个四维向量,一般是[1, height, width, 1]
# 因为我们不想在batch和channels上做池化,所以这两个维度设为了1
# strides:和卷积类似,窗口在每一个维度上滑动的步长,一般也是[1, stride,stride, 1]
# padding:和卷积类似,可以取'VALID' 或者'SAME'。same是指在移动池化核时,若数据不够,则补0继续移动
# max_pool池化层输出:
# 返回一个Tensor,类型不变,shape仍然是[batch, out_width, out_height, in_channels]这种形式
# [?, 50, 15, 32]
# 学习参数:
# 2*32
# 连接个数:
# 15*50*32*(2*2+1)=120000
conv1 = tf.nn.max_pool(conv1, ksize = [1,2,2,1],strides =[1,2,2,1],padding='SAME')
# dropout层
#conv1 = tf.nn.dropout(conv1,self.keep_prob)

# 第二个卷积层
w_c2 = tf.Variable(w_alpha * tf.random_normal([3,3,32,64]))
b_c2 = tf.Variable(b_alpha * tf.random_normal([64]))
# 卷机之后为[?, 50, 15, 64]
conv2 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(conv1, w_c2, strides =[1,1,1,1], padding='SAME'),b_c2))
# 池化之后为[?, 25, 8, 64]:因为15不能整除2,所以padding补0,最后得到8
conv2 = tf.nn.max_pool(conv2, ksize = [1,2,2,1],strides= [1,2,2,1],padding='SAME')
# 若有dropout层:
# conv2 = tf.nn.dropout(conv2,self.keep_prob)
# 第三个卷积层
w_c3 = tf.Variable(w_alpha * tf.random_normal([3,3,64,64]))
b_c3 = tf.Variable(b_alpha * tf.random_normal([64]))
# 卷积之后:[?, 25, 8, 64]
conv3 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(conv2, w_c3, strides=[1,1,1,1], padding='SAME'),b_c2))
# 池化之后为:[?,13,4,64]
conv3 = tf.nn.max_pool(conv3, ksize = [1,2,2,1],strides= [1,2,2,1],padding='SAME')
#conv3 = tf.nn.dropout(conv3, self.keep_prob)

# 全连接层
# [3328,1024]
w_d = tf.Variable(w_alpha * tf.random_normal([4*13*64, 1024]))
b_d = tf.Variable(b_alpha * tf.random_normal([1024]))
#[? 3328] 将conv3降维为1*3328形状,作为全连接层的输入
dense = tf.reshape(conv3, [-1, w_d.get_shape().as_list()[0]])
# [? 1024]
dense = tf.nn.relu(tf.add(tf.matmul(dense, w_d),b_d))
dense = tf.nn.dropout(dense, self.keep_prob)

# 输出
w_out = tf.Variable(w_alpha * tf.random_normal([1024, self.max_captcha*self.char_set_len]))
b_out = tf.Variable(b_alpha * tf.random_normal([self.max_captcha*self.char_set_len]))
# 输出 [? , 252]
out = tf.add(tf.matmul(dense, w_out),b_out)

return out

4. 训练

  • 数据+网络模型已准备好,开始训练。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# 训练函数
def train_crack_captcha_cnn(self):
output = self.crack_captcha_cnn()
# 损失函数
diff = tf.nn.sigmoid_cross_entropy_with_logits(logits=output, labels= self.Y)
loss = tf.reduce_mean(diff)
# 计算损失,并使用tf.summary.scaler记录
tf.summary.scalar('loss',loss)

# 使用AdamOptimizer优化器训练模型,最小化交叉熵损失
optimizer = tf.train.AdamOptimizer(learning_rate=0.01).minimize(loss)

# 计算准确率
y = tf.reshape(output, [-1, self.max_captcha, self.char_set_len])
y_ = tf.reshape(self.Y, [-1, self.max_captcha, self.char_set_len])
correct_pred = tf.equal(tf.argmax(y, 2),tf.argmax(y_, 2))
accuracy = tf.reduce_mean(tf.cast(correct_pred,tf.float32))
# 计算准确率,并使用tf.summary.scaler记录
tf.summary.scalar('accuracy',accuracy)

# 将summary合并
merged = tf.summary.merge_all()
saver = tf.train.Saver()
with tf.Session(config = self.config) as sess:
# 将运行后输出的数据写到指定的磁盘路径中
train_writer = tf.summary.FileWriter(self.log_dir + '/train', sess.graph)
test_writer = tf.summary.FileWriter(self.log_dir + '/test')
sess.run(tf.global_variables_initializer())
# 遍历self.max_steps次
for i in range(self.max_steps):
# 迭代500次,打乱一下数据集
if i % 499 == 0:
self.test_imgs, self.test_labels, self.train_imgs, self.train_labels = self.get_imgs()
# 每10次,使用测试集,测试一下准确率
if i % 10 == 0: # 记录测试集的summary与accuracy
batch_x_test, batch_y_test = self.get_next_batch(False, 100)
summary, acc = sess.run([merged, accuracy], feed_dict={self.X: batch_x_test, self.Y: batch_y_test, self.keep_prob: 1})
print("迭代第%d次 accuracy:%f" %(i+1, acc))
test_writer.add_summary(summary,i)
# 如果模型准确率大于0.95,则保存并退出
if acc > 0.95:
train_writer.close()
test_writer.close()
saver.save(sess, "crack_capcha.model", global_step=i)
break
else:
batch_x, batch_y = self.get_next_batch(True,100)
loss_value, _ = sess.run([loss, optimizer], feed_dict={self.X: batch_x, self.Y: batch_y, self.keep_prob: 1})
print('迭代第%d次 loss:%f' % (i+1, loss_value))
curve = sess.run(merged, feed_dict={self.X: batch_x_test, self.Y: batch_y_test, self.keep_prob: 1})
train_writer.add_summary(curve, i)

train_writer.close()
test_writer.close()
saver.save(sess, "crack_capcha.model", global_step=self.max_steps)
  • 上述代码中涉及到summary的方法都是有关Tensorboard的操作
  • 在迭代到500次的时候重新获取下数据集,这样做其实就是打乱了一次数据集。为什么要打乱数据集呢?因为如果不打乱数据集,在训练的时候,Tensorboard绘图会有如下现象:

  • 可以看到,准确率曲线和Loss曲线存在跳变,这就是因为我们没有在迭代一定次数之后打乱数据集造成的。

5. 整体的训练代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np
import os,random,cv2

class Discuz():
def __init__(self):
# 指定GPU
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
self.config = tf.ConfigProto(allow_soft_placement = true)
gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction = 1)
self.config.gpu_options.allow_growth = True
# 数据集路径
self.data_path = './Discuz/'
# 将log写入指定磁盘路径
self.log_dir = './Crack_Discuz/'
# 数据集大小
self.width = 30
self.heigth = 100
# 最大迭代次数
self.max_steps = 1000000
# 读取数据集
self.test_imgs, self.test_labels, self.train_imgs, self.train_labels = self.get_imgs()
# 训练集大小
self.train_size = len(self.train_imgs)
# 测试集大小
self.test_size = len(self.test_imgs)
# 每次获得batch_size大小的当前训练集指针
self.train_ptr = 0
# 每次获取batch_size大小的当前测试集指针
self.test_ptr = 0
# 字符字典大小:0-9 a-z A-Z _(验证码如果小于4,用_补齐) 一共63个字符
self.char_set_len = 63
# 验证码最长的长度为4
self.max_captcha = 4
# 输入X的占位符
self.X = tf.placeholder(tf.float32, [None, self.width*self.heigth])
# 输入Y的占位符
self.Y = tf.placeholder(tf.float32, [None, self.char_set_len*self.max_captcha])
# keepprob占位符
self.keep_prob = tf.placeholder(tf.float32)


"""
获取图片,并划分训练集和测试集
Parameters:
rate:测试集和训练集的比例,即测试集个数/训练集个数
Returns:
test_imgs:测试集
test_labels:测试集标签
train_imgs:训练集
test_labels:训练集标签
"""
def get_imgs(self, rate = 0.2):
# 读取图片
imgs = os.listdir(self.data_path)
# 打乱图片顺序
random.shuffle(imgs)
# 数据集总共个数
imgs_num = len(imgs)
# 按照比例求出测试集个数
test_num = int(imgs_num * rate / (1 + rate))
# 测试集
test_imgs = imgs[:test_num]
# 根据文件名获取测试集标签
test_labels = list(map(lambda x: x.split('.')[0], test_imgs))
# 训练集
train_imgs = imgs[test_num:]
# 根据文件名获取训练集标签
train_labels = list(map(lambda x: x.split('.')[0], train_imgs))
return test_imgs, test_labels, train_imgs, train_labels



"""
文本转向量
Parameters:
text:文本
Returns:
vector:向量
"""
def text2vec(text):
if len(text) > 4:
raise ValueError('验证码最长4个字符')
vector = np.zeros(4 * 63)
def char2pos(c):
if c =='_':
k = 62
return k
k = ord(c) - 48
if k > 9:
k = ord(c) - 55
if k > 35:
k = ord(c) - 61
if k > 61:
raise ValueError('No Map')
return k
for i, c in enumerate(text):
idx = i * 63 + char2pos(c)
vector[idx] = 1
return vector


"""
向量转文本
Parameters:
vec:向量
Returns:
文本
"""
def vec2text(vec):
char_pos = vec.nonzero()[0]
text = []
for i, c in enumerate(char_pos):
char_at_pos = i #c/63
char_idx = c % 63
if char_idx < 10:
char_code = char_idx + ord('0')
elif char_idx < 36:
char_code = char_idx - 10 + ord('A')
elif char_idx < 62:
char_code = char_idx - 36 + ord('a')
elif char_idx == 62:
char_code = ord('_')
else:
raise ValueError('error')
text.append(chr(char_code))
return "".join(text)




"""
获得batch_size大小的数据集
Parameters:
batch_size:batch_size大小
train_flag:是否从训练集获取数据
Returns:
batch_x:大小为batch_size的数据x
batch_y:大小为batch_size的数据y
"""
def get_next_batch(self,train_flag = True, batch_size=100):
# 训练集中获取数据集
if train_flag == True:
# 如果还有剩余的batch_size个数据
if (batch_size + self.train_ptr) < self.train_size:
trains = train_imgs[self.train_ptr:(self.train_ptr + batch_size)]
labels = train_labels[self.train_ptr:(self.train_ptr + batch_size)]
self.train_ptr += batch_size
# 如果剩余不足batch_size个数据,则还要加上前面的某些个数据凑数
else:
new_ptr = (self.train_ptr + batch_size) % self.train_size
trains = train_imgs[self.train_ptr:] + train_imgs[:new_ptr]
labels = train_labels[self.train_ptr:] + train_labels[:new_ptr]
self.train_ptr = new_ptr
batch_x = np.zeros([batch_size,self.width * self.heigth])
batch_y = np.zeros([max_captcha * self.char_set_len])
# enumerate方法用来遍历trains数据,index是对应数据的下标,train是数据
for index, train in enumerate(trains):
# cv2.imread读取彩色图片返回加载的图像,mean方法是求均值,-1的作用目前不知道
img = np.mean(cv2.imread(self.data_path + train),-1)
# 将多维转为1维:batch_x从开始一个一个的数据,flatten方法按行的方向降维
batch_x[index,:] = img.flatten() / 255
for index, label in enumerate(labels):
batch_y[index,:] = self.text2vec(label)
# 测试集中获取数据
else:
if (batch_size + self.test_ptr) < self.test_size:
tests = self.test_imgs[self.test_ptr:(self.test_ptr + batch_size)]
labels = self.test_labels[self.test_ptr:(self.test_ptr + batch_size)]
self.test_ptr += batch_size
else:
new_ptr = (self.test_ptr + batch_size) % self.test_size
tests = self.test_imgs[self.test_ptr:] + self.test_imgs[:new_ptr]
labels = self.test_labels[self.test_ptr:] + self.test_labels[:new_ptr]
self.test_ptr = new_ptr
batch_x = np.zeros([batch_size, self.heigth*self.width])
batch_y = np.zeros([batch_size, self.max_captcha*self.char_set_len])
for index, test in enumerate(tests):
img = np.mean(cv2.imread(self.data_path + test), -1)
# 将多维降维1维
batch_x[index,:] = img.flatten() / 255
for index, label in enumerate(labels):
batch_y[index,:] = self.text2vec(label)
return batch_x,batch_y


# 定义cnn网络模型,w_alpha:权重系数,b_alpha:偏置系数
def crack_captcha_cnn(self, w_alpha=0.01, b_alpha=0.1):
# 卷积的input: 一个Tensor。数据维度是四维[batch, in_height, in_width, in_channels]
# 具体含义是[batch大小, 图像高度, 图像宽度, 图像通道数]
# 因为是灰度图,所以是单通道的[?, 100, 30, 1].-1表示该参数会由函数自动计算
x = tf.reshape(self.X, shape = [-1, 100, 30, 1])
# 卷积的filter:一个Tensor。数据维度是四维[filter_height, filter_width, in_channels, out_channels]
# 具体含义是[卷积核的高度, 卷积核的宽度, 图像通道数, 卷积核个数]
# random_normal用于从服从指定正太分布的数值中取出随机数
# b_alpha和w_alpha:给一些偏差防止死亡节点
w_c1 = tf.Variable(w_alpha * tf.random_normal([3, 3, 1, 32]))
# 偏置项bias
b_c1 = tf.Variable(b_alpha * tf.random_normal([32]))
# conv2d卷积层输入:
# strides: 一个长度是4的一维整数类型数组,每一维度对应的是 input 中每一维的对应移动步数
# padding:一个字符串,取值为 SAME 或者 VALID 前者使得卷积后图像尺寸不变, 后者尺寸变化
# conv2d卷积层输出:
# 一个四维的Tensor, 数据维度为 [batch, out_width, out_height, in_channels * out_channels]
# [?, 100, 30, 32]
# 输出计算公式H0 = (H - F + 2 * P) / S + 1
# 对于本卷积层而言,因为padding为SAME,所以P为1。
# 其中H为图像高度,F为卷积核高度,P为边填充,S为步长
# 学习参数:
# 32*(3*3+1)=320
# 连接个数:
# (输出图像宽度*输出图像高度)(卷积核高度*卷积核宽度+1)*卷积核数量(100*30)(3*3+1)*32=100*30*320=960000个
# bias_add:将偏差项bias加到value上。这个操作可以看做是tf.add的一个特例,其中bias是必须的一维。
# 该API支持广播形式,因此value可以是任何维度。但是,该API又不像tf.add,可以让bias的维度和value的最后一维不同
conv1 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(x,w_c1,strides=[1,1,1,1],padding = 'SAME'),b_c1))
# max_pool池化层输入:
# ksize:池化窗口的大小,取一个四维向量,一般是[1, height, width, 1]
# 因为我们不想在batch和channels上做池化,所以这两个维度设为了1
# strides:和卷积类似,窗口在每一个维度上滑动的步长,一般也是[1, stride,stride, 1]
# padding:和卷积类似,可以取'VALID' 或者'SAME'。此处的same是指在移动池化核时,若数据不够,则补0继续移动
# max_pool池化层输出:
# 返回一个Tensor,类型不变,shape仍然是[batch, out_width, out_height, in_channels]这种形式
# [?, 50, 15, 32]
# 学习参数:
# 2*32
# 连接个数:
# 15*50*32*(2*2+1)=120000
conv1 = tf.nn.max_pool(conv1, ksize = [1,2,2,1],strides =[1,2,2,1],padding='SAME')
# dropout层
#conv1 = tf.nn.dropout(conv1,self.keep_prob)

# 第二个卷积层
w_c2 = tf.Variable(w_alpha * tf.random_normal([3,3,32,64]))
b_c2 = tf.Variable(b_alpha * tf.random_normal([64]))
# 卷机之后为[?, 50, 15, 64]
conv2 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(conv1, w_c2, strides =[1,1,1,1], padding='SAME'),b_c2))
# 池化之后为[?, 25, 8, 64]:因为15不能整除2,所以padding补0,最后得到8
conv2 = tf.nn.max_pool(conv2, ksize = [1,2,2,1],strides= [1,2,2,1],padding='SAME')
# 若有dropout层:
# conv2 = tf.nn.dropout(conv2,self.keep_prob)
# 第三个卷积层
w_c3 = tf.Variable(w_alpha * tf.random_normal([3,3,64,64]))
b_c3 = tf.Variable(b_alpha * tf.random_normal([64]))
# 卷积之后:[?, 25, 8, 64]
conv3 = tf.nn.relu(tf.bias_add(tf.nn.conv2d(conv2, w_c3, strides=[1,1,1,1], padding='SAME'),b_c2))
# 池化之后为:[?,13,4,64]
conv3 = tf.nn.max_pool(conv3, ksize = [1,2,2,1],strides= [1,2,2,1],padding='SAME')
#conv3 = tf.nn.dropout(conv3, self.keep_prob)

# 全连接层
# [3328,1024]
w_d = tf.Variable(w_alpha * tf.random_normal([4*13*64, 1024]))
b_d = tf.Variable(b_alpha * tf.random_normal([1024]))
#[? 3328] 将conv3降维为1*3328形状,作为全连接层的输入
dense = tf.reshape(conv3, [-1, w_d.get_shape().as_list()[0]])
# [? 1024]
dense = tf.nn.relu(tf.add(tf.matmul(dense, w_d),b_d))
dense = tf.nn.dropout(dense, self.keep_prob)

# 输出
w_out = tf.Variable(w_alpha * tf.random_normal([1024, self.max_captcha*self.char_set_len]))
b_out = tf.Variable(b_alpha * tf.random_normal([self.max_captcha*self.char_set_len]))
# 输出 [? , 252]
out = tf.add(tf.matmul(dense, w_out),b_out)

return out


# 训练函数
def train_crack_captcha_cnn(self):
output = self.crack_captcha_cnn()
# 损失函数
diff = tf.nn.sigmoid_cross_entropy_with_logits(logits=output, labels= self.Y)
loss = tf.reduce_mean(diff)
# 计算损失,并使用tf.summary.scaler记录
tf.summary.scalar('loss',loss)

# 使用AdamOptimizer优化器训练模型,最小化交叉熵损失
optimizer = tf.train.AdamOptimizer(learning_rate=0.01).minimize(loss)

# 计算准确率
y = tf.reshape(output, [-1, self.max_captcha, self.char_set_len])
y_ = tf.reshape(self.Y, [-1, self.max_captcha, self.char_set_len])
correct_pred = tf.equal(tf.argmax(y, 2),tf.argmax(y_, 2))
accuracy = tf.reduce_mean(tf.cast(correct_pred,tf.float32))
# 计算准确率,并使用tf.summary.scaler记录
tf.summary.scalar('accuracy',accuracy)

# 将summary合并
merged = tf.summary.merge_all()
saver = tf.train.Saver()
with tf.Session(config = self.config) as sess:
# 将运行后输出的数据写到指定的磁盘路径中
train_writer = tf.summary.FileWriter(self.log_dir + '/train', sess.graph)
test_writer = tf.summary.FileWriter(self.log_dir + '/test')
sess.run(tf.global_variables_initializer())
# 遍历self.max_steps次
for i in range(self.max_steps):
# 迭代500次,打乱一下数据集
if i % 499 == 0:
self.test_imgs, self.test_labels, self.train_imgs, self.train_labels = self.get_imgs()
# 每10次,使用测试集,测试一下准确率
if i % 10 == 0: # 记录测试集的summary与accuracy
batch_x_test, batch_y_test = self.get_next_batch(False, 100)
summary, acc = sess.run([merged, accuracy], feed_dict={self.X: batch_x_test, self.Y: batch_y_test, self.keep_prob: 1})
print("迭代第%d次 accuracy:%f" %(i+1, acc))
test_writer.add_summary(summary,i)
# 如果模型准确率大于0.85,则保存并退出
if acc > 0.85:
train_writer.close()
test_writer.close()
saver.save(sess, "crack_capcha.model", global_step=i)
break
else:
batch_x, batch_y = self.get_next_batch(True,100)
loss_value, _ = sess.run([loss, optimizer], feed_dict={self.X: batch_x, self.Y: batch_y, self.keep_prob: 1})
print('迭代第%d次 loss:%f' % (i+1, loss_value))
curve = sess.run(merged, feed_dict={self.X: batch_x_test, self.Y: batch_y_test, self.keep_prob: 1})
train_writer.add_summary(curve, i)

train_writer.close()
test_writer.close()
saver.save(sess, "crack_capcha.model", global_step=self.max_steps)

if __name__ == '__main__':
dz = Discuz
dz.train_crack_captcha_cnn()

5. 训练结果

  • 环境:GTX1080的显卡,GPU,tensorflow_gpu版本,CUDA,CUDNN。

  • 使用tensorboard查看结果:

  • 这里遇到了一个关于tensorboard的问题,在服务器端训练结束之后,tensorboard有关记录文件在服务器端,如果这时在服务器端使用tensorboard命令,则必须先将服务器端的该端口映射到本地来,才能够访问,但是博主没有服务器的权限,所以没有办法映射。

    • 记录一下:在服务器端运行 tensorboard --logdir='./Crack_Discuz' --port 8008命令,其次映射到本地:ssh -L 8008:localhost:8008 username@user_ip
    • 最后在本地浏览器里访问 localhost:8008 即可访问远程的tensorboard。
  • 解决办法:将训练完成的文件拷贝到本地,本地只需安装tensorboard即可使用。

    • 安装:pip3 install tensorboard
    • 在log文件目录下执行tensorboard --logdir='./Crack_Discuz' --port 8008
    • 即可在本地8008端口访问
    • 下面是tensorboard用到的文件,类似这种的文件是tensorboard文件。

  • 访问8008端口,查看结果

  • 训练的acc和loss,因为设置的是大于85%就停止,所以只迭代了12w次:

  • test的acc和loss:

6. 利用已有模型测试

  • 训练好的模型文件如下:

  • 加载已经训练好的模型进行预测,在和train.py相同目录下,创建test.py文件,添加如下代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
#-*- coding:utf-8 -*-
import tensorflow as tf
import numpy as np
import train

"""
使用模型做预测
Parameters:
captcha_image:数据
captcha_label:标签
"""

def crack_captcha(captcha_image, captcha_label):
# 输出了前向传播结果,但是还未给数据的值初始化,所以这是获取了一个计算图
output = dz.crack_captcha_cnn()
saver = tf.train.Saver()
with tf.Session(config=dz.config) as sess:
# 从当前目录加载模型
saver.restore(sess, tf.train.latest_checkpoint('.'))
for i in range(len(captcha_label)):
img = captcha_image[i].flatten() #把训练图片转成了一维数组
label = captcha_label[i]
# 2 表示仅在CHAR_SET_LEN,即标签维度上去取最大值。
# 输出的是一个长度为batch的一维数组,这个一维数组中的值就表示了每一个样例对应的数字结果。
predict = tf.argmax(tf.reshape(output, [-1, dz.max_captcha, dz.char_set_len]), 2)
text_list = sess.run(predict, feed_dict={dz.X: [img], dz.keep_prob: 1})
text = text_list[0].tolist()
vector = np.zeros(dz.max_captcha*dz.char_set_len)
i = 0
for n in text:
vector[i*dz.char_set_len + n] = 1
i += 1
prediction_text = dz.vec2text(vector)
print("正确: {} 预测: {}".format(dz.vec2text(label), prediction_text))

if __name__ == '__main__':
dz = train.Discuz()
batch_x, batch_y = dz.get_next_batch(False, 5)
crack_captcha(batch_x, batch_y)
  • 测试结果:

-------------The End-------------
谢谢大锅请我喝杯阔乐~