Skip to content

CWGANGP

Bases: ConditionalModel, WGAN_GP

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class CWGANGP(ConditionalModel, WGAN_GP):

    __MODEL__='CWGAN_GP'

    def __init__(self, model_parameters,
                 n_generator: Optional[int]=1,
                 n_critic: Optional[int]=1,
                 gradient_penalty_weight:int=10):
        """
        Adapts the WGAN_GP synthesizer implementation to be conditional.

        Several conditional WGAN implementations can be found online, here are a few:
            https://cameronfabbri.github.io/papers/conditionalWGAN.pdf
            https://www.sciencedirect.com/science/article/abs/pii/S0020025519309715
            https://arxiv.org/pdf/2008.09202.pdf
        """
        WGAN_GP.__init__(self, model_parameters,
                         n_generator=n_generator,
                         n_critic=n_critic,
                         gradient_penalty_weight=gradient_penalty_weight)

    def define_gan(self, activation_info: Optional[NamedTuple] = None):
        """Define the trainable model components.

        Args:
            activation_info (Optional[NamedTuple]): Defaults to None
        """
        self.generator = Generator(self.batch_size). \
            build_model(input_shape=(self.noise_dim,),
                        label_shape=(self.label_dim, ),
                        dim=self.layers_dim,
                        data_dim=self.data_dim,
                        activation_info = activation_info,
                        tau = self.tau)

        self.critic = Critic(self.batch_size). \
            build_model(input_shape=(self.data_dim,),
                        label_shape=(self.label_dim,),
                        dim=self.layers_dim)

        g_optimizer = Adam(self.g_lr, beta_1=self.beta_1, beta_2=self.beta_2)
        c_optimizer = Adam(self.d_lr, beta_1=self.beta_1, beta_2=self.beta_2)
        return g_optimizer, c_optimizer

    def gradient_penalty(self, real, fake, label):
        """Compute gradient penalty.

        Args:
            real: real event.
            fake: fake event.
            label: ground truth.
        Returns:
            gradient_penalty
        """
        epsilon = random.uniform([real.shape[0], 1], 0.0, 1.0, dtype=dtypes.float32)
        x_hat = epsilon * real + (1 - epsilon) * fake
        with GradientTape() as t:
            t.watch(x_hat)
            d_hat = self.critic([x_hat, label])
        gradients = t.gradient(d_hat, x_hat)
        ddx = sqrt(reduce_sum(gradients ** 2))
        d_regularizer = reduce_mean((ddx - 1.0) ** 2)
        return d_regularizer

    @staticmethod
    def get_data_batch(data, batch_size, seed=0):
        """Produce real data batches from the passed data object.

        Args:
            train: real data.
            batch_size: batch size.
            seed (int, optional):Defaults to 0.

        Returns:
            data batch.
        """
        start_i = (batch_size * seed) % len(data)
        stop_i = start_i + batch_size
        shuffle_seed = (batch_size * seed) // len(data)
        np.random.seed(shuffle_seed)
        data_ix = np.random.choice(data.shape[0], replace=False, size=len(data))  # wasteful to shuffle every time
        return dtypes.cast(data[data_ix[start_i: stop_i]], dtype=dtypes.float32)

    def c_lossfn(self, real):
        """Compute the critic loss.

        Args:
            real: A real sample

        Returns:
            Critic loss
        """
        real, label = real
        # generating noise from a uniform distribution
        noise = random.uniform([real.shape[0], self.noise_dim], minval=0.999, maxval=1.0 , dtype=dtypes.float32)
        # run noise through generator
        fake = self.generator([noise, label])
        # discriminate x and x_gen
        logits_real = self.critic([real, label])
        logits_fake = self.critic([fake, label])

        # gradient penalty
        gp = self.gradient_penalty(real, fake, label)
        # getting the loss of the critic.
        c_loss = (reduce_mean(logits_fake)
                  - reduce_mean(logits_real)
                  + gp * self.gradient_penalty_weight)
        return c_loss

    def g_lossfn(self, real):
        """
        Forward pass on the generator and computes the loss.

        Args:
            real: Data batch we are analyzing
        Returns:
            Generator loss
        """
        real, label = real

        # generating noise from a uniform distribution
        noise = random.uniform([real.shape[0], self.noise_dim], minval=0.0, maxval=0.001 ,dtype=dtypes.float32)

        fake = self.generator([noise, label])
        logits_fake = self.critic([fake, label])
        g_loss = -reduce_mean(logits_fake)
        return g_loss

    def fit(self, data: DataFrame,
            label_cols: List[str],
            train_arguments: TrainParameters,
            num_cols: List[str],
            cat_cols: List[str]):
        """
        Train the synthesizer on a provided dataset based on a specified condition column.

        Args:
            data: A pandas DataFrame with the data to be synthesized
            label: The name of the column to be used as a label and condition for the training
            train_arguments: GAN training arguments.
            num_cols: List of columns of the data object to be handled as numerical
            cat_cols: List of columns of the data object to be handled as categorical
        """
        data, label = self._prep_fit(data, label_cols, num_cols, cat_cols)

        processed_data = self.processor.transform(data)
        self.data_dim = processed_data.shape[1]
        self.label_dim = len(label_cols)

        #Init the GAN model and optimizers
        optimizers = self.define_gan(self.processor.col_transform_info)

        # Merging labels with processed data
        processed_data = hstack([processed_data, label])

        iterations = int(abs(processed_data.shape[0] / self.batch_size) + 1)
        print(f'Number of iterations per epoch: {iterations}')

        for epoch in trange(train_arguments.epochs):
            for _ in range(iterations):
                # ---------------------
                #  Train Discriminator
                # ---------------------
                batch_x = self.get_data_batch(processed_data, self.batch_size)  # Batches are retrieved with labels
                batch_x, label = batch_x[:, :-self.label_dim], batch_x[:, -self.label_dim:]  # Separate labels from batch

                cri_loss, ge_loss = self.train_step((batch_x, label), optimizers)

            print(
                "Epoch: {} | critic_loss: {} | gen_loss: {}".format(
                    epoch, cri_loss, ge_loss
                ))

            # If at save interval => save model state and generated image samples
            if epoch % train_arguments.sample_interval == 0:
                self._run_checkpoint(train_arguments, epoch)

    def _run_checkpoint(self, train_arguments, epoch):
        "Run checkpoint and store model state and generated samples."
        if path.exists('./cache') is False:
            os.mkdir('./cache')
        model_checkpoint_base_name = './cache/' + train_arguments.cache_prefix + '_{}_model_weights_step_{}.h5'
        self.generator.save_weights(model_checkpoint_base_name.format('generator', epoch))
        self.critic.save_weights(model_checkpoint_base_name.format('critic', epoch))

