第8章代码及详解

Author: sunjiaxin111

chapter8/GAN.py

@@ -0,0 +1,127 @@
+import keras
+from keras import layers
+import numpy as np
+import os
+from keras.preprocessing import image
+
+# GAN生成网络
+latent_dim = 32
+height = 32
+width = 32
+channels = 3
+
+generator_input = keras.Input(shape=(latent_dim,))
+
+x = layers.Dense(128 * 16 * 16)(generator_input)
+x = layers.LeakyReLU()(x)
+x = layers.Reshape((16, 16, 128))(x)
+
+x = layers.Conv2D(256, 5, padding='same')(x)
+x = layers.LeakyReLU()(x)
+
+x = layers.Conv2DTranspose(256, 4, strides=2, padding='same')(x)
+x = layers.LeakyReLU()(x)
+
+x = layers.Conv2D(256, 5, padding='same')(x)
+x = layers.LeakyReLU()(x)
+x = layers.Conv2D(256, 5, padding='same')(x)
+x = layers.LeakyReLU()(x)
+
+x = layers.Conv2D(channels, 7, activation='tanh', padding='same')(x)
+generator = keras.models.Model(generator_input, x)
+generator.summary()
+
+# GAN鉴别网络
+discriminator_input = layers.Input(shape=(height, width, channels))
+x = layers.Conv2D(128, 3)(discriminator_input)
+x = layers.LeakyReLU()(x)
+x = layers.Conv2D(128, 4, strides=2)(x)
+x = layers.LeakyReLU()(x)
+x = layers.Conv2D(128, 4, strides=2)(x)
+x = layers.LeakyReLU()(x)
+x = layers.Conv2D(128, 4, strides=2)(x)
+x = layers.LeakyReLU()(x)
+x = layers.Flatten()(x)
+
+x = layers.Dropout(0.4)(x)
+
+x = layers.Dense(1, activation='sigmoid')(x)
+
+discriminator = keras.models.Model(discriminator_input, x)
+discriminator.summary()
+
+discriminator_optimizer = keras.optimizers.RMSprop(lr=0.0008, clipvalue=1.0, decay=1e-8)
+discriminator.compile(optimizer=discriminator_optimizer, loss='binary_crossentropy')
+
+# 对抗网络
+discriminator.trainable = False
+
+gan_input = keras.Input(shape=(latent_dim,))
+gan_output = discriminator(generator(gan_input))
+gan = keras.models.Model(gan_input, gan_output)
+
+gan_optimizer = keras.optimizers.RMSprop(lr=0.0004, clipvalue=1.0, decay=1e-8)
+gan.compile(optimizer=gan_optimizer, loss='binary_crossentropy')
+
+# GAN训练过程
+(x_train, y_train), (_, _) = keras.datasets.cifar10.load_data()
+
+# 选择青蛙类别
+x_train = x_train[y_train.flatten() == 6]
+
+# 正规化数据
+x_train = x_train.reshape((x_train.shape[0],) + (height, width, channels)).astype('float32') / 255.
+
+iterations = 10000
+batch_size = 20
+save_dir = 'save_dir'
+
+# 训练循环
+start = 0
+for step in range(iterations):
+    # 在latent space中随机取点
+    random_latent_vectors = np.random.normal(size=(batch_size, latent_dim))
+
+    # 解码成合成图像
+    generated_images = generator.predict(random_latent_vectors)
+
+    # 把它们与真实图像混合
+    stop = start + batch_size
+    real_images = x_train[start: stop]
+    combined_images = np.concatenate([generated_images, real_images])
+    labels = np.concatenate([np.ones((batch_size, 1)), np.zeros((batch_size, 1))])
+
+    # 在标签中加噪声
+    labels += 0.05 * np.random.random(labels.shape)
+
+    # 训练鉴别器
+    d_loss = discriminator.train_on_batch(combined_images, labels)
+
+    # 在latent space中随机取点
+    random_latent_vectors = np.random.normal(size=(batch_size, latent_dim))
+
+    # 把标签全部设置为真
+    misleading_targets = np.zeros((batch_size, 1))
+
+    # 训练生成器,鉴别器的权重被frozen
+    a_loss = gan.train_on_batch(random_latent_vectors, misleading_targets)
+
+    start += batch_size
+    if start > len(x_train) - batch_size:
+        start = 0
+
+    if step % 100 == 0:
+        # 保存模型权重
+        gan.save_weights('gan.h5')
+
+        # 打印loss
+        print('discriminator loss:', d_loss)
+        print('adversarial loss:', a_loss)
+
+        # 保存一个生成的图片
+        img = image.array_to_img(generated_images[0] * 255., scale=False)
+        img.save(os.path.join(save_dir, 'generated_frog' + str(step) + '.png'))
+
+        # 保存一个真实图片，用来对比
+        img = image.array_to_img(real_images[0] * 255., scale=False)
+        img.save(os.path.join(save_dir, 'real_frog' + str(step) + '.png'))

