First I would like to get a better understanding of the symbolic tensor in TensorFlow. I have read plenty of stackoverflow questions and they pretty much all say the same thing. Symbolic tensors don't hold values because they are symbolic. While I understand that this is useful neural networks to predefine the network, at some point actual numbers have to come into play. How else would the neural network calculate numerical gradients? It's definitely not calculating them symbolically. Now when I thing of symbolic, I'm thinking of it in math sense. For example, you can calculate the derivative of y=x^2 symbolically (y=2x) and evaluate it at x=2, or you can calculate it numerically by picking two x-points close to 2 (say x1=2.1 and x0=1.9) and solve (y(x1)-y(x0))/(x1-x0). If I am misunderstanding the definition of symbolic in the context of these tensors, please let me know. This leads my to my issues in my code. I am trying to emulate this paper from Google: https://doi.org/10.1609/aaai.v35i1.16086 In the paper they have a GAN architecture in which they try to have the generator learn the dynamics of the ECG. Instead of the generator output getting send directly to the discriminator, the output gets integrated in time and then goes to the discriminator. I am doing something very similar, just with different data. The issue I have is that the output of the generator is a symbolic tensor, and therefore, my function to integrate it using the trapezoidal method gives the error:
NotImplementedError: Cannot convert a symbolic tf.Tensor (add_1:0) to a numpy array. This error may indicate that you're trying to pass a Tensor to a NumPy call, which is not supported.
My attempt to convert the generator output to a numpy array gets met with:
AttributeError: 'Tensor' object has no attribute 'numpy'
I know there are actual numeric values in that output, because tf.print() will print them. I just can't assign them to a numpy variable.
My complete code is listed at the end.
I have also tried:
with tf.compat.v1.Session() as sess:
for i in range(1,y.shape[1]):
y[:,i,0] = sess.run(dt*(y_prime[:,i,0] + y_prime[:,i-1,0])/2.0 + y[:,i-1,0])
Which resulted in a very large error message. I have tried with eager execution both enabled and disabled. I have read that some people fixed the symbolic tensor error by upgrading python, but I ran the code in Google Colab which is always up to date and still got the same error. So the goal is to be able to take the output of the generator (the variable 'x_fake') and numerically integrate it before passing it to the discriminator. I would appreciate any help on this as I have been stuck on this for a while.
import time
import os
import tensorflow as tf
from tensorflow.keras.models import Model, Sequential
from tensorflow.keras.layers import Input, Dense, LeakyReLU, Conv1DTranspose, Conv1D
from tensorflow.keras.callbacks import ModelCheckpoint, TensorBoard, Callback
from tensorflow.keras.optimizers import Adam
import numpy as np
import matplotlib.pyplot as plt
def define_generator(latent_dim):
inputs = Input(shape=latent_dim)
h = Conv1DTranspose(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(inputs)
h = Conv1DTranspose(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(h)
h = Conv1DTranspose(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(h)
outputs = Conv1DTranspose(1, 11, strides=1, activation='linear')(h)
model = Model(inputs=inputs, outputs=outputs, name='GENERATOR')
return model
def generator_loss(D_labels, D_pred):
bce = tf.keras.losses.BinaryCrossentropy(from_logits=False)
loss = bce(D_labels, D_pred)
return loss
def define_discriminator(data_shape):
inputs = Input(shape=data_shape)
h = Conv1D(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(inputs)
h = Conv1D(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(h)
h = Conv1D(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(h)
h = Conv1D(1, 11, strides=1, activation=LeakyReLU(alpha=0.2))(h)
outputs = Conv1D(1, 10, strides=1, activation='sigmoid')(h)
model = Model(inputs=inputs, outputs=outputs, name='DISCRIMINATOR')
return model
def discriminator_loss(D_labels, D_pred):
bce = tf.keras.losses.BinaryCrossentropy(from_logits=False)
loss = bce(D_labels, D_pred)
return loss
def trapz_integrate(y_prime):
#tf.print(y_prime) # This actually prints out the numeric values!
#y_prime = y_prime.numpy()
dt = 0.05 # time step
y = np.zeros((y_prime.shape[0], y_prime.shape[1], y_prime.shape[2])) # initialize integrated data
y[:,0,0] = np.random.normal(loc=0, scale=2) # initial condition
for i in range(1,y.shape[1]):
y[:,i,0] = dt*(y_prime[:,i,0] + y_prime[:,i-1,0])/2.0 + y[:,i-1,0]
return y
class define_GAN(Model):
def __init__(self, gen, disc, latent_dim, n_disc_train):
super().__init__()
self.generator = gen
self.discriminator = disc
self.latent_dim = latent_dim
self.n_disc_train = n_disc_train
def compile(self, gen_loss_fcn, disc_loss_fcn, gen_lr=0.001, gen_beta1=0.9, gen_beta2=0.999, disc_lr=0.001, disc_beta1=0.9, disc_beta2=0.999):
super().compile()
self.gen_optimizer = Adam(learning_rate=gen_lr, beta_1=gen_beta1, beta_2=gen_beta2)
self.disc_optimizer = Adam(learning_rate=disc_lr, beta_1=disc_beta1, beta_2=disc_beta2)
self.gen_loss_fcn = gen_loss_fcn
self.disc_loss_fcn = disc_loss_fcn
def train_step(self, x_real):
batch_size = tf.shape(x_real)[0]
for i in range(self.n_disc_train):
latent_vec = tf.random.normal(shape=(batch_size, self.latent_dim))
x_fake = self.generator(latent_vec)
x_fake = trapz_integrate(x_fake)
with tf.GradientTape() as tape:
D_real = self.discriminator(x_real, training=True)
y_real = tf.ones([batch_size, 1])
D_fake = self.discriminator(x_fake, training=True)
y_fake = tf.zeros([batch_size, 1])
pred = tf.concat([D_real, D_fake], 0)
label = tf.concat([y_real, y_fake], 0)
disc_loss = self.disc_loss_fcn(label, pred)
disc_gradient = tape.gradient(disc_loss, self.discriminator.trainable_variables)
self.disc_optimizer.apply_gradients(zip(disc_gradient, self.discriminator.trainable_variables))
latent_vec = tf.random.normal(shape=(batch_size, self.latent_dim))
with tf.GradientTape() as tape:
x_fake = self.generator(latent_vec, training=True)
x_fake = trapz_integrate(x_fake)
D_fake = self.discriminator(x_fake, training=False)
label = tf.zeros([batch_size, 1])
gen_loss = self.gen_loss_fcn(label, D_fake)
gen_gradient = tape.gradient(gen_loss, self.generator.trainable_variables)
self.gen_optimizer.apply_gradients(zip(gen_gradient, self.generator.trainable_variables))
return {"gen_loss":gen_loss, "disc_loss":disc_loss}
c1 = np.random.normal(loc=0, scale=2, size=(1000,1))
c2 = np.random.normal(loc=0, scale=2, size=(1000,1))
t = np.linspace(0.05, 2.5, 50).reshape(1,50)
y = c1*np.cos(3*t)
latent_dim = 10
batch_size = 100
n_epochs = 5
gen = define_generator((latent_dim,1))
dis = define_discriminator((50,1))
GAN = define_GAN(gen, dis, latent_dim, n_disc_train=5)
GAN.compile(generator_loss, discriminator_loss)
history = GAN.fit(y, batch_size=batch_size, epochs=n_epochs)