__init__(model_parameters, n_generator=1, n_critic=1, gradient_penalty_weight=10)

Adapts the WGAN_GP synthesizer implementation to be conditional.

Several conditional WGAN implementations can be found online, here are a few: https://cameronfabbri.github.io/papers/conditionalWGAN.pdf https://www.sciencedirect.com/science/article/abs/pii/S0020025519309715 https://arxiv.org/pdf/2008.09202.pdf

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(self, model_parameters,
             n_generator: Optional[int]=1,
             n_critic: Optional[int]=1,
             gradient_penalty_weight:int=10):
    """
    Adapts the WGAN_GP synthesizer implementation to be conditional.

    Several conditional WGAN implementations can be found online, here are a few:
        https://cameronfabbri.github.io/papers/conditionalWGAN.pdf
        https://www.sciencedirect.com/science/article/abs/pii/S0020025519309715
        https://arxiv.org/pdf/2008.09202.pdf
    """
    WGAN_GP.__init__(self, model_parameters,
                     n_generator=n_generator,
                     n_critic=n_critic,
                     gradient_penalty_weight=gradient_penalty_weight)

c_lossfn(real)

Compute the critic loss.

Parameters:

Name Type Description Default
real

A real sample

 required

Returns:

Type Description

Critic loss
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def c_lossfn(self, real):
    """Compute the critic loss.

    Args:
        real: A real sample

    Returns:
        Critic loss
    """
    real, label = real
    # generating noise from a uniform distribution
    noise = random.uniform([real.shape[0], self.noise_dim], minval=0.999, maxval=1.0 , dtype=dtypes.float32)
    # run noise through generator
    fake = self.generator([noise, label])
    # discriminate x and x_gen
    logits_real = self.critic([real, label])
    logits_fake = self.critic([fake, label])

    # gradient penalty
    gp = self.gradient_penalty(real, fake, label)
    # getting the loss of the critic.
    c_loss = (reduce_mean(logits_fake)
              - reduce_mean(logits_real)
              + gp * self.gradient_penalty_weight)
    return c_loss

define_gan(activation_info=None)

Define the trainable model components.

Parameters:

Name Type Description Default
activation_info Optional[NamedTuple]

Defaults to None

 None
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def define_gan(self, activation_info: Optional[NamedTuple] = None):
    """Define the trainable model components.

    Args:
        activation_info (Optional[NamedTuple]): Defaults to None
    """
    self.generator = Generator(self.batch_size). \
        build_model(input_shape=(self.noise_dim,),
                    label_shape=(self.label_dim, ),
                    dim=self.layers_dim,
                    data_dim=self.data_dim,
                    activation_info = activation_info,
                    tau = self.tau)

    self.critic = Critic(self.batch_size). \
        build_model(input_shape=(self.data_dim,),
                    label_shape=(self.label_dim,),
                    dim=self.layers_dim)

    g_optimizer = Adam(self.g_lr, beta_1=self.beta_1, beta_2=self.beta_2)
    c_optimizer = Adam(self.d_lr, beta_1=self.beta_1, beta_2=self.beta_2)
    return g_optimizer, c_optimizer