chapter8/VAE.py

@@ -0,0 +1,117 @@
+import keras
+from keras import layers
+from keras import backend as K
+from keras.models import Model
+import numpy as np
+from keras.datasets import mnist
+from keras.utils import plot_model
+import matplotlib.pyplot as plt
+from scipy.stats import norm
+
+# VAE encoder网络
+img_shape = (28, 28, 1)
+batch_size = 16
+latent_dim = 2  # latent space的维度
+
+input_img = keras.Input(shape=img_shape)
+
+x = layers.Conv2D(32, 3, padding='same', activation='relu')(input_img)
+x = layers.Conv2D(64, 3, padding='same', activation='relu', strides=(2, 2))(x)
+x = layers.Conv2D(64, 3, padding='same', activation='relu')(x)
+x = layers.Conv2D(64, 3, padding='same', activation='relu')(x)
+shape_before_flattening = K.int_shape(x)
+
+x = layers.Flatten()(x)
+x = layers.Dense(32, activation='relu')(x)
+
+z_mean = layers.Dense(latent_dim)(x)
+z_log_var = layers.Dense(latent_dim)(x)
+
+
+# latent space取样函数
+def sampling(args):
+    z_mean, z_log_var = args
+    epsilon = K.random_normal(shape=(K.shape(z_mean)[0], latent_dim),
+                              mean=0.,
+                              stddev=1.)
+    return z_mean + K.exp(z_log_var) * epsilon
+
+
+z = layers.Lambda(sampling)([z_mean, z_log_var])
+
+# VAE decoder网络
+decoder_input = layers.Input(K.int_shape(z)[1:])
+
+# 不取样到正确的位置
+x = layers.Dense(np.prod(shape_before_flattening[1:]),
+                 activation='relu')(decoder_input)
+
+x = layers.Reshape(shape_before_flattening[1:])(x)
+x = layers.Conv2DTranspose(32, 3, padding='same',
+                           activation='relu', strides=(2, 2))(x)
+x = layers.Conv2D(1, 3, padding='same', activation='sigmoid')(x)
+
+decoder = Model(decoder_input, x)
+z_decoded = decoder(z)
+
+
+# 自定义一个计算VAE loss的层
+class CustomVariationalLayer(keras.layers.Layer):
+
+    def vae_loss(self, x, z_decoded):
+        x = K.flatten(x)
+        z_decoded = K.flatten(z_decoded)
+        xent_loss = keras.metrics.binary_crossentropy(x, z_decoded)
+        kl_loss = -5e-4 * K.mean(
+            1 + z_log_var - K.square(z_mean) - K.exp(z_log_var), axis=-1)
+        return K.mean(xent_loss + kl_loss)
+
+    def call(self, inputs):
+        x = inputs[0]
+        z_decoded = inputs[1]
+        loss = self.vae_loss(x, z_decoded)
+        self.add_loss(loss, inputs=inputs)
+        return x
+
+
+y = CustomVariationalLayer()([input_img, z_decoded])
+
+vae = Model(input_img, y)
+vae.compile(optimizer='rmsprop',
+            loss=None)
+vae.summary()
+plot_model(vae, show_shapes=True, to_file='model.png')
+
+(x_train, _), (x_test, y_test) = mnist.load_data()
+
+x_train = x_train.astype('float32') / 255.
+x_train = x_train.reshape(x_train.shape + (1,))
+x_test = x_test.astype('float32') / 255.
+x_test = x_test.reshape(x_test.shape + (1,))
+
+vae.fit(x=x_train, y=None,
+        shuffle=True,
+        epochs=10,
+        batch_size=batch_size,
+        validation_data=(x_test, None))
+
+# 从2维latent space中取样并解码成图像
+n = 15
+digit_size = 28
+figure = np.zeros((digit_size * n, digit_size * n))
+
+grid_x = norm.ppf(np.linspace(0.05, 0.95, n))
+grid_y = norm.ppf(np.linspace(0.05, 0.95, n))
+
+for i, yi in enumerate(grid_x):
+    for j, xi in enumerate(grid_y):
+        z_sample = np.array([[xi, yi]])
+        z_sample = np.tile(z_sample, batch_size).reshape(batch_size, 2)
+        x_decoded = decoder.predict(z_sample, batch_size=batch_size)
+        digit = x_decoded[0].reshape(digit_size, digit_size)
+        figure[i * digit_size: (i + 1) * digit_size,
+        j * digit_size: (j + 1) * digit_size] = digit
+
+plt.figure(figsize=(10, 10))
+plt.imshow(figure, cmap='Greys_r')
+plt.show()

