local adversarial lossrefiner在學習為真實圖檔模組化的時候不應該引入artifacts， 當我們訓練一個強判别器的時候，refiner會傾向于強調一些圖檔特征來fool目前的判别器，進而導緻生成了一些artifacts。那麼怎麼解決呢？我可以可以觀察到如果我們從refine的合成圖檔上挖出一塊(patch)，這一塊的統計資訊(statistics)應該是和真實圖檔的統計資訊應該是相似的。是以，我們可以不用定義一個全局的判别器(對整張圖檔判斷合成Or真實)，我們可以對圖檔上的每一塊都判别一下。這樣的話，不僅僅是限定了接收域(receptive field)，也為訓練判别器提供了更多的樣本。判别器是一個全卷積網絡，它的輸出是w*h個patches是合成圖檔的機率。是以在更新refiner的時候，我們可以把這些w*h個patches的交叉熵loss相加。
import requests
import threading
URL = "https://captcha.delorean.codes/u/rickyhan/challenge"
DIR = "challenges/"
NUM_CHALLENGES =
lock = threading.Lock()
import json, base64, os
IMG_DIR = "./orig"
fnames = ["{}/challenge-{}".format(DIR, i) for i in range(NUM_CHALLENGES)]
if not os.path.exists(IMG_DIR):
os.mkdir(IMG_DIR)
def save_imgs(fname):
with open(fname,'r') as f:
l = json.loads(f.read(), encoding="latin-1")
for image in l['images']:
byte_image = bytes(map(ord,image['jpg_base64']))
b = base64.decodebytes(byte_image)
name = image['name']
with open(IMG_DIR+"/{}.jpg".format(name), 'wb') as f:
f.write(b)
for fname in fnames:
save_imgs(fname)
assert len(os.listdir(IMG_DIR)) == * NUM_CHALLENGES
解壓之後的圖檔長這個樣子:
from PIL import Image
imgpath = IMG_DIR + "/"+ os.listdir(IMG_DIR)[]
imgpath2 = IMG_DIR + "/"+ os.listdir(IMG_DIR)[]
im = Image.open(example_image_path)
im2 = Image.open(example_image_path2)
IMG_FNAMES = [IMG_DIR + '/' + p for p in os.listdir(IMG_DIR)]
def gray(img_path):
# convert to grayscale, then binarize
#L = R * 299/1000 + G * 587/1000 + B * 114/1000
img = Image.open(img_path).convert("L") # convert to gray scale, one 8-bit byte per pixel
img = img.point(lambda x: if x > or x == else x) # value found through T&E
img = img.point(lambda x: if x < else , "1") # convert to binary image
img.save(img_path)
for img_path in IMG_FNAMES:
gray(img_path)
mask = np.ones((height, width))
for im in ims:
mask &= im
這裡是把所有圖檔相加取平均:
import numpy as np
WIDTH, HEIGHT = im.size
MASK_DIR = "avg.png"
def generateMask():
N=*NUM_CHALLENGES
arr=np.zeros((HEIGHT, WIDTH),np.float)
for fname in IMG_FNAMES:
imarr=np.array(Image.open(fname),dtype=np.float)
arr=arr+imarr/N
arr=np.array(np.round(arr),dtype=np.uint8)
out=Image.fromarray(arr,mode="L") # save as gray scale
out.save(MASK_DIR)
generateMask()
im = Image.open(MASK_DIR) # ok this can be done with binary mask: &=
im
SimGAN-Captcha代碼閱讀與複現
再修正一下
im = Image.open(MASK_DIR)
im = im.point(lambda x: if x > else x)
im = im.point(lambda x: if x< else , "1") # 1-bit bilevel, stored with the leftmost pixel in the most significant bit. 0 means black, 1 means white.
im.save(MASK_DIR)
im
SimGAN-Captcha代碼閱讀與複現
真實圖檔的生成器
我們在訓練的時候也需要把真實的圖檔丢進去，是以這裡直接用keras的
flow_from_directory
來自動生成圖檔并且把圖檔做一些預處理。
from keras import models
from keras import layers
from keras import optimizers
from keras import applications
from keras.preprocessing import image
import tensorflow as tf
# Real data generator
datagen = image.ImageDataGenerator(
preprocessing_function=applications.xception.preprocess_input
# 調用imagenet_utils的preoprocess input函數
# tf: will scale pixels between -1 and 1,sample-wise.
)
flow_from_directory_params = {'target_size': (HEIGHT, WIDTH),
'color_mode': 'grayscale',
'class_mode': None,
'batch_size': BATCH_SIZE}
real_generator = datagen.flow_from_directory(
directory=".",
**flow_from_directory_params
)
def refiner_network(input_image_tensor):
"""
:param input_image_tensor: Input tensor that corresponds to a synthetic image.
:return: Output tensor that corresponds to a refined synthetic image.
"""
def resnet_block(input_features, nb_features=, nb_kernel_rows=, nb_kernel_cols=):
"""
A ResNet block with two `nb_kernel_rows` x `nb_kernel_cols` convolutional layers,
each with `nb_features` feature maps.
See Figure 6 in https://arxiv.org/pdf/1612.07828v1.pdf.
:param input_features: Input tensor to ResNet block.
:return: Output tensor from ResNet block.
"""
y = layers.Convolution2D(nb_features, nb_kernel_rows, nb_kernel_cols, border_mode='same')(input_features)
y = layers.Activation('relu')(y)
y = layers.Convolution2D(nb_features, nb_kernel_rows, nb_kernel_cols, border_mode='same')(y)
y = layers.merge([input_features, y], mode='sum')
return layers.Activation('relu')(y)
# an input image of size w × h is convolved with 3 × 3 filters that output 64 feature maps
x = layers.Convolution2D(, , , border_mode='same', activation='relu')(input_image_tensor)
# the output is passed through 4 ResNet blocks
for _ in range():
x = resnet_block(x)
# the output of the last ResNet block is passed to a 1 × 1 convolutional layer producing 1 feature map
# corresponding to the refined synthetic image
return layers.Convolution2D(, , , border_mode='same', activation='tanh')(x)
def discriminator_network(input_image_tensor):
"""
:param input_image_tensor: Input tensor corresponding to an image, either real or refined.
:return: Output tensor that corresponds to the probability of whether an image is real or refined.
"""
x = layers.Convolution2D(, , , border_mode='same', subsample=(, ), activation='relu')(input_image_tensor)
x = layers.Convolution2D(, , , border_mode='same', subsample=(, ), activation='relu')(x)
x = layers.MaxPooling2D(pool_size=(, ), border_mode='same', strides=(, ))(x)
x = layers.Convolution2D(, , , border_mode='same', subsample=(, ), activation='relu')(x)
x = layers.Convolution2D(, , , border_mode='same', subsample=(, ), activation='relu')(x)
x = layers.Convolution2D(, , , border_mode='same', subsample=(, ), activation='relu')(x)
# here one feature map corresponds to `is_real` and the other to `is_refined`,
# and the custom loss function is then `tf.nn.sparse_softmax_cross_entropy_with_logits`
return layers.Reshape((, ))(x) # (batch_size, # of local patches, 2)
把它們合起來
refiner 加到discriminator中去。這裡有兩個loss:
self_regularization_loss
論文中是這麼寫的: The self-regularization term minimizes the image difference
between the synthetic and the refined images. 就是用來控制refine的圖檔不至于跟原來的圖檔差别太大，由于paper中沒有具體寫公式，但是大緻就是讓生成的像素值和原始圖檔的像素值之間的距離不要太大。這裡項目的原作者是用了:
def self_regularization_loss(y_true, y_pred):
delta = # FIXME: need to figure out an appropriate value for this
return tf.multiply(delta, tf.reduce_sum(tf.abs(y_pred - y_true)))
def local_adversarial_loss(y_true, y_pred):
# y_true and y_pred have shape (batch_size, # of local patches, 2), but really we just want to average over
# the local patches and batch size so we can reshape to (batch_size * # of local patches, 2)
y_true = tf.reshape(y_true, (, ))
y_pred = tf.reshape(y_pred, (, ))
loss = tf.nn.softmax_cross_entropy_with_logits(labels=y_true, logits=y_pred)
return tf.reduce_mean(loss)
合起來如下:
# Refiner
synthetic_image_tensor = layers.Input(shape=(HEIGHT, WIDTH, )) #合成的圖檔
refined_image_tensor = refiner_network(synthetic_image_tensor)
refiner_model = models.Model(input=synthetic_image_tensor, output=refined_image_tensor, name='refiner')
# Discriminator
refined_or_real_image_tensor = layers.Input(shape=(HEIGHT, WIDTH, )) #真實的圖檔
discriminator_output = discriminator_network(refined_or_real_image_tensor)
discriminator_model = models.Model(input=refined_or_real_image_tensor, output=discriminator_output,
name='discriminator')
# Combined
refiner_model_output = refiner_model(synthetic_image_tensor)
combined_output = discriminator_model(refiner_model_output)
combined_model = models.Model(input=synthetic_image_tensor, output=[refiner_model_output, combined_output],
name='combined')
def self_regularization_loss(y_true, y_pred):
delta = # FIXME: need to figure out an appropriate value for this
return tf.multiply(delta, tf.reduce_sum(tf.abs(y_pred - y_true)))
# define custom local adversarial loss (softmax for each image section) for the discriminator
# the adversarial loss function is the sum of the cross-entropy losses over the local patches
def local_adversarial_loss(y_true, y_pred):
# y_true and y_pred have shape (batch_size, # of local patches, 2), but really we just want to average over
# the local patches and batch size so we can reshape to (batch_size * # of local patches, 2)
y_true = tf.reshape(y_true, (, ))
y_pred = tf.reshape(y_pred, (, ))
loss = tf.nn.softmax_cross_entropy_with_logits(labels=y_true, logits=y_pred)
return tf.reduce_mean(loss)
# compile models
BATCH_SIZE =
sgd = optimizers.RMSprop()
refiner_model.compile(optimizer=sgd, loss=self_regularization_loss)
discriminator_model.compile(optimizer=sgd, loss=local_adversarial_loss)
discriminator_model.trainable = False
combined_model.compile(optimizer=sgd, loss=[self_regularization_loss, local_adversarial_loss])
# the target labels for the cross-entropy loss layer are 0 for every yj (real) and 1 for every xi (refined)
# discriminator_model.output_shape = num of local patches
y_real = np.array([[[, ]] * discriminator_model.output_shape[]] * BATCH_SIZE)
y_refined = np.array([[[, ]] * discriminator_model.output_shape[]] * BATCH_SIZE)
assert y_real.shape == (BATCH_SIZE, discriminator_model.output_shape[], )
from image_history_buffer import ImageHistoryBuffer
k_d = # number of discriminator updates per step
k_g = # number of generative network updates per step
nb_steps =
# TODO: what is an appropriate size for the image history buffer?
image_history_buffer = ImageHistoryBuffer((, HEIGHT, WIDTH, ), BATCH_SIZE * , BATCH_SIZE)
combined_loss = np.zeros(shape=len(combined_model.metrics_names))
disc_loss_real = np.zeros(shape=len(discriminator_model.metrics_names))
disc_loss_refined = np.zeros(shape=len(discriminator_model.metrics_names))
# see Algorithm 1 in https://arxiv.org/pdf/1612.07828v1.pdf
for i in range(nb_steps):
print('Step: {} of {}.'.format(i, nb_steps))
# train the refiner
for _ in range(k_g * ):
# sample a mini-batch of synthetic images
synthetic_image_batch = get_image_batch(synth_generator())
# update θ by taking an SGD step on mini-batch loss LR(θ)
combined_loss = np.add(combined_model.train_on_batch(synthetic_image_batch,
[synthetic_image_batch, y_real]), combined_loss) #注意combine模型的local adversarial loss是要用y_real來對抗學習，進而迫使refiner去修改圖檔來做到跟真實圖檔很像
for _ in range(k_d):
# sample a mini-batch of synthetic and real images
synthetic_image_batch = get_image_batch(synth_generator())
real_image_batch = get_image_batch(real_generator)
# refine the synthetic images w/ the current refiner
refined_image_batch = refiner_model.predict_on_batch(synthetic_image_batch)
# use a history of refined images
half_batch_from_image_history = image_history_buffer.get_from_image_history_buffer()
image_history_buffer.add_to_image_history_buffer(refined_image_batch)
if len(half_batch_from_image_history):
refined_image_batch[:batch_size // ] = half_batch_from_image_history
# update φ by taking an SGD step on mini-batch loss LD(φ)
disc_loss_real = np.add(discriminator_model.train_on_batch(real_image_batch, y_real), disc_loss_real)
disc_loss_refined = np.add(discriminator_model.train_on_batch(refined_image_batch, y_refined),
disc_loss_refined)
if not i % LOG_INTERVAL:
# log loss summary
print('Refiner model loss: {}.'.format(combined_loss / (LOG_INTERVAL * k_g * )))
print('Discriminator model loss real: {}.'.format(disc_loss_real / (LOG_INTERVAL * k_d * )))
print('Discriminator model loss refined: {}.'.format(disc_loss_refined / (LOG_INTERVAL * k_d * )))
combined_loss = np.zeros(shape=len(combined_model.metrics_names))
disc_loss_real = np.zeros(shape=len(discriminator_model.metrics_names))
disc_loss_refined = np.zeros(shape=len(discriminator_model.metrics_names))
# save model checkpoints
model_checkpoint_base_name = os.path.join(MODEL_DIR, '{}_model_step_{}.h5')
refiner_model.save(model_checkpoint_base_name.format('refiner', i))
discriminator_model.save(model_checkpoint_base_name.format('discriminator', i))
n_class = len(alphanumeric)
def mnist_generator(batch_size=):
X = np.zeros((batch_size, HEIGHT, WIDTH, ), dtype=np.uint8)
y = [np.zeros((batch_size, n_class), dtype=np.uint8) for _ in range()] # 4 chars
while True:
for i in range(batch_size):
im, random_str = gen_one()
X[i] = im
for j, ch in enumerate(random_str):
y[j][i, :] =
y[j][i, alphanumeric.find(ch)] = # one_hot形式，讓目前字母的index為1
yield refiner_model.predict(np.array(X)), y
mg = mnist_generator().next()
模組化
from keras.layers import *
input_tensor = Input((HEIGHT, WIDTH, ))
x = input_tensor
x = Conv2D(, kernel_size=(, ),
activation='relu')(x)
# 4個conv-max_polling
for _ in range():
x = Conv2D(, (, ), activation='relu')(x)
x = MaxPooling2D(pool_size=(, ))(x)
x = Dropout()(x)
x = Flatten()(x)
x = Dense(, activation='relu')(x)
x = Dropout()(x)
x = [Dense(n_class, activation='softmax', name='c%d'%(i+))(x) for i in range()] # 4個輸出
model = models.Model(inputs=input_tensor, outputs=x)
model.compile(loss='categorical_crossentropy',
optimizer='rmsprop',
metrics=['accuracy'])
from keras.callbacks import History
history = History() # history call back現在已經是每個模型在訓練的時候都會自帶的了，fit函數會傳回，主要用于記錄事件，比如loss之類的
model.fit_generator(mnist_generator(), steps_per_epoch=, epochs=, callbacks=[history])
測試模型
先看一下在合成圖檔上的預測：
def decode(y):
y = np.argmax(np.array(y), axis=)[:,]
return ''.join([alphanumeric[x] for x in y])
X, y = next(mnist_generator())
y_pred = model.predict(X)
plt.title('real: %s\npred:%s'%(decode(y), decode(y_pred)))
plt.imshow(X[, :, :, ], cmap='gray')
plt.axis('off')
SimGAN-Captcha代碼閱讀與複現
看一下對于要預測的圖檔的預測：
X = next(real_generator)
X = refiner_model.predict(X)
# 不确定作者為什麼要用refiner來predict，應該是可以省去這一步的
# 事實證明是不可以的，後面會分析
y_pred = model.predict(X)
plt.title('pred:%s'%(decode(y_pred)))
plt.imshow(X[,:,:,], cmap='gray')
plt.axis('off')
