Keras を使って学習を進めるとき、画像の枚数が少なければ、画像をすべて読み込んだ上で、学習が速く進む。しかし、画像が大量にあるとき、メモリ容量の関係で、それらの画像をすべて読み込めない場合がある。このような場合、大量の画像を少量ずつで読み込んで学習する、というサイクルを繰り返して学習を進めることができる。
Keras では、画像を少量ずつ読み込んで学習するために、画像を少量ずつ読み込んで学習データに変換するというジェネレーターを定義する必要がある。このジェネレーターを、下では BatchGenerator
の名前で定義している。このジェネレーターでは、概ね次のような処理を行なっている。
- ジェネレーターが新たに作られると
__init__
が呼び出されて、画像ファイルへのパスimage_path
とラベルlabel
がそれぞれジェネレーターのx
とy
にセットされる。続いて、学習時に入力する画像のサイズimage_shape
とバッチサイズbatch_size
もジェネレーターにセットされる。image_path
は 1 次元リストで、各要素は文字列で、画像ファイルへのパスとなっている。label
は、image_path
の画像がどのクラスに属しているのかを示している。ただし、文字列としてのラベルではなく、ラベルをすでに One-hot 表現に変換したデータである必要がある。label
とimage_path
の各要素が対応している。
- 学習開始後、このジェネレーターの
__getitem__
メソッドが繰り返し実行される。具体的に 1 バッチごとに 1 回実行され、すべての画像データ(x
)に対して一回り実行が完了したとき、1 エポックとなる。__getitem__
で、x
のどこまでがすでに学習済みかをチェックし、まだ学習していないデータから 1 バッチ分(batch_size
)読み込んで、前処理を施して、その結果を返している。- 画像を読み込んだ直後では、画像は 0 〜 255 までの整数値で表されている。このまま学習に回すと、値が大きく、学習に悪影響を及ぼすので、全体を 255 で割って、値の範囲を 0 〜 1 の小数値に標準化する。
import cv2
import keras
class BatchGenerator(keras.utils.Sequence):
def __init__(self, image_path, label, image_shape=(299, 299, 3), batch_size=64):
self.x = image_path
self.y = label
self.length = len(image_path)
self.batch_size = batch_size
self.image_shape = image_shape
self.batches_per_epoch = int((self.length - 1) / batch_size) + 1
def __getitem__(self, idx):
batch_from = self.batch_size * idx
batch_to = batch_from + self.batch_size
if batch_to > self.length:
batch_to = self.length
x_batch = []
y_batch = []
for i in range(batch_from, batch_to):
img = cv2.imread(self.x[i], cv2.IMREAD_COLOR)
x_batch.append(img)
y_batch.append(self.y[i])
x_batch = np.asarray(x_batch)
x_batch = x_batch.astype('float32') / 255.0
y_batch = np.asarray(y_batch)
return x_batch, y_batch
def __len__(self):
return self.batches_per_epoch
def on_epoch_end(self):
pass
ジェネレーターを定義してから、次に、このジェネレーターを呼び出して、Keras の fit_generator
というメソッドで学習を進めることができるようになる。
import glob
import keras
best_model_path = 'best_model.h5'
final_model_path = 'final_model.h5'
image_shape = (299, 299, 3)
batch_size = 64
# prepare train and test data sets
## x_train = ['sample1.jpg', 'sample2.jpg', ..., 'samplex.jpg']
## y_train = [[1, 0], [0, 1], ..., [1, 0]]
## x_test = ['test1.jpg', 'test2.jpg', ..., 'testx.jpg']
## y_test = [[1, 0], [1, 0], ..., [0, 1]]
train_batch_generator = BatchGenerator(x_train, y_train, image_shape, batch_size)
test_batch_generator = BatchGenerator(x_test, y_test, image_shape, batch_size)
# create a model using Keras
## model = ...
# start training
chk_point = keras.callbacks.ModelCheckpoint(filepath = best_model_path, monitor='val_loss',
verbose=1, save_best_only=True, save_weights_only=False,
mode='min', period=1)
fit_history = model.fit_generator(train_batch_generator, epochs=epochs,
steps_per_epoch=train_batch_generator.batches_per_epoch,
verbose=1,
validation_data=test_batch_generator,
validation_steps=test_batch_generator.batches_per_epoch,
shuffle=True,
callbacks=[chk_point])
model.save(final_model_path)