chapter8/character-level-text-generation.py

@@ -0,0 +1,101 @@
+import keras
+import numpy as np
+from keras import layers
+import random
+import sys
+
+# 下载并解析数据
+path = keras.utils.get_file('nietzsche.txt',
+                            origin='https://s3.amazonaws.com/text-datasets/nietzsche.txt')
+text = open(path).read().lower()
+print('Corpus length:', len(text))
+
+# 向量化序列字符
+# 提取序列字符的最大长度
+maxlen = 60
+
+# 取样一个新的序列每step个字符
+step = 3
+
+# 用下面的list保存提取出来的序列
+sentences = []
+
+# 用下面的list保存targets（接下来的字符）
+next_chars = []
+
+# 取样
+for i in range(0, len(text) - maxlen, step):
+    sentences.append(text[i: i + maxlen])
+    next_chars.append(text[i + maxlen])
+print('Number of sequences:', len(sentences))
+
+# 语料库中独一无二字符的list
+chars = sorted(list(set(text)))
+print('Unique characters:', len(chars))
+# 用字典映射字符和它们在chars中的下标
+char_indices = dict((char, chars.index(char)) for char in chars)
+
+# 使用one-hot把字符编码成二进制数组
+print('Vectorization...')
+x = np.zeros((len(sentences), maxlen, len(chars)), dtype=np.bool)
+y = np.zeros((len(sentences), len(chars)), dtype=np.bool)
+for i, sentence in enumerate(sentences):
+    for t, char in enumerate(sentence):
+        x[i, t, char_indices[char]] = 1
+    y[i, char_indices[next_chars[i]]] = 1
+
+# 构建网络模型
+model = keras.models.Sequential()
+model.add(layers.LSTM(128, input_shape=(maxlen, len(chars))))
+model.add(layers.Dense(len(chars), activation='softmax'))
+
+# 模型编译
+optimizer = keras.optimizers.RMSprop(lr=0.01)
+model.compile(loss='categorical_crossentropy', optimizer=optimizer)
+
+
+# 取样函数
+def sample(preds, temperature=1.0):
+    preds = np.asarray(preds).astype('float64')
+    preds = np.log(preds) / temperature
+    exp_preds = np.exp(preds)
+    preds = exp_preds / np.sum(exp_preds)
+    # multinomial函数用来取样
+    # 第一个参数代表每次实验的实验次数
+    # 第二个参数代表样本的概率分布
+    # 第三个参数代表实验几次
+    probas = np.random.multinomial(1, preds, 1)
+    return np.argmax(probas)
+
+
+# 生成文本的循环
+for epoch in range(1, 60):
+    print('epoch', epoch)
+    # 让模型在训练数据上训练一个epoch
+    model.fit(x, y, batch_size=128, epochs=1)
+
+    # 随机选一个初始文本
+    start_index = random.randint(0, len(text) - maxlen - 1)
+    generated_text = text[start_index: start_index + maxlen]
+    print('---Generating with seed: "' + generated_text + '"')
+
+    for temperature in [0.2, 0.5, 1.0, 1.2]:
+        print('------ temperature:', temperature)
+        generated_text = text[start_index: start_index + maxlen]
+        sys.stdout.write(generated_text)
+
+        # 生成400个字符
+        for i in range(400):
+            sampled = np.zeros((1, maxlen, len(chars)))
+            for t, char in enumerate(generated_text):
+                sampled[0, t, char_indices[char]] = 1.
+
+            preds = model.predict(sampled, verbose=0)[0]
+            next_index = sample(preds, temperature)
+            next_char = chars[next_index]
+            generated_text += next_char
+            generated_text = generated_text[1:]
+
+            sys.stdout.write(next_char)
+            sys.stdout.flush()
+        print()