> 原文博客：Doi技術團隊
鏈接地址：https://blog.doiduoyi.com/authors/1584446358138
初心：記錄優秀的Doi技術團隊學習經歷
*本篇文章基於 PaddlePaddle 0.10.0、Python 2.7
前言¶
在上一篇文章中介紹了驗證碼的識別，但是使用的傳統的驗證碼分割，然後通過圖像分類的方法來實現驗證碼的識別的，這中方法比較繁瑣，工作量比較多。在本篇文章會介紹驗證碼端到端的識別，直接一步到位，不用圖像分割那麼麻煩了。好吧，現在開始吧！
數據集介紹¶
在本篇文章中同樣是使用方正系統的驗證碼，該數據集在上一篇文章《我的PaddlePaddle學習之路》筆記五——驗證碼的識別已有介紹，在這裏我就不介紹了，需要了解的可以點擊鏈接去到上一篇文章查看。
獲取驗證碼¶
下載驗證碼和修改驗證碼同樣在上一篇文章有介紹，如果讀者需要同樣可以回到上一篇文章查看。
驗證碼我們有了，有看過上一篇文章的讀者會第一反應說還缺圖像列表。沒錯，訓練和測試都需要一個圖像列表
把圖像轉成灰度圖¶
在生成列表之前，我們還有對圖像做一些處理，就是把圖像灰度化。
注意：在此之前應該把圖像文件命名，文件名爲驗證碼對應的字符，並把所有的驗證碼放在data_temp
然後執行以下的程序批量處理
# coding=utf-8
import os
from PIL import Image
def Image2GRAY(path):
# 獲取臨時文件夾中的所有圖像路徑
imgs = os.listdir(path)
i = 0
for img in imgs:
# 每10個數據取一個作爲測試數據，剩下的作爲訓練數據
if i % 10 == 0:
# 使圖像灰度化並保存
im = Image.open(path + '/' + img).convert('L')
im.save('data/test_data/' + img)
else:
# 使圖像灰度化並保存
im = Image.open(path + '/' + img).convert('L')
im.save('data/train_data/' + img)
i = i + 1
if __name__ == '__main__':
# 臨時數據存放路徑
path = 'data/data_temp'
Image2GRAY(path)
生成圖像列表¶
經過上面一步，在data/train_data我們有了訓練數據集，data/test_data測試數據集。然後就在這兩個文件夾下生成對應的圖像列表。
首先我們要了解圖像列表的格式要求，我們來看看它的格式是怎樣的
10iw.png 10iw
218j.png 218j
28hi.png 28hi
3n1g.png 3n1g
47q7.png 47q7
4ju5.png 4ju5
4uqh.png 4uqh
這個圖像類別是以Tab鍵區分路徑和label的，瞭解圖像列表的格式要求之後，那麼我們就編寫一個程序來生成這樣格式的一個圖像列表。代碼如下：
# coding=utf-8
import os
class CreateDataList:
def __init__(self):
pass
def createDataList(self, data_path, isTrain):
# 判斷生成的列表是訓練圖像列表還是測試圖像列表
if isTrain:
list_name = 'trainer.list'
else:
list_name = 'test.list'
list_path = os.path.join(data_path, list_name)
# 判斷該列表是否存在，如果存在就刪除，避免在生成圖像列表時把該路徑也寫進去了
if os.path.exists(list_path):
os.remove(list_path)
# 讀取所有的圖像路徑，此時圖像列表不存在，就不用擔心寫入非圖像文件路徑了
imgs = os.listdir(data_path)
for img in imgs:
name = img.split('.')[0]
with open(list_path, 'a') as f:
# 寫入圖像路徑和label，用Tab隔開
f.write(img + '\t' + name + '\n')
if __name__ == '__main__':
createDataList = CreateDataList()
# 生成訓練圖像列表
createDataList.createDataList('data/train_data/', True)
# 生成測試圖像列表
createDataList.createDataList('data/test_data/', False)
經過上面的程序，會在data/train_data生成圖像列表trainer.list，會在data/test_data生成圖像列表test.list。到這裏，我們的數據集已經準備好了，準備開始使用數據集訓練了。
數據的讀取¶
讀取數據成list¶
數據列表是有了，但是我們使用它就要用到文件讀取，生成一個我們方便使用的的數據格式。在本例子項目中，我把圖像的路徑和label生成是一個list。讀取方式如下：
def get_file_list(image_file_list):
'''
生成用於訓練和測試數據的文件列表。
:param image_file_list: 圖像文件和列表文件的路徑
:type image_file_list: str
'''
dirname = os.path.dirname(image_file_list)
path_list = []
with open(image_file_list) as f:
for line in f:
# 使用Tab鍵分離路徑和label
line_split = line.strip().split('\t')
filename = line_split[0].strip()
path = os.path.join(dirname, filename)
label = line_split[1].strip()
if label:
path_list.append((path, label))
return path_list
有了這個程序，我們就可以輕鬆拿到訓練數據和測試數據的list了，如下：
# 獲取訓練列表
train_file_list = get_file_list(train_file_list_path)
# 獲取測試列表
test_file_list = get_file_list(test_file_list_path)
生成和讀取標籤字典¶
在這個項目中，要使用到我們之前沒有使用過的文件：標籤字典。這個標籤字典是訓練數據集中出現的字符，如：
r 81
4 77
h 75
i 74
2 72
通過每個字符的key就可以找到對應的字符了。
我們要編寫一個從訓練數據集的list中獲取所有的字符，並生成一個標籤字典
def build_label_dict(file_list, save_path):
"""
從訓練數據建立標籤字典
:param file_list: 包含標籤的訓練數據列表
:type file_list: list
:params save_path: 保存標籤字典的路徑
:type save_path: str
"""
values = defaultdict(int)
for path, label in file_list:
for c in label:
if c:
values[c] += 1
values['<unk>'] = 0
with open(save_path, "w") as f:
for v, count in sorted(
values.iteritems(), key=lambda x: x[1], reverse=True):
f.write("%s\t%d\n" % (v, count))
然後只要傳入在上一步讀取到的train_file_list和保存標籤字典的路徑就可以生成標籤字典了。
build_label_dict(train_file_list, label_dict_path)
保存字典之後，我們還要使用到這個字典。所以我們還要編寫一個程序來讀取標籤字典，代碼如下：
def load_dict(dict_path):
"""
從字典路徑加載標籤字典
:param dict_path: 標籤字典的路徑
:type dict_path: str
"""
return dict((line.strip().split("\t")[0], idx)
for idx, line in enumerate(open(dict_path, "r").readlines()))
然後通過傳入標籤字典的路徑就可以讀取標籤字典內容了，如下：
# 獲取標籤字典
char_dict = load_dict(label_dict_path)
讀取訓練和測試的數據¶
如果學習前面幾個例子的，應該會知道trainer傳入的數據是reader的，在上面獲取的訓練數據和測試數據都是list類型的，我們要把它轉成reader類型的。同下面的程序，把訓練和測試的數據根據其路徑來加載成一維向量
# coding=utf-8
import cv2
import paddle.v2 as paddle
class Reader(object):
def __init__(self, char_dict, image_shape):
'''
:param char_dict: 標籤的字典類
:type char_dict: class
:param image_shape: 圖像的固定形狀
:type image_shape: tuple
'''
self.image_shape = image_shape
self.char_dict = char_dict
def train_reader(self, file_list):
'''
訓練讀取數據
:param file_list: 用預訓練的圖像列表，包含標籤和圖像路徑
:type file_list: list
'''
def reader():
UNK_ID = self.char_dict['<unk>']
for image_path, label in file_list:
label = [self.char_dict.get(c, UNK_ID) for c in label]
yield self.load_image(image_path), label
return reader
def load_image(self, path):
'''
加載圖像並將其轉換爲1維矢量
:param path: 圖像數據的路徑
:type path: str
'''
image = paddle.image.load_image(path,is_color=False)
# 將所有圖像調整爲固定形狀
if self.image_shape:
image = cv2.resize(
image, self.image_shape, interpolation=cv2.INTER_CUBIC)
image = image.flatten() / 255.
return image
我們通過傳入標籤字典和圖像的大小(寬度,高度)獲取reader
my_reader = Reader(char_dict=char_dict, image_shape=IMAGE_SHAPE)
然後通過執行下面的方法，同時傳入訓練的list：train_file_list和測試的list：test_file_list就可以生成reader了。
# 獲取測試數據的reader
test_reader = paddle.batch(
my_reader.train_reader(test_file_list),
batch_size=BATCH_SIZE)
# 獲取訓練數據的reader
train_reader = paddle.batch(
paddle.reader.shuffle(
my_reader.train_reader(train_file_list),
buf_size=1000),
batch_size=BATCH_SIZE)
定義網絡模型¶
這次使用的網絡模型不是單純的CNN模型了，還有結合了RNN來映射字符的分佈和使用CTC來計算CTC任務的成本，具體是如何定義的呢，請往下細看。
跟之前一樣，我們同樣要定義數據的和label，更之前不一樣的是這次我們定義數據的時候指定了寬度和高度，因爲我們這個數據集只長方形的。
在定義label的時候，之前我們要傳入類別的總數，我們這次還是同樣的道理。還記得上一步獲得的標籤字典吧，標籤字典就是我們訓練集的所有出現過字符，只要獲取字符的大小就可以了。
# 獲取字典大小
dict_size = len(char_dict)
以下就是類初始化的數據和定義數據和label的操作：
class Model(object):
def __init__(self, num_classes, shape, is_infer=False):
'''
:param num_classes: 字符字典的大小
:type num_classes: int
:param shape: 輸入圖像的大小
:type shape: tuple of 2 int
:param is_infer: 是否用於預測
:type shape: bool
'''
self.num_classes = num_classes
self.shape = shape
self.is_infer = is_infer
self.image_vector_size = shape[0] * shape[1]
self.__declare_input_layers__()
self.__build_nn__()
def __declare_input_layers__(self):
'''
定義輸入層
'''
# 圖像輸入爲一個浮動向量
self.image = paddle.layer.data(
name='image',
type=paddle.data_type.dense_vector(self.image_vector_size),
# shape是(寬度,高度)
height=self.shape[1],
width=self.shape[0])
# 將標籤輸入爲ID列表
if not self.is_infer:
self.label = paddle.layer.data(
name='label',
type=paddle.data_type.integer_value_sequence(self.num_classes))
定義網絡模型，該網絡模型
首先是通過CNN獲取圖像的特徵，
然後使用這些特徵來輸出展開成一系列特徵向量，
然後使用RNN向前和向後捕獲序列信息，
然後將RNN的輸出映射到字符分佈，
最後使用扭曲CTC來計算CTC任務的成本，獲得了cost和額外層。
def __build_nn__(self):
'''
建立網絡拓撲
'''
# 通過CNN獲取圖像特徵
def conv_block(ipt, num_filter, groups, num_channels=None):
return paddle.networks.img_conv_group(
input=ipt,
num_channels=num_channels,
conv_padding=1,
conv_num_filter=[num_filter] * groups,
conv_filter_size=3,
conv_act=paddle.activation.Relu(),
conv_with_batchnorm=True,
pool_size=2,
pool_stride=2, )
# 因爲是灰度圖所以最後一個參數是1
conv1 = conv_block(self.image, 16, 2, 1)
conv2 = conv_block(conv1, 32, 2)
conv3 = conv_block(conv2, 64, 2)
conv_features = conv_block(conv3, 128, 2)
# 將CNN的輸出展開成一系列特徵向量。
sliced_feature = paddle.layer.block_expand(
input=conv_features,
num_channels=128,
stride_x=1,
stride_y=1,
block_x=1,
block_y=11)
# 使用RNN向前和向後捕獲序列信息。
gru_forward = paddle.networks.simple_gru(
input=sliced_feature, size=128, act=paddle.activation.Relu())
gru_backward = paddle.networks.simple_gru(
input=sliced_feature,
size=128,
act=paddle.activation.Relu(),
reverse=True)
# 將RNN的輸出映射到字符分佈。
self.output = paddle.layer.fc(input=[gru_forward, gru_backward],
size=self.num_classes + 1,
act=paddle.activation.Linear())
self.log_probs = paddle.layer.mixed(
input=paddle.layer.identity_projection(input=self.output),
act=paddle.activation.Softmax())
# 使用扭曲CTC來計算CTC任務的成本。
if not self.is_infer:
# 定義cost
self.cost = paddle.layer.warp_ctc(
input=self.output,
label=self.label,
size=self.num_classes + 1,
norm_by_times=True,
blank=self.num_classes)
# 定義額外層
self.eval = paddle.evaluator.ctc_error(input=self.output, label=self.label)
最後通過調用該類就可以獲取到模型了，傳入的參數是
dict_size是標籤字典的大小，在上面有介紹是用來生成label的
IMAGE_SHAPE這個是圖像的寬度和高度，格式是：(寬度,高度)
model = Model(dict_size, IMAGE_SHAPE, is_infer=False)
生成訓練器¶
首先使用PaddlePaddle要先初始化PaddlePaddle，我們使用的是GPU，使用不了CPU，原因下面一部分會說到。
```python‘
初始化PaddlePaddle¶
paddle.init(use_gpu=True, trainer_count=1)
生成訓練器在之前的例子中，我們知道要用到損失函數，訓練參數和優化方法，這次我們多了一個額外層。
損失函數和額外層可以通過上一步的模型直接獲取
```python
cost = model.cost
extra_layers = model.eval
這次的優化方法非常簡單
optimizer = paddle.optimizer.Momentum(momentum=0)
參數也可以通過上的損失函數生成
params = paddle.parameters.create(model.cost)
最後結合這四個就可以生成一個訓練器了
trainer = paddle.trainer.SGD(cost=model.cost,
parameters=params,
update_equation=optimizer,
extra_layers=model.eval)
定義訓練¶
經過上面獲得的訓練器，就可以開始訓練了
# 開始訓練
trainer.train(reader=train_reader,
feeding=feeding,
event_handler=event_handler,
num_passes=1000)
這個用到的train_reader就是在數據讀取的時候獲得的reader。
feeding是說明數據層之間的關係，定義如下：
feeding = {'image': 0, 'label': 1}
訓練事件event_handler，通過這個訓練事件我們可以在訓練的時候處理一下事情，如輸出訓練日誌用於觀察訓練的效果，方便分析模型的性能。還可以保持模型，用於之後可預測或者再訓練。定義如下：
# 訓練事件
def event_handler(event):
if isinstance(event, paddle.event.EndIteration):
if event.batch_id % 100 == 0:
print("Pass %d, batch %d, Samples %d, Cost %f, Eval %s" %
(event.pass_id, event.batch_id, event.batch_id *
BATCH_SIZE, event.cost, event.metrics))
if isinstance(event, paddle.event.EndPass):
# 這裏由於訓練和測試數據共享相同的格式
# 我們仍然使用reader.train_reader來讀取測試數據
test_reader = paddle.batch(
my_reader.train_reader(test_file_list),
batch_size=BATCH_SIZE)
result = trainer.test(reader=test_reader, feeding=feeding)
print("Test %d, Cost %f, Eval %s" % (event.pass_id, result.cost, result.metrics))
# 檢查保存model的路徑是否存在，如果不存在就創建
if not os.path.exists(model_save_dir):
os.mkdir(model_save_dir)
with gzip.open(
os.path.join(model_save_dir, "params_pass.tar.gz"), "w") as f:
trainer.save_parameter_to_tar(f)
最後的num_passes就是訓練輪數。
啓動訓練¶
由官方文檔可知，由於模型依賴的 warp CTC 只有CUDA的實現，本模型只支持 GPU 運行。所以讀者要在自己的電腦安裝paddlepaddle-gpu，如果讀者的電腦是有GPU的話。
由於筆者的電腦沒有GPU，所以不得不使用雲服務器來訓練我們的模型。筆者使用的是百度深度學習GPU集羣，這有個非常好的地方就是購買來的服務器就已經安裝了PaddlePaddle，無需我們再安裝了，這省去了很多時間。不過筆者在使用的時候，出現了找不到libwarpctc.so這個庫，所以要自己動手去安裝該庫，如果讀者沒有報該錯，請忽略以下操作：
安裝libwarpctc.so庫¶
先從GitHub上獲取源碼
git clone https://github.com/baidu-research/warp-ctc.git
cd warp-ctc
創建build目錄
mkdir build
cd build
默認是沒有安裝cmake的，所以要先安裝cmake
apt install cmake
安裝完成之後就可以cmake和編譯了，這裏的編譯筆者使用6個線程，這個會快一點
cmake ../
make -j6
編譯完成之後，就生成了一個libwarpctc.so，這個就是我們需要的庫，執行以下命令，將其複製到相應的目錄
cp libwarpctc.so /usr/lib/x86_64-linux-gnu/
最後測試一下是否正常了
./test_gpu
執行訓練main方法¶
通過上面的操作，訓練的程序就已經完成了，可以啓動訓練了
if __name__ == "__main__":
# 訓練列表的的路徑
train_file_list_path = '../data/train_data/trainer.list'
# 測試列表的路徑
test_file_list_path = '../data/test_data/test.list'
# 標籤字典的路徑
label_dict_path = '../data/label_dict.txt'
# 保存模型的路徑
model_save_dir = '../models'
train(train_file_list_path, test_file_list_path, label_dict_path, model_save_dir)
輸出的日誌大概如下：
Pass 0, batch 0, Samples 0, Cost 16.149542, Eval {}
Pass 0, batch 100, Samples 1000, Cost 15.090727, Eval {}
Test 0, Cost 15.079704, Eval {}
Pass 1, batch 0, Samples 0, Cost 14.775064, Eval {}
Pass 1, batch 100, Samples 1000, Cost 15.448521, Eval {}
Test 1, Cost 14.826180, Eval {}
開始預測¶
通過之前的訓練，我們有了訓練參數，可以使用這些參數進行預測了。
def infer(img_path, model_path, image_shape, label_dict_path):
# 獲取標籤字典
char_dict = load_dict(label_dict_path)
# 獲取反轉的標籤字典
reversed_char_dict = load_reverse_dict(label_dict_path)
# 獲取字典大小
dict_size = len(char_dict)
# 獲取reader
my_reader = Reader(char_dict=char_dict, image_shape=image_shape)
# 初始化PaddlePaddle
paddle.init(use_gpu=True, trainer_count=1)
# 加載訓練好的參數
parameters = paddle.parameters.Parameters.from_tar(gzip.open(model_path))
# 獲取網絡模型
model = Model(dict_size, image_shape, is_infer=True)
# 獲取預測器
inferer = paddle.inference.Inference(output_layer=model.log_probs, parameters=parameters)
# 加載數據
test_batch = [[my_reader.load_image(img_path)]]
# 開始預測
return start_infer(inferer, test_batch, reversed_char_dict)
上面使用的反轉的標籤字典定義如下，通過標籤字典的文件即可生成反轉的標籤字典
def load_reverse_dict(dict_path):
"""
從字典路徑加載反轉的標籤字典
:param dict_path: 標籤字典的路徑
:type dict_path: str
"""
return dict((idx, line.strip().split("\t")[0])
for idx, line in enumerate(open(dict_path, "r").readlines()))
通過傳入上面獲取是的inferer和圖像的一維向量，還有反轉的標籤字典就可以進行預測了。
def start_infer(inferer, test_batch, reversed_char_dict):
# 獲取初步預測結果
infer_results = inferer.infer(input=test_batch)
num_steps = len(infer_results) // len(test_batch)
probs_split = [
infer_results[i * num_steps:(i + 1) * num_steps]
for i in range(0, len(test_batch))]
# 最佳路徑解碼
result = ''
for i, probs in enumerate(probs_split):
result = ctc_greedy_decoder(
probs_seq=probs, vocabulary=reversed_char_dict)
return result
這個還使用到了最佳路徑解碼，使用的解碼器如下：
def ctc_greedy_decoder(probs_seq, vocabulary):
"""CTC貪婪（最佳路徑）解碼器。
由最可能的令牌組成的路徑被進一步後處理
刪除連續的重複和所有的空白。
:param probs_seq: 每個詞彙表上概率的二維列表字符。
每個元素都是浮點概率列表爲一個字符。
:type probs_seq: list
:param vocabulary: 詞彙表
:type vocabulary: list
:return: 解碼結果字符串
:rtype: baseline
"""
# 尺寸驗證
for probs in probs_seq:
if not len(probs) == len(vocabulary) + 1:
raise ValueError("probs_seq dimension mismatchedd with vocabulary")
# argmax以獲得每個時間步長的最佳指標
max_index_list = list(np.array(probs_seq).argmax(axis=1))
# 刪除連續的重複索引
index_list = [index_group[0] for index_group in groupby(max_index_list)]
# 刪除空白索引
blank_index = len(vocabulary)
index_list = [index for index in index_list if index != blank_index]
# 將索引列表轉換爲字符串
return ''.join([vocabulary[index] for index in index_list])
最後在main方法中直接運行預測程序就可以了。
if __name__ == "__main__":
# 要預測的圖像
img_path = '../data/test_data/4uqh.png'
# 模型的路徑
model_path = '../models/params_pass.tar.gz'
# 圖像的大小
image_shape = (72, 27)
# 標籤的路徑
label_dict_path = '../data/label_dict.txt'
# 獲取預測結果
result = infer(img_path, model_path, image_shape, label_dict_path)
print '預測結果：%s' % result
預測輸出
預測結果：4uqh
項目代碼¶
GitHub地址:https://github.com/yeyupiaoling/LearnPaddle
上一章：《我的PaddlePaddle學習之路》筆記五——驗證碼的識別¶
下一章：《我的PaddlePaddle學習之路》筆記七——車牌端到端的識別¶
參考資料¶
- http://paddlepaddle.org/
- http://blog.csdn.net/qq_26819733/article/details/53608308
- https://github.com/baidu-research/warp-ctc