fit(data, label_cols, train_arguments, num_cols, cat_cols)

Train the synthesizer on a provided dataset based on a specified condition column.

Parameters:

Name Type Description Default
data DataFrame

A pandas DataFrame with the data to be synthesized

 required
label

The name of the column to be used as a label and condition for the training

 required
train_arguments TrainParameters

GAN training arguments.

 required
num_cols List[str]

List of columns of the data object to be handled as numerical

 required
cat_cols List[str]

List of columns of the data object to be handled as categorical

 required
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def fit(self, data: DataFrame,
        label_cols: List[str],
        train_arguments: TrainParameters,
        num_cols: List[str],
        cat_cols: List[str]):
    """
    Train the synthesizer on a provided dataset based on a specified condition column.

    Args:
        data: A pandas DataFrame with the data to be synthesized
        label: The name of the column to be used as a label and condition for the training
        train_arguments: GAN training arguments.
        num_cols: List of columns of the data object to be handled as numerical
        cat_cols: List of columns of the data object to be handled as categorical
    """
    data, label = self._prep_fit(data, label_cols, num_cols, cat_cols)

    processed_data = self.processor.transform(data)
    self.data_dim = processed_data.shape[1]
    self.label_dim = len(label_cols)

    #Init the GAN model and optimizers
    optimizers = self.define_gan(self.processor.col_transform_info)

    # Merging labels with processed data
    processed_data = hstack([processed_data, label])

    iterations = int(abs(processed_data.shape[0] / self.batch_size) + 1)
    print(f'Number of iterations per epoch: {iterations}')

    for epoch in trange(train_arguments.epochs):
        for _ in range(iterations):
            # ---------------------
            #  Train Discriminator
            # ---------------------
            batch_x = self.get_data_batch(processed_data, self.batch_size)  # Batches are retrieved with labels
            batch_x, label = batch_x[:, :-self.label_dim], batch_x[:, -self.label_dim:]  # Separate labels from batch

            cri_loss, ge_loss = self.train_step((batch_x, label), optimizers)

        print(
            "Epoch: {} | critic_loss: {} | gen_loss: {}".format(
                epoch, cri_loss, ge_loss
            ))

        # If at save interval => save model state and generated image samples
        if epoch % train_arguments.sample_interval == 0:
            self._run_checkpoint(train_arguments, epoch)

g_lossfn(real)

Forward pass on the generator and computes the loss.

Parameters:

Name Type Description Default
real

Data batch we are analyzing

 required

Returns:

Type Description

Generator loss
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def g_lossfn(self, real):
    """
    Forward pass on the generator and computes the loss.

    Args:
        real: Data batch we are analyzing
    Returns:
        Generator loss
    """
    real, label = real

    # generating noise from a uniform distribution
    noise = random.uniform([real.shape[0], self.noise_dim], minval=0.0, maxval=0.001 ,dtype=dtypes.float32)

    fake = self.generator([noise, label])
    logits_fake = self.critic([fake, label])
    g_loss = -reduce_mean(logits_fake)
    return g_loss

get_data_batch(data, batch_size, seed=0) staticmethod

Produce real data batches from the passed data object.

Parameters:

Name Type Description Default
train

real data.

 required
batch_size

batch size.

 required
seed int

Defaults to 0.

 0

Returns:

Type Description

data batch.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@staticmethod
def get_data_batch(data, batch_size, seed=0):
    """Produce real data batches from the passed data object.

    Args:
        train: real data.
        batch_size: batch size.
        seed (int, optional):Defaults to 0.

    Returns:
        data batch.
    """
    start_i = (batch_size * seed) % len(data)
    stop_i = start_i + batch_size
    shuffle_seed = (batch_size * seed) // len(data)
    np.random.seed(shuffle_seed)
    data_ix = np.random.choice(data.shape[0], replace=False, size=len(data))  # wasteful to shuffle every time
    return dtypes.cast(data[data_ix[start_i: stop_i]], dtype=dtypes.float32)

gradient_penalty(real, fake, label)

Compute gradient penalty.

Parameters:

Name Type Description Default
real

real event.

 required
fake

fake event.

 required
label

ground truth.

 required

Returns:

Type Description

gradient_penalty
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/ydata_synthetic/synthesizers/regular/cwgangp/model.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def gradient_penalty(self, real, fake, label):
    """Compute gradient penalty.

    Args:
        real: real event.
        fake: fake event.
        label: ground truth.
    Returns:
        gradient_penalty
    """
    epsilon = random.uniform([real.shape[0], 1], 0.0, 1.0, dtype=dtypes.float32)
    x_hat = epsilon * real + (1 - epsilon) * fake
    with GradientTape() as t:
        t.watch(x_hat)
        d_hat = self.critic([x_hat, label])
    gradients = t.gradient(d_hat, x_hat)
    ddx = sqrt(reduce_sum(gradients ** 2))
    d_regularizer = reduce_mean((ddx - 1.0) ** 2)
    return d_regularizer