https://github.com/MingtaoGuo/DCGAN_WGAN_WGAN-GP_LSGAN_SNGAN_RSGAN_RaSGAN_TensorFlow
from PIL import Image
import numpy as np
import scipy.io as sio
import os
import tensorflow as tf
import matplotlib.pyplot as plt
img_H = 10
img_W = 10
img_C = 1
GAN_type = "SNGAN"
batchsize = 128
epsilon = 1e-14
def deconv(inputs, shape, strides, out_num, is_sn=False):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-2]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d_transpose(inputs, spectral_norm("sn", filters), out_num, strides) + bias
else:
return tf.nn.conv2d_transpose(inputs, filters, out_num, strides) + bias
def conv(inputs, shape, strides, is_sn=False):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-1]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d(inputs, spectral_norm("sn", filters), strides, "SAME") + bias
else:
return tf.nn.conv2d(inputs, filters, strides, "SAME") + bias
def fully_connected(inputs, num_out, is_sn=False):
W = tf.get_variable("W", [inputs.shape[-1], num_out], initializer=tf.random_normal_initializer(stddev=0.02))
b = tf.get_variable("b", [num_out], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.matmul(inputs, spectral_norm("sn", W)) + b
else:
return tf.matmul(inputs, W) + b
def leaky_relu(inputs, slope=0.2):
return tf.maximum(slope*inputs, inputs)
def spectral_norm(name, w, iteration=1):
#Spectral normalization which was published on ICLR2018,please refer to "https://www.researchgate.net/publication/318572189_Spectral_Normalization_for_Generative_Adversarial_Networks"
#This function spectral_norm is forked from "https://github.com/taki0112/Spectral_Normalization-Tensorflow"
w_shape = w.shape.as_list()
w = tf.reshape(w, [-1, w_shape[-1]])
with tf.variable_scope(name, reuse=False):
u = tf.get_variable("u", [1, w_shape[-1]], initializer=tf.truncated_normal_initializer(), trainable=False)
u_hat = u
v_hat = None
def l2_norm(v, eps=1e-12):
return v / (tf.reduce_sum(v ** 2) ** 0.5 + eps)
for i in range(iteration):
v_ = tf.matmul(u_hat, tf.transpose(w))
v_hat = l2_norm(v_)
u_ = tf.matmul(v_hat, w)
u_hat = l2_norm(u_)
sigma = tf.matmul(tf.matmul(v_hat, w), tf.transpose(u_hat))
w_norm = w / sigma
with tf.control_dependencies([u.assign(u_hat)]):
w_norm = tf.reshape(w_norm, w_shape)
return w_norm
def mapping(x):
max = np.max(x)
min = np.min(x)
return (x - min) * 255.0 / (max - min + epsilon)
def instanceNorm(inputs):
mean, var = tf.nn.moments(inputs, axes=[1, 2], keep_dims=True)
scale = tf.get_variable("scale", shape=mean.shape, initializer=tf.constant_initializer([1.0]))
shift = tf.get_variable("shift", shape=mean.shape, initializer=tf.constant_initializer([0.0]))
return (inputs - mean) * scale / (tf.sqrt(var + epsilon)) + shift
class Generator:
def __init__(self, name):
self.name = name
def __call__(self, Z,reuse=False):
with tf.variable_scope(name_or_scope=self.name, reuse=reuse):
with tf.variable_scope(name_or_scope="linear"):
inputs = tf.reshape(tf.nn.relu((fully_connected(Z, 2*2*512))), [batchsize, 2, 2, 512])
print(inputs.shape)
with tf.variable_scope(name_or_scope="deconv1"):
inputs = tf.nn.relu(instanceNorm(deconv(inputs, [3, 3, 256, 512], [1, 2, 2, 1], [batchsize, 3, 3, 256])))
print(inputs.shape)
with tf.variable_scope(name_or_scope="deconv2"):
inputs = tf.nn.relu(instanceNorm(deconv(inputs, [3, 3, 128, 256], [1, 2, 2, 1], [batchsize, 5, 5, 128])))
print(inputs.shape)
with tf.variable_scope(name_or_scope="deconv3"):
inputs = tf.nn.relu(instanceNorm(deconv(inputs, [3, 3, 64, 128], [1, 2, 2, 1], [batchsize, 10, 10, 64])))
print(inputs.shape)
with tf.variable_scope(name_or_scope="deconv4"):
inputs = tf.nn.tanh(deconv(inputs, [3, 3, img_C, 64], [1, 1, 1, 1], [batchsize, img_H, img_W, img_C]))
print(inputs.shape)
return inputs
@property
def var(self):
return tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, self.name)
class Discriminator:
def __init__(self, name):
self.name = name
def __call__(self, inputs, reuse=False, is_sn=False):
with tf.variable_scope(name_or_scope=self.name, reuse=reuse):
with tf.variable_scope("conv1"):
inputs = leaky_relu(conv(inputs, [3, 3, img_C, 128], [1, 2, 2, 1], is_sn))
print(inputs.shape)
with tf.variable_scope("conv2"):
inputs = leaky_relu(instanceNorm(conv(inputs, [3, 3, 128, 256], [1, 2, 2, 1], is_sn)))
print(inputs.shape)
with tf.variable_scope("conv3"):
inputs = leaky_relu(instanceNorm(conv(inputs, [3, 3, 256, 512], [1, 2, 2, 1], is_sn)))
print(inputs.shape)
inputs = tf.contrib.layers.flatten(inputs)
print(inputs.shape)
return fully_connected(inputs, 1, is_sn)
@property
def var(self):
return tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope=self.name)
class GAN:
#Architecture of generator and discriminator just like DCGAN.
def __init__(self):
self.Z = tf.placeholder("float", [None, 100])
self.img = tf.placeholder("float", [None, img_H, img_W, img_C])
D = Discriminator("discriminator")
G = Generator("generator")
self.fake_img = G(self.Z)
if GAN_type == "DCGAN":
#DCGAN, paper: UNSUPERVISED REPRESENTATION LEARNING WITH DEEP CONVOLUTIONAL GENERATIVE ADVERSARIAL NETWORKS
self.fake_logit = tf.nn.sigmoid(D(self.fake_img))
self.real_logit = tf.nn.sigmoid(D(self.img, reuse=True))
self.d_loss = - (tf.reduce_mean(tf.log(self.real_logit + epsilon)) + tf.reduce_mean(tf.log(1 - self.fake_logit + epsilon)))
self.g_loss = - tf.reduce_mean(tf.log(self.fake_logit + epsilon))
self.opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "WGAN":
#WGAN, paper: Wasserstein GAN
self.fake_logit = D(self.fake_img)
self.real_logit = D(self.img, reuse=True)
self.d_loss = -tf.reduce_mean(self.real_logit) + tf.reduce_mean(self.fake_logit)
self.g_loss = -tf.reduce_mean(self.fake_logit)
self.clip = []
for _, var in enumerate(D.var):
self.clip.append(tf.clip_by_value(var, -0.01, 0.01))
self.opt_D = tf.train.RMSPropOptimizer(5e-5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.RMSPropOptimizer(5e-5).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "WGAN-GP":
#WGAN-GP, paper: Improved Training of Wasserstein GANs
self.fake_logit = D(self.fake_img)
self.real_logit = D(self.img, reuse=True)
e = tf.random_uniform([batchsize, 1, 1, 1], 0, 1)
x_hat = e * self.img + (1 - e) * self.fake_img
grad = tf.gradients(D(x_hat, reuse=True), x_hat)[0]
self.d_loss = tf.reduce_mean(self.fake_logit - self.real_logit) + 10 * tf.reduce_mean(tf.square(tf.sqrt(tf.reduce_sum(tf.square(grad), axis=[1, 2, 3])) - 1))
self.g_loss = tf.reduce_mean(-self.fake_logit)
self.opt_D = tf.train.AdamOptimizer(1e-4, beta1=0., beta2=0.9).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(1e-4, beta1=0., beta2=0.9).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "LSGAN":
#LSGAN, paper: Least Squares Generative Adversarial Networks
self.fake_logit = D(self.fake_img)
self.real_logit = D(self.img, reuse=True)
self.d_loss = tf.reduce_mean(0.5 * tf.square(self.real_logit - 1) + 0.5 * tf.square(self.fake_logit))
self.g_loss = tf.reduce_mean(0.5 * tf.square(self.fake_logit - 1))
self.opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "SNGAN":
#SNGAN, paper: SPECTRAL NORMALIZATION FOR GENERATIVE ADVERSARIAL NETWORKS
self.fake_logit = tf.nn.sigmoid(D(self.fake_img, is_sn=True))
self.real_logit = tf.nn.sigmoid(D(self.img, reuse=True, is_sn=True))
self.d_loss = - (tf.reduce_mean(tf.log(self.real_logit + epsilon) + tf.log(1 - self.fake_logit + epsilon)))
self.g_loss = - tf.reduce_mean(tf.log(self.fake_logit + epsilon))
self.opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "RSGAN":
#RSGAN, paper: The relativistic discriminator: a key element missing from standard GAN
self.fake_logit = D(self.fake_img)
self.real_logit = D(self.img, reuse=True)
self.d_loss = - tf.reduce_mean(tf.log(tf.nn.sigmoid(self.real_logit - self.fake_logit) + epsilon))
self.g_loss = - tf.reduce_mean(tf.log(tf.nn.sigmoid(self.fake_logit - self.real_logit) + epsilon))
self.opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.g_loss, var_list=G.var)
elif GAN_type == "RaSGAN":
#RaSGAN, paper: The relativistic discriminator: a key element missing from standard GAN
self.fake_logit = D(self.fake_img)
self.real_logit = D(self.img, reuse=True)
self.avg_fake_logit = tf.reduce_mean(self.fake_logit)
self.avg_real_logit = tf.reduce_mean(self.real_logit)
self.D_r_tilde = tf.nn.sigmoid(self.real_logit - self.avg_fake_logit)
self.D_f_tilde = tf.nn.sigmoid(self.fake_logit - self.avg_real_logit)
self.d_loss = - tf.reduce_mean(tf.log(self.D_r_tilde + epsilon)) - tf.reduce_mean(tf.log(1 - self.D_f_tilde + epsilon))
self.g_loss = - tf.reduce_mean(tf.log(self.D_f_tilde + epsilon)) - tf.reduce_mean(tf.log(1 - self.D_r_tilde + epsilon))
self.opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.d_loss, var_list=D.var)
self.opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(self.g_loss, var_list=G.var)
self.sess = tf.Session()
self.sess.run(tf.global_variables_initializer())
self.saver = tf.train.Saver()
def save(self,data,epoch,idx):
if not os.path.exists("gen_conv"):
os.makedirs("gen_conv")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('gen_conv\%depoch_%d_batch_%d.jpg' %(epoch,idx,i))
plt.close()
def __call__(self):
epoch_nums = 500
facedata = np.load('data/data_10x10.npy')
facedata = facedata.reshape(-1,10,10,1)
print(facedata.shape)
# print(facedata[0])
#For face data, i random select about 10,000 images from CelebA and resize them to 64x64 by Matlab.
for epoch in range(epoch_nums):
for i in range(facedata.__len__()//batchsize-1):
batch = facedata[i*batchsize:i*batchsize+batchsize, :, :, :]
batch = batch * 2 - 1
# print("batch",batch[0])
z = np.random.standard_normal([batchsize, 100])
d_loss = self.sess.run(self.d_loss, feed_dict={self.img: batch, self.Z: z})
g_loss = self.sess.run(self.g_loss, feed_dict={self.img: batch, self.Z: z})
self.sess.run(self.opt_D, feed_dict={self.img: batch, self.Z: z})
if GAN_type == "WGAN":
self.sess.run(self.clip)#WGAN weight clipping
self.sess.run(self.opt_G, feed_dict={self.img: batch, self.Z: z})
if i % 20 == 0:
print("epoch: %d, step: %d, d_loss: %g, g_loss: %g"%(epoch, i, d_loss, g_loss))
if i % 3000 == 0:
z = np.random.standard_normal([batchsize, 100])
imgs = self.sess.run(self.fake_img, feed_dict={self.img: batch, self.Z: z})
imgs = imgs.reshape(-1,10,10)
imgs = (imgs+1)/2
print(imgs[0])
self.save(imgs[0:2,:,:],epoch,i)
self.saver.save(self.sess, "checkpoint_conv/model%d.ckpt" % i)
def test(self):
self.saver = tf.train.Saver()
self.saver.restore(self.sess,tf.train.latest_checkpoint("checkpoint_conv"))
z = np.random.standard_normal([batchsize, 100])
G = Generator("generator")
gen = self.sess.run(G(self.Z,reuse=True), feed_dict={self.Z: z})
gen = gen.reshape(-1,10,10)
gen = (gen + 1) / 2
print(gen.shape)
print(gen[1])
self.sess.close()
if __name__ == "__main__":
gan = GAN()
gan.test()
import tensorflow as tf
import scipy.io as sio
import numpy as np
from PIL import Image
import os
import matplotlib.pyplot as plt
data = np.load('data/data_10x10.npy')
data = data.reshape(-1,10,10,1)
epsilon = 1e-14
epochs = 200
batch_size = 128
def spectral_norm(name, w, iteration=1):
#Spectral normalization which was published on ICLR2018,please refer to "https://www.researchgate.net/publication/318572189_Spectral_Normalization_for_Generative_Adversarial_Networks"
#This function spectral_norm is forked from "https://github.com/taki0112/Spectral_Normalization-Tensorflow"
w_shape = w.shape.as_list()
w = tf.reshape(w, [-1, w_shape[-1]])
with tf.variable_scope(name, reuse=False):
u = tf.get_variable("u", [1, w_shape[-1]], initializer=tf.truncated_normal_initializer(), trainable=False)
u_hat = u
v_hat = None
def l2_norm(v, eps=1e-12):
return v / (tf.reduce_sum(v ** 2) ** 0.5 + eps)
for i in range(iteration):
v_ = tf.matmul(u_hat, tf.transpose(w))
v_hat = l2_norm(v_)
u_ = tf.matmul(v_hat, w)
u_hat = l2_norm(u_)
sigma = tf.matmul(tf.matmul(v_hat, w), tf.transpose(u_hat))
w_norm = w / sigma
with tf.control_dependencies([u.assign(u_hat)]):
w_norm = tf.reshape(w_norm, w_shape)
return w_norm
def leaky_relu(inputs, slope=0.2):
return tf.maximum(slope*inputs, inputs)
def fully_connected(inputs, out_shape, is_sn=False):
W = tf.get_variable("W", [inputs.shape[-1], out_shape], initializer=tf.random_normal_initializer(stddev=0.02))
b = tf.get_variable("b", [out_shape], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.matmul(inputs, spectral_norm("sn", W)) + b
else:
return tf.matmul(inputs, W) + b
def conv(inputs, shape, strides, is_sn=False ,padding="SAME"):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-1]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d(inputs, spectral_norm("sn", filters), strides, padding) + bias
else:
return tf.nn.conv2d(inputs, filters, strides, padding) + bias
def deconv(inputs, shape, strides, out_shape, padding="SAME", is_sn=False):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-2]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d_transpose(inputs, spectral_norm("sn", filters), out_shape, strides, padding) + bias
else:
return tf.nn.conv2d_transpose(inputs, filters, out_shape, strides, padding) + bias
def bn(inputs):
mean, var = tf.nn.moments(inputs, axes=[1, 2], keep_dims=True)
scale = tf.get_variable("scale", shape=mean.shape, initializer=tf.constant_initializer([1.0]))
shift = tf.get_variable("shift", shape=mean.shape, initializer=tf.constant_initializer([0.0]))
return (inputs - mean) * scale / (tf.sqrt(var + epsilon)) + shift
# 定义生成器
def generator(noise_img, reuse=False, name="generator"):
with tf.variable_scope(name, reuse=reuse):
# linear
with tf.variable_scope(name_or_scope="linear"):
output = fully_connected(noise_img, 2*2*512)
output = tf.nn.relu(output)
output = tf.reshape(output, [-1, 2, 2, 512])
# deconv1
# deconv(inputs, filter_shape, strides, out_shape, padding="SAME")
with tf.variable_scope(name_or_scope="deconv1"):
output = deconv(output, [3, 3, 256, 512], [1, 2, 2, 1], [batch_size, 3, 3, 256], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv2
with tf.variable_scope(name_or_scope="deconv2"):
output = deconv(output, [3, 3, 128, 256], [1, 2, 2, 1], [batch_size, 5, 5, 128], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv3
with tf.variable_scope(name_or_scope="deconv3"):
output = deconv(output, [3, 3, 64, 128], [1, 2, 2, 1], [batch_size, 10, 10, 64], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv4
with tf.variable_scope(name_or_scope="deconv4"):
output = deconv(output, [3, 3, 1, 64], [1, 1, 1, 1], [batch_size, 10, 10, 1])
output = tf.nn.tanh(output)
print(output.shape)
return output
# 定义判别器
def discriminator(img, reuse=False, is_sn = False ,name="discriminator"):
with tf.variable_scope(name, reuse=reuse):
# conv1
# conv(inputs, filter_shape, strides, padding="SAME")
with tf.variable_scope("conv1"):
output = conv(img, [3, 3, 1, 128], [1, 2, 2, 1], is_sn, padding="SAME")
output = leaky_relu(output)
# conv2
with tf.variable_scope("conv2"):
output = conv(output, [3, 3, 128, 256], [1, 2, 2, 1], is_sn, padding="SAME")
output = bn(output)
output = leaky_relu(output)
# conv3
with tf. variable_scope("conv3"):
output = conv(output, [3, 3, 256, 512], [1, 2, 2, 1], is_sn, padding="SAME")
output = bn(output)
output = leaky_relu(output)
output = tf.contrib.layers.flatten(output)
output = fully_connected(output, 1, is_sn)
print(output.shape)
return output
# 输入placeholder
def get_inputs():
real_img = tf.placeholder("float", [batch_size, 10, 10, 1], name='real_img')
noise_img = tf.placeholder("float", [batch_size, 100], name='noise_img')
return real_img, noise_img
tf.reset_default_graph()
real_img, noise_img = get_inputs()
real_data = real_img
fake_data = generator(noise_img)
fake_logit = tf.nn.sigmoid(discriminator(fake_data, is_sn=True))
real_logit = tf.nn.sigmoid(discriminator(real_data, reuse=True, is_sn=True))
d_cost = - (tf.reduce_mean(tf.log(real_logit + epsilon) + tf.log(1 - fake_logit + epsilon)))
g_cost = - tf.reduce_mean(tf.log(fake_logit + epsilon))
d_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="discriminator")
g_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="generator")
opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(d_cost, var_list=d_vars)
opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(g_cost, var_list=g_vars)
saver = tf.train.Saver()
def save(data,epoch,idx):
if not os.path.exists("sngan_gen_is_sn"):
os.makedirs("sngan_gen_is_sn")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('sngan_gen_is_sn\%depoch_%d_batch_%d.jpg' %(epoch,idx,i))
plt.close()
def save_gen(data):
if not os.path.exists("sngan_gen1"):
os.makedirs("sngan_gen1")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('sngan_gen1\%d.jpg' %(i))
plt.close()
def train():
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for epoch in range(epochs):
for i in range(data.__len__()//batch_size-1):
batch = data[i*batch_size:i*batch_size+batch_size, :, :, :]
batch_image = batch * 2 - 1
batch_noise = np.random.standard_normal([batch_size, 100])
d_loss = sess.run(d_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
g_loss = sess.run(g_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
sess.run(opt_D, feed_dict={real_img: batch_image, noise_img: batch_noise})
sess.run(opt_G, feed_dict={real_img: batch_image, noise_img: batch_noise})
if i % 20 == 0:
print("epoch: %d, step: %d, d_loss: %g, g_loss: %g"%(epoch, i, d_loss, g_loss))
if i % 3000 == 0:
sample_noise = np.random.standard_normal([batch_size, 100])
imgs = sess.run(fake_data, feed_dict={noise_img: sample_noise})
gen = imgs.reshape(-1, 10, 10)
gen = (gen + 1) / 2
print(gen[0])
save(gen[0:2,:,:],epoch,i)
saver.save(sess, "sngan_checkpoints_is_sn/sngan%d.ckpt" % epoch)
def test():
saver = tf.train.Saver()
with tf.Session() as sess:
saver.restore(sess,tf.train.latest_checkpoint("sngan_checkpoints_is_sn"))
sample_noises = np.random.standard_normal([128, 100])
gen =sess.run(fake_data, feed_dict={noise_img:sample_noises})
gen = gen.reshape(-1,10,10)
gen = (gen + 1) / 2
print(gen.shape)
print(gen[0])
save_gen(gen)
if __name__ == "__main__":
# train()
test()
import tensorflow as tf
import scipy.io as sio
import numpy as np
from PIL import Image
import os
import matplotlib.pyplot as plt
data = np.load('data/data_10x10.npy')
data = data.reshape(-1,10,10,1)
epsilon = 1e-14
epochs = 200
batch_size = 128
def spectral_norm(name, w, iteration=1):
#Spectral normalization which was published on ICLR2018,please refer to "https://www.researchgate.net/publication/318572189_Spectral_Normalization_for_Generative_Adversarial_Networks"
#This function spectral_norm is forked from "https://github.com/taki0112/Spectral_Normalization-Tensorflow"
w_shape = w.shape.as_list()
w = tf.reshape(w, [-1, w_shape[-1]])
with tf.variable_scope(name, reuse=False):
u = tf.get_variable("u", [1, w_shape[-1]], initializer=tf.truncated_normal_initializer(), trainable=False)
u_hat = u
v_hat = None
def l2_norm(v, eps=1e-12):
return v / (tf.reduce_sum(v ** 2) ** 0.5 + eps)
for i in range(iteration):
v_ = tf.matmul(u_hat, tf.transpose(w))
v_hat = l2_norm(v_)
u_ = tf.matmul(v_hat, w)
u_hat = l2_norm(u_)
sigma = tf.matmul(tf.matmul(v_hat, w), tf.transpose(u_hat))
w_norm = w / sigma
with tf.control_dependencies([u.assign(u_hat)]):
w_norm = tf.reshape(w_norm, w_shape)
return w_norm
def leaky_relu(inputs, slope=0.2):
return tf.maximum(slope*inputs, inputs)
def fully_connected(inputs, out_shape, is_sn=False):
W = tf.get_variable("W", [inputs.shape[-1], out_shape], initializer=tf.random_normal_initializer(stddev=0.02))
b = tf.get_variable("b", [out_shape], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.matmul(inputs, spectral_norm("sn", W)) + b
else:
return tf.matmul(inputs, W) + b
def conv(inputs, shape, strides, is_sn=False ,padding="SAME"):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-1]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d(inputs, spectral_norm("sn", filters), strides, padding) + bias
else:
return tf.nn.conv2d(inputs, filters, strides, padding) + bias
def deconv(inputs, shape, strides, out_shape, padding="SAME", is_sn=False):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-2]], initializer=tf.constant_initializer([0]))
if is_sn:
return tf.nn.conv2d_transpose(inputs, spectral_norm("sn", filters), out_shape, strides, padding) + bias
else:
return tf.nn.conv2d_transpose(inputs, filters, out_shape, strides, padding) + bias
def bn(inputs):
mean, var = tf.nn.moments(inputs, axes=[1, 2], keep_dims=True)
scale = tf.get_variable("scale", shape=mean.shape, initializer=tf.constant_initializer([1.0]))
shift = tf.get_variable("shift", shape=mean.shape, initializer=tf.constant_initializer([0.0]))
return (inputs - mean) * scale / (tf.sqrt(var + epsilon)) + shift
# 定义生成器
def generator(noise_img, reuse=False, name="generator"):
with tf.variable_scope(name, reuse=reuse):
# linear
with tf.variable_scope(name_or_scope="linear"):
output = fully_connected(noise_img, 2*2*512)
output = tf.nn.relu(output)
output = tf.reshape(output, [-1, 2, 2, 512])
# deconv1
# deconv(inputs, filter_shape, strides, out_shape, padding="SAME")
with tf.variable_scope(name_or_scope="deconv1"):
output = deconv(output, [3, 3, 256, 512], [1, 2, 2, 1], [batch_size, 3, 3, 256], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv2
with tf.variable_scope(name_or_scope="deconv2"):
output = deconv(output, [3, 3, 128, 256], [1, 2, 2, 1], [batch_size, 5, 5, 128], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv3
with tf.variable_scope(name_or_scope="deconv3"):
output = deconv(output, [3, 3, 64, 128], [1, 2, 2, 1], [batch_size, 10, 10, 64], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv4
with tf.variable_scope(name_or_scope="deconv4"):
output = deconv(output, [3, 3, 1, 64], [1, 1, 1, 1], [batch_size, 10, 10, 1])
output = tf.nn.tanh(output)
print(output.shape)
return output
# 定义判别器
def discriminator(img, reuse=False, is_sn = False ,name="discriminator"):
with tf.variable_scope(name, reuse=reuse):
# conv1
# conv(inputs, filter_shape, strides, padding="SAME")
with tf.variable_scope("conv1"):
output = conv(img, [3, 3, 1, 128], [1, 2, 2, 1], is_sn, padding="SAME")
output = leaky_relu(output)
# conv2
with tf.variable_scope("conv2"):
output = conv(output, [3, 3, 128, 256], [1, 2, 2, 1], is_sn, padding="SAME")
output = bn(output)
output = leaky_relu(output)
# conv3
with tf. variable_scope("conv3"):
output = conv(output, [3, 3, 256, 512], [1, 2, 2, 1], is_sn, padding="SAME")
output = bn(output)
output = leaky_relu(output)
output = tf.contrib.layers.flatten(output)
output = fully_connected(output, 1, is_sn)
print(output.shape)
return output
# 输入placeholder
def get_inputs():
real_img = tf.placeholder("float", [batch_size, 10, 10, 1], name='real_img')
noise_img = tf.placeholder("float", [batch_size, 100], name='noise_img')
return real_img, noise_img
tf.reset_default_graph()
real_img, noise_img = get_inputs()
real_data = real_img
fake_data = generator(noise_img)
fake_logit = tf.nn.sigmoid(discriminator(fake_data, is_sn=True))
real_logit = tf.nn.sigmoid(discriminator(real_data, reuse=True, is_sn=True))
d_cost = - (tf.reduce_mean(tf.log(real_logit + epsilon) + tf.log(1 - fake_logit + epsilon)))
g_cost = - tf.reduce_mean(tf.log(fake_logit + epsilon))
d_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="discriminator")
g_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="generator")
opt_D = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(d_cost, var_list=d_vars)
opt_G = tf.train.AdamOptimizer(2e-4, beta1=0.5).minimize(g_cost, var_list=g_vars)
saver = tf.train.Saver()
def save(data,epoch,idx):
if not os.path.exists("sngan_gen_is_sn"):
os.makedirs("sngan_gen_is_sn")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('sngan_gen_is_sn\%depoch_%d_batch_%d.jpg' %(epoch,idx,i))
plt.close()
def save_gen(data):
if not os.path.exists("sngan_gen1"):
os.makedirs("sngan_gen1")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('sngan_gen1\%d.jpg' %(i))
plt.close()
def train():
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for epoch in range(epochs):
for i in range(data.__len__()//batch_size-1):
batch = data[i*batch_size:i*batch_size+batch_size, :, :, :]
batch_image = batch * 2 - 1
batch_noise = np.random.standard_normal([batch_size, 100])
d_loss = sess.run(d_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
g_loss = sess.run(g_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
sess.run(opt_D, feed_dict={real_img: batch_image, noise_img: batch_noise})
sess.run(opt_G, feed_dict={real_img: batch_image, noise_img: batch_noise})
if i % 20 == 0:
print("epoch: %d, step: %d, d_loss: %g, g_loss: %g"%(epoch, i, d_loss, g_loss))
if i % 3000 == 0:
sample_noise = np.random.standard_normal([batch_size, 100])
imgs = sess.run(fake_data, feed_dict={noise_img: sample_noise})
gen = imgs.reshape(-1, 10, 10)
gen = (gen + 1) / 2
print(gen[0])
save(gen[0:2,:,:],epoch,i)
saver.save(sess, "sngan_checkpoints_is_sn/sngan%d.ckpt" % epoch)
def test():
saver = tf.train.Saver()
with tf.Session() as sess:
saver.restore(sess,tf.train.latest_checkpoint("sngan_checkpoints_is_sn"))
sample_noises = np.random.standard_normal([128, 100])
gen =sess.run(fake_data, feed_dict={noise_img:sample_noises})
gen = gen.reshape(-1,10,10)
gen = (gen + 1) / 2
print(gen.shape)
print(gen[0])
save_gen(gen)
if __name__ == "__main__":
# train()
test()
import tensorflow as tf
import scipy.io as sio
import numpy as np
from PIL import Image
import os
import matplotlib.pyplot as plt
data = np.load('data/data_10x10.npy')
data = data.reshape(-1,10,10,1)
epsilon = 1e-14
epochs = 10000000
batch_size = 100
print(data.shape)
# 定义全连接
def fully_connected(inputs, out_shape):
W = tf.get_variable("W", [inputs.shape[-1], out_shape], initializer=tf.random_normal_initializer(stddev=0.02))
b = tf.get_variable("b", [out_shape], initializer=tf.constant_initializer([0]))
return tf.matmul(inputs, W) + b
# 定义卷积
def conv(inputs, shape, strides, padding="SAME"):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-1]], initializer=tf.constant_initializer([0]))
return tf.nn.conv2d(inputs, filters, strides, padding) + bias
# 定义反卷积
def deconv(inputs, shape, strides, out_shape, padding="SAME"):
filters = tf.get_variable("kernel", shape=shape, initializer=tf.random_normal_initializer(stddev=0.02))
bias = tf.get_variable("bias", shape=[shape[-2]], initializer=tf.constant_initializer([0]))
return tf.nn.conv2d_transpose(inputs, filters, out_shape, strides, padding) + bias
# 定义Leaky_relu
def leaky_relu(inputs, slope=0.2):
return tf.maximum(slope*inputs, inputs)
# 定义bn
def bn(inputs):
mean, var = tf.nn.moments(inputs, axes=[1, 2], keep_dims=True)
scale = tf.get_variable("scale", shape=mean.shape, initializer=tf.constant_initializer([1.0]))
shift = tf.get_variable("shift", shape=mean.shape, initializer=tf.constant_initializer([0.0]))
return (inputs - mean) * scale / (tf.sqrt(var + epsilon)) + shift
# 定义生成器
def generator(noise_img, reuse=False, name="generator"):
with tf.variable_scope(name, reuse=reuse):
# linear
with tf.variable_scope(name_or_scope="linear"):
output = fully_connected(noise_img, 2*2*512)
output = tf.nn.relu(output)
output = tf.reshape(output, [-1, 2, 2, 512])
# deconv1
# deconv(inputs, filter_shape, strides, out_shape, padding="SAME")
with tf.variable_scope(name_or_scope="deconv1"):
output = deconv(output, [3, 3, 256, 512], [1, 2, 2, 1], [batch_size, 3, 3, 256], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv2
with tf.variable_scope(name_or_scope="deconv2"):
output = deconv(output, [3, 3, 128, 256], [1, 2, 2, 1], [batch_size, 5, 5, 128], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv3
with tf.variable_scope(name_or_scope="deconv3"):
output = deconv(output, [3, 3, 64, 128], [1, 2, 2, 1], [batch_size, 10, 10, 64], padding="SAME")
output = bn(output)
output = tf.nn.relu(output)
# deconv4
with tf.variable_scope(name_or_scope="deconv4"):
output = deconv(output, [3, 3, 1, 64], [1, 1, 1, 1], [batch_size, 10, 10, 1])
output = tf.nn.tanh(output)
print(output.shape)
return output
# 定义判别器
def discriminator(img, reuse=False, name="discriminator"):
with tf.variable_scope(name, reuse=reuse):
# conv1
# conv(inputs, filter_shape, strides, padding="SAME")
with tf.variable_scope("conv1"):
output = conv(img, [3, 3, 1, 128], [1, 2, 2, 1], padding="SAME")
output = leaky_relu(output)
# conv2
with tf.variable_scope("conv2"):
output = conv(output, [3, 3, 128, 256], [1, 2, 2, 1], padding="SAME")
output = bn(output)
output = leaky_relu(output)
# conv3
with tf. variable_scope("conv3"):
output = conv(output, [3, 3, 256, 512], [1, 2, 2, 1], padding="SAME")
output = bn(output)
output = leaky_relu(output)
output = tf.contrib.layers.flatten(output)
output = fully_connected(output, 1)
print(output.shape)
return output
# 输入placeholder
def get_inputs():
real_img = tf.placeholder("float", [batch_size, 10, 10, 1], name='real_img')
noise_img = tf.placeholder("float", [None, 100], name='noise_img')
return real_img, noise_img
tf.reset_default_graph()
real_img, noise_img = get_inputs()
real_data = real_img
fake_data = generator(noise_img)
fake_logit = discriminator(fake_data)
real_logit = discriminator(real_data, reuse=True)
e = tf.random_uniform([batch_size, 1, 1, 1], 0, 1)
x_hat = e * real_data + (1 - e) * fake_data
grad = tf.gradients(discriminator(x_hat,reuse=True), x_hat)[0]
d_cost = tf.reduce_mean(fake_logit - real_logit) + 10 * tf.reduce_mean(tf.square(tf.sqrt(tf.reduce_sum(tf.square(grad), axis=[1, 2, 3])) - 1))
g_cost = tf.reduce_mean(-fake_logit)
d_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="discriminator")
g_vars = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope="generator")
opt_D = tf.train.AdamOptimizer(1e-4, beta1=0., beta2=0.9).minimize(d_cost, var_list=d_vars)
opt_G = tf.train.AdamOptimizer(1e-4, beta1=0., beta2=0.9).minimize(g_cost, var_list=g_vars)
saver = tf.train.Saver()
def save(data,batch):
if not os.path.exists("wgan-gp_gen"):
os.makedirs("wgan-gp_gen")
for i in range(len(data)):
threshold = 0.95
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('wgan-gp_gen\%dbatch_%d.jpg' %(batch,i))
plt.close()
def train():
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for e in range(epochs):
# 从数据中随机挑选出batch个数据，作为一批训练
idx = np.random.randint(0, data.shape[0], batch_size)
batch = data[idx]
batch_image = batch * 2 - 1
batch_noise = np.random.standard_normal([batch_size, 100])
d_loss = sess.run(d_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
g_loss = sess.run(g_cost, feed_dict={real_img: batch_image, noise_img: batch_noise})
sess.run(opt_G, feed_dict={real_img: batch_image, noise_img: batch_noise})
for i in range(2):
sess.run(opt_D, feed_dict={real_img: batch_image, noise_img: batch_noise})
if e % 100 == 0:
print("step: %d, d_loss: %g, g_loss: %g" %(e, d_loss, g_loss))
if e % 1000 == 0:
sample_noise = np.random.standard_normal([batch_size, 100])
imgs = sess.run(fake_data, feed_dict={noise_img: sample_noise})
gen = imgs.reshape(-1, 10, 10)
gen = (gen + 1) / 2
print(gen[0])
save(gen[0:2,:,:],e)
if e % 6000 == 0:
saver.save(sess, "wgan-gp/wgan-gp%d.ckpt" % e)
if __name__ == "__main__":
train()
import os
import math
import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt
from datetime import datetime
class AvatarModel:
def __init__(self):
# 真实图片shape (height, width, depth)
self.img_shape = (10,10,1)
# 一个batch的图片向量shape (batch, height, width, depth)
self.batch_shape = (128,10,10,1)
# 一个batch包含图片数量
self.batch_size = 128
# batch数量
self.mode = 'dcgan' #dcgan,wgan-gp
# 噪音图片size
self.noise_img_size = 100
# 卷积转置输出通道数量
self.gf_size = 64
# 卷积输出通道数量
self.df_size = 64
# 训练循环次数
self.epoch_size = 100
# 学习率
self.learning_rate = 0.0002
# 优化指数衰减率
self.beta1 = 0.5
# 生成图片数量
self.sample_size = 64
self.data = np.load('data/data_10x10.npy')
self.data = self.data.reshape(-1,10,10,1)
self.chunk_size = len(self.data) // self.batch_size
@staticmethod
def conv_out_size_same(size, stride):
return int(math.ceil(float(size) / float(stride)))
@staticmethod
def linear(images, output_size, stddev=0.02, bias_start=0.0, name='Linear'):
shape = images.get_shape().as_list()
with tf.variable_scope(name):
w = tf.get_variable("w", [shape[1], output_size], tf.float32,
tf.random_normal_initializer(stddev=stddev))
b = tf.get_variable("b", [output_size],
initializer=tf.constant_initializer(bias_start))
return tf.matmul(images, w) + b, w, b
@staticmethod
def batch_normalizer(x, epsilon=1e-5, momentum=0.9, train=True, name='batch_norm'):
with tf.variable_scope(name):
return tf.contrib.layers.batch_norm(x, decay=momentum, updates_collections=None, epsilon=epsilon,
scale=True, is_training=train)
@staticmethod
def conv2d(images, output_dim, strides_shape = [1, 2, 2, 1], stddev=0.02, name="conv2d"):
with tf.variable_scope(name):
# filter : [height, width, in_channels, output_channels]
# 注意与转置卷积的不同
filter_shape = [5, 5, images.get_shape()[-1], output_dim]
w = tf.get_variable('w', filter_shape, initializer=tf.truncated_normal_initializer(stddev=stddev))
b = tf.get_variable('b', [output_dim], initializer=tf.constant_initializer(0.0))
conv = tf.nn.conv2d(images, w, strides=strides_shape, padding='SAME')
conv = tf.reshape(tf.nn.bias_add(conv, b), conv.get_shape())
return conv
@staticmethod
def deconv2d(images, output_shape, strides_shape=[1, 2, 2, 1], stddev=0.02, name='deconv2d'):
with tf.variable_scope(name):
# filter : [height, width, output_channels, in_channels]
# 注意与卷积的不同
filter_shape = [5, 5, output_shape[-1], images.get_shape()[-1]]
# strides
w = tf.get_variable('w', filter_shape, initializer=tf.random_normal_initializer(stddev=stddev))
b = tf.get_variable('biases', [output_shape[-1]], initializer=tf.constant_initializer(0.0))
deconv = tf.nn.conv2d_transpose(images, w, output_shape=output_shape, strides=strides_shape)
deconv = tf.nn.bias_add(deconv, b)
return deconv, w, b
@staticmethod
def lrelu(x, leak=0.2):
return tf.maximum(x, leak * x)
def generator(self, noise_imgs, train=True):
with tf.variable_scope('generator'):
# 分别对应每个layer的height, width
s_h, s_w, _ = self.img_shape
s_h2, s_w2 = self.conv_out_size_same(s_h, 2), self.conv_out_size_same(s_w, 2)
s_h4, s_w4 = self.conv_out_size_same(s_h2, 2), self.conv_out_size_same(s_w2, 2)
s_h8, s_w8 = self.conv_out_size_same(s_h4, 2), self.conv_out_size_same(s_w4, 2)
# s_h16, s_w16 = self.conv_out_size_same(s_h8, 2), self.conv_out_size_same(s_w8, 2)
# layer 0
# 对输入噪音图片进行线性变换
z, h0_w, h0_b = self.linear(noise_imgs, self.gf_size*8*s_h8*s_w8)
# reshape为合适的输入层格式
h0 = tf.reshape(z, [-1, s_h8, s_w8, self.gf_size * 8])
# 对数据进行归一化处理 加快收敛速度
h0 = self.batch_normalizer(h0, train=train, name='g_bn0')
# 激活函数
h0 = tf.nn.relu(h0)
print("h0:",h0)
# layer 1
# 卷积转置进行上采样
h1, h1_w, h1_b = self.deconv2d(h0, [self.batch_size, s_h4, s_w4, self.gf_size*4], name='g_h1')
h1 = self.batch_normalizer(h1, train=train, name='g_bn1')
h1 = tf.nn.relu(h1)
print("h1:",h1)
# layer 2
h2, h2_w, h2_b = self.deconv2d(h1, [self.batch_size, s_h2, s_w2, self.gf_size*2], name='g_h2')
h2 = self.batch_normalizer(h2, train=train, name='g_bn2')
h2 = tf.nn.relu(h2)
print("h2:",h2)
# layer 3
h3, h3_w, h3_b = self.deconv2d(h2, [self.batch_size, s_h, s_w, self.gf_size*1], name='g_h3')
h3 = self.batch_normalizer(h3, train=train, name='g_bn3')
h3 = tf.nn.relu(h3)
print("h3:",h3)
# layer 4
h4, h4_w, h4_b = self.deconv2d(h3, self.batch_shape,[1, 1, 1, 1], name='g_h4')
print("h4:",h4)
return tf.nn.tanh(h4)
def discriminator(self, real_imgs, reuse=False):
with tf.variable_scope("discriminator", reuse=reuse):
# layer 0
# 卷积操作
h0 = self.conv2d(real_imgs, self.df_size, name='d_h0_conv')
# 激活函数
h0 = self.lrelu(h0)
print("d_h0:",h0)
# layer 1
h1 = self.conv2d(h0, self.df_size*2, name='d_h1_conv')
h1 = self.batch_normalizer(h1, name='d_bn1')
h1 = self.lrelu(h1)
print("d_h1:",h1)
# layer 2
h2 = self.conv2d(h1, self.df_size*4, name='d_h2_conv')
h2 = self.batch_normalizer(h2, name='d_bn2')
h2 = self.lrelu(h2)
print("d_h2:",h2)
# layer 4
h3, _, _ = self.linear(tf.reshape(h2, [self.batch_size, -1]), 1, name='d_h3_lin')
print("d_h4:",h3)
return tf.nn.sigmoid(h3), h3
# @staticmethod
def loss_graph(self,real_logits, fake_logits, fake_imgs, real_imgs):
if self.mode == 'dcgan':
# 生成器图片loss
# 生成器希望判别器判断出来的标签为1
gen_loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(logits=fake_logits, labels=tf.ones_like(fake_logits)))
# 判别器识别生成器图片loss
# 判别器希望识别出来的标签为0
fake_loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(logits=fake_logits, labels=tf.zeros_like(fake_logits)))
# 判别器识别真实图片loss
# 判别器希望识别出来的标签为1
real_loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(logits=real_logits, labels=tf.ones_like(real_logits)))
# 判别器总loss
dis_loss = tf.add(fake_loss, real_loss)
return gen_loss, fake_loss, real_loss, dis_loss
elif self.mode == 'wgan-gp':
lamda = 10
gen_loss = -tf.reduce_mean(fake_logits)
dis_loss = tf.reduce_mean(fake_logits) - tf.reduce_mean(real_logits)
print(fake_imgs.shape)
print(real_imgs.shape)
#优化器
alpha = tf.random_uniform(shape=[self.batch_size,1,1,1],minval=0.,maxval=1.)
interpolates = alpha*fake_imgs + (1-alpha)*real_imgs
print(interpolates.shape)
_prob,_output = self.discriminator(interpolates,reuse=True)
gradients = tf.gradients(_output,[interpolates,])[0]
slopes = tf.sqrt(tf.reduce_sum(tf.square(gradients),axis=[1,2,3]))
gradient_penalty = tf.reduce_mean((slopes-1.)**2)
dis_loss += lamda * gradient_penalty
return gen_loss, dis_loss
@staticmethod
def optimizer_graph(gen_loss, dis_loss, learning_rate, beta1):
# 所有定义变量
train_vars = tf.trainable_variables()
# 生成器变量
gen_vars = [var for var in train_vars if var.name.startswith('generator')]
# 判别器变量
dis_vars = [var for var in train_vars if var.name.startswith('discriminator')]
# optimizer
# 生成器与判别器作为两个网络需要分别优化
gen_optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate, beta1=beta1).minimize(gen_loss, var_list=gen_vars)
dis_optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate, beta1=beta1).minimize(dis_loss, var_list=dis_vars)
return gen_optimizer, dis_optimizer
def save(self,data,epoch,idx):
if not os.path.exists('gen1'):
os.makedirs('gen1')
for i in range(len(data)):
threshold = 0.9
zhu_x = []
zhu_y = []
for j in range(len(data[i][0])):
if data[i][0][j] < threshold and data[i][3][j] < threshold:
zhu_x.append(data[i][0][j])
zhu_y.append(data[i][3][j])
zuo_x = []
zuo_y = []
for j in range(len(data[i][1])):
if data[i][1][j] < threshold and data[i][4][j] < threshold:
zuo_x.append(data[i][1][j])
zuo_y.append(data[i][4][j])
you_x = []
you_y = []
for j in range(len(data[i][2])):
if data[i][2][j] < threshold and data[i][5][j] < threshold:
you_x.append(data[i][2][j])
you_y.append(data[i][5][j])
if len(zhu_x) == len(zhu_y) and len(zuo_x) == len(zuo_y) and len(you_x) == len(you_y):
plt.plot(zhu_x,zhu_y, color='red')
plt.plot(zuo_x,zuo_y, color='green')
plt.plot(you_x,you_y, color='blue')
plt.xlim(0.,1.)
plt.ylim(0.,1.)
plt.savefig('gen1\%depoch_%d_batch_%d.jpg' %(epoch,idx,i))
plt.close()
def train(self):
# 真实图片
real_imgs = tf.placeholder(tf.float32, self.batch_shape, name='real_images')
# 噪声图片
noise_imgs = tf.placeholder(tf.float32, [None, self.noise_img_size], name='noise_images')
# 生成器图片
fake_imgs = self.generator(noise_imgs)
# 判别器
real_outputs, real_logits = self.discriminator(real_imgs)
fake_outputs, fake_logits = self.discriminator(fake_imgs, reuse=True)
# 损失
if self.mode == 'dcgan':
gen_loss, fake_loss, real_loss, dis_loss = self.loss_graph(real_logits, fake_logits,fake_imgs,real_imgs)
elif self.mode == 'wgan-gp':
gen_loss, dis_loss = self.loss_graph(real_logits, fake_logits,fake_imgs,real_imgs)
# 优化
gen_optimizer, dis_optimizer = self.optimizer_graph(gen_loss, dis_loss, self.learning_rate, self.beta1)
# 开始训练
saver = tf.train.Saver()
step = 0
# 指定占用GPU比例
# tensorflow默认占用全部GPU显存 防止在机器显存被其他程序占用过多时可能在启动时报错
gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=0.8)
with tf.Session(config=tf.ConfigProto(gpu_options=gpu_options)) as sess:
sess.run(tf.global_variables_initializer())
for epoch in range(self.epoch_size):
for idx in range(self.chunk_size):
batch_imgs = self.data[idx*self.batch_size:(idx+1)*self.batch_size]
batch_imgs = batch_imgs * 2 - 1
# generator的输入噪声
noises = np.random.uniform(-1, 1, size=(self.batch_size, self.noise_img_size)).astype(np.float32)
# 优化
_ = sess.run(dis_optimizer, feed_dict={real_imgs: batch_imgs, noise_imgs: noises})
_ = sess.run(gen_optimizer, feed_dict={noise_imgs: noises})
_ = sess.run(gen_optimizer, feed_dict={noise_imgs: noises})
step += 1
if self.mode == 'dcgan':
# 每一轮结束计算loss
# 判别器损失
loss_dis = sess.run(dis_loss, feed_dict={real_imgs: batch_imgs, noise_imgs: noises})
# 判别器对真实图片
loss_real = sess.run(real_loss, feed_dict={real_imgs: batch_imgs, noise_imgs: noises})
# 判别器对生成器图片
loss_fake = sess.run(fake_loss, feed_dict={real_imgs: batch_imgs, noise_imgs: noises})
# 生成器损失
loss_gen = sess.run(gen_loss, feed_dict={noise_imgs: noises})
print(datetime.now().strftime('%c'), ' epoch:', epoch, ' step:', step, ' loss_dis:', loss_dis,
' loss_real:', loss_real, ' loss_fake:', loss_fake, ' loss_gen:', loss_gen)
else:
loss_dis = sess.run(dis_loss, feed_dict={real_imgs: batch_imgs, noise_imgs: noises})
loss_gen = sess.run(gen_loss, feed_dict={noise_imgs: noises})
print(datetime.now().strftime('%c'), ' epoch:', epoch, ' step:', step, ' loss_dis:', loss_dis,
' loss_gen:', loss_gen)
if idx % 3000 == 0:
#保存每一轮的结果
sample_noise = np.random.uniform(-1, 1, size=(self.batch_size, self.noise_img_size))
samples = sess.run(fake_imgs, feed_dict={noise_imgs: sample_noise})
samples = (samples + 1) / 2
samples = samples.reshape(-1,10,10)
print(samples[0])
self.save(samples[0:2,:,:],epoch,idx)
saver.save(sess,'test_dcgan/dcgan%d.ckpt' % epoch )
def gen(self):
# 生成图片
noise_imgs = tf.placeholder(tf.float32, [None, self.noise_img_size], name='noise_imgs')
sample_imgs = self.generator(noise_imgs, train=False)
saver = tf.train.Saver()
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
saver.restore(sess, tf.train.latest_checkpoint('.'))
sample_noise = np.random.uniform(-1, 1, size=(self.sample_size, self.noise_img_size))
samples = sess.run(sample_imgs, feed_dict={noise_imgs: sample_noise})
for num in range(len(samples)):
if not os.path.exists('samples_dcgan'):
os.makedirs('samples_dcgan')
self.avatar.save_img(samples[num], 'samples_dcgan'+os.sep+str(num)+'.jpg')
if __name__ == '__main__':
avatar = AvatarModel()
avatar.train()
# avatar.gen()