<jupyter_start><jupyter_text>GMM Generative Model<jupyter_code>class GMM(object):
    def __init__(self, weights, means, scales):
        self.num_centers = len(weights)
        self.weights = weights / np.sum(weights)
        self.means = means
        self.scales = scales 
    
    def sample(self, batch_size=1):
        centers = np.random.choice(self.num_centers, batch_size, p=self.weights)
        eps = np.random.randn(batch_size)
        return self.means[centers] + eps * self.scales[centers]
def plot_gmm(true_model, sampling_model,  title):
    gaussians = [norm(mean, scale) for mean, scale in zip(true_model.means, true_model.scales)]
    scale = sum(true_model.weights)
    fig, ax = plt.subplots(1, 1)
    X = np.linspace(-1.25, 1.25, 1000)
    y = np.zeros_like(X)
    for i, (weight, gaussian) in enumerate(zip(true_model.weights, gaussians)):
        y += weight * gaussian.pdf(X) / scale

    ax.plot(X, y, label='Exact PDF')
    try:
        ax.hist(sampling_model.sample(10000), bins=100, density=True, label='Samples')
    except ValueError:
        ax.hist(sampling_model.sample(10000)[0][:, 0], bins=100, density=True, label='Samples')
    ax.legend(loc='best')
    ax.set_xlim([-1.25, 1.25])
    ax.set_title(title)
    IPython.display.clear_output(wait=True)
    IPython.display.display(fig)
    plt.close()<jupyter_output><empty_output><jupyter_text>GAN Architecture<jupyter_code>class Generator(nn.Module):
    """Given a random input, produce a random output."""

    def __init__(self, input_dim: int, output_dim: int, noise='uniform'):
        super(Generator, self).__init__()
        self.input_dim = input_dim
        
        self.noise = noise
        
        self.main = nn.Sequential(
            nn.Linear(input_dim, 15),
            nn.ReLU(True),
            nn.Linear(15, output_dim),
            nn.Tanh()  # Distribution is bounded between -1 and 1.
        )

    def forward(self, x):
        return self.main(x)

    def rsample(self, batch_size=1):
        """Get a differentiable sample of the generator model."""
        if self.noise == 'uniform':
            noise = torch.rand(batch_size, self.input_dim)

        elif self.noise == 'normal':
            noise = torch.randn(batch_size, self.input_dim)
        else:
            raise NotImplementedError
            
        return self(noise).squeeze(-1)

    def sample(self, batch_size=1):
        """Get a sample of the generator model."""
        return self.rsample(batch_size).detach()


class Discriminator(nn.Module):
    """Discriminate if true from fake samples."""

    def __init__(self, input_dim: int):
        super(Discriminator, self).__init__()
        self.main = nn.Sequential(
            nn.Linear(input_dim, 25),
            nn.ReLU(True),
            nn.Linear(25, 1),
            nn.Sigmoid()  # Output is bounded between 0 and 1.
        )

    def forward(self, x):
        return self.main(x).squeeze(-1)<jupyter_output><empty_output><jupyter_text>GAN Training Algorithm<jupyter_code>def train_gan(generator, discriminator, true_model, generator_optimizer, discriminator_optimizer, 
              num_iter, discriminator_loss, generator_loss, plot_freq=1000, batch_size=64):
    loss = nn.BCELoss()
    for i in range(num_iter):
        true_data = torch.tensor(true_model.sample(batch_size)).float().unsqueeze(-1)
        fake_data = generator.rsample(batch_size).unsqueeze(-1)
        # equivalently, fake_data = generator(torch.randn(batch_size, code_size)).squeeze()

        true_label = torch.full((batch_size,), 1.)
        fake_label = torch.full((batch_size,), 0.)

        ###################################################################################
        # Update G network: maximize log(D(G(z)))                                         #
        ###################################################################################
        generator_optimizer.zero_grad()
        loss_g = loss(discriminator(fake_data), true_label)  # true label.
        loss_g.backward()
        generator_optimizer.step()

        generator_loss.append(loss_g.item())

        ###################################################################################
        # Update D network: maximize log(D(x)) + log(1 - D(G(z)))                         #
        ###################################################################################
        discriminator_optimizer.zero_grad()

        # train on true data.
        loss_d_true = loss(discriminator(true_data), true_label)
        loss_d_true.backward()

        # train on fake data.
        loss_d_fake = loss(discriminator(fake_data.detach()), fake_label)
        loss_d_fake.backward()

        discriminator_optimizer.step()

        loss_d = loss_d_true + loss_d_fake
        discriminator_loss.append(loss_d.item())

        if plot_freq and i % plot_freq == 0:
            ax = plot_gmm(true_model, generator, f"Episode {i}")
    
    return discriminator_loss, generator_loss


def train_gan_interactive(num_iter, true_model, noise_model, noise_dim, generator_lr, discriminator_lr):
    torch.manual_seed(0)
    np.random.seed(0)
    generator = Generator(input_dim=noise_dim, output_dim=1, noise=noise_model)
    discriminator = Discriminator(input_dim=1)
    
    generator_optimizer = torch.optim.Adam(generator.parameters(), lr=generator_lr, betas=(0.5, 0.999))
    discriminator_optimizer = torch.optim.Adam(discriminator.parameters(), lr=discriminator_lr, betas=(0.5, 0.99))

    discriminator_loss, generator_loss = [], []
    try:
        train_gan(generator, discriminator, true_model, generator_optimizer, discriminator_optimizer, num_iter, discriminator_loss, generator_loss)
    except KeyboardInterrupt:
        pass

    plot_gmm(true_model, generator, "Final Generator Model")
    plt.plot(generator_loss, label='Generator Loss')
    plt.plot(discriminator_loss, label='Discriminator Loss')
    plt.xlabel('Iteration Number')
    plt.ylabel(' Loss')
    plt.legend(loc='best')
    plt.show()<jupyter_output><empty_output><jupyter_text>GAN's for fitting a Gaussian<jupyter_code>rcParams['figure.figsize'] = (20, 8)
rcParams['font.size'] = 16

gaussian_model = GMM(weights=np.array([1.]),means=np.array([0.5]), scales=np.array([0.2])) 
plot_gmm(gaussian_model, gaussian_model, 'Exact Model')
rcParams['figure.figsize'] = (20, 8)
rcParams['font.size'] = 16
num_iter = 15000
interact_manual(lambda noise_model, noise_dim, generator_lr, discriminator_lr: train_gan_interactive(
    num_iter, gaussian_model, noise_model, noise_dim, generator_lr, discriminator_lr),
                noise_model=ipywidgets.Dropdown(options=['uniform', 'normal'], value='normal', description='Noise model:', style={'description_width': 'initial'}, continuous_update=False),
                noise_dim=ipywidgets.IntSlider(min=1, max=10, value=4, description='Noise dimension:', style={'description_width': 'initial'}, continuous_update=False),
                generator_lr=ipywidgets.FloatLogSlider(value=1e-4, min=-6, max=0, description="Generator lr", style={'description_width': 'initial'}, continuous_update=False),
                discriminator_lr=ipywidgets.FloatLogSlider(value=1e-4, min=-6, max=0, description="Discriminator lr", style={'description_width': 'initial'}, continuous_update=False),
               );<jupyter_output><empty_output><jupyter_text>GAN's for fitting a GMM<jupyter_code>rcParams['figure.figsize'] = (20, 8)
rcParams['font.size'] = 16

gmm_model = GMM(weights=np.array([0.3, 0.5, 0.2]),
                 means=np.array([-3., 0., 2.]) / 5,
                 scales=np.array([0.5, 1.0, 0.1]) / 5)
plot_gmm(gmm_model, gmm_model, 'Exact Model')
rcParams['figure.figsize'] = (20, 8)
rcParams['font.size'] = 16
num_iter = 15000
interact_manual(lambda noise_model, noise_dim, generator_lr, discriminator_lr: train_gan_interactive(
    num_iter, gmm_model, noise_model, noise_dim, generator_lr, discriminator_lr),
                noise_model=ipywidgets.Dropdown(options=['uniform', 'normal'], value='normal', description='Noise model:', style={'description_width': 'initial'}, continuous_update=False),
                noise_dim=ipywidgets.IntSlider(min=1, max=10, value=8, description='Noise dimension:', style={'description_width': 'initial'}, continuous_update=False),
                generator_lr=ipywidgets.FloatLogSlider(value=1e-4, min=-6, max=0, description="Generator lr", style={'description_width': 'initial'}, continuous_update=False),
                discriminator_lr=ipywidgets.FloatLogSlider(value=1e-3, min=-6, max=0, description="Discriminator lr", style={'description_width': 'initial'}, continuous_update=False),
               );
# Generator lr <= 1e-5 shows mode collapse
# Generator lr >= 1e-3 shows oscillation

from sklearn.mixture import GaussianMixture
torch.manual_seed(0)
np.random.seed(0)

gmm_model = GMM(weights=np.array([0.3, 0.5, 0.2]),
                 means=np.array([-3., 0., 2.]) / 5,
                 scales=np.array([0.5, 1.0, 0.1]) / 5)

X = gmm_model.sample(1000)
def interact_gmm_fit(n_components):
    gmm = GaussianMixture(n_components=n_components).fit(X[:, np.newaxis])
    plot_gmm(gmm_model, gmm, 'Fitted GMM')
    
interact(interact_gmm_fit, n_components=ipywidgets.IntSlider(value=3, min=1, max=10, description="Num Components", style={'description_width': 'initial'}, continuous_update=False)
        );<jupyter_output><empty_output><jupyter_text>MNIST DCGANs<jupyter_code># Number of channels in the training images. For mnist images this is 3
nc = 1

# Size of feature maps in generator
ngf = 64

# Size of feature maps in discriminator
ndf = 64

# Number of GPUs available. Use 0 for CPU mode.
ngpu = 1

# Decide which device we want to run on
device = torch.device("cuda:0" if (torch.cuda.is_available() and ngpu > 0) else "cpu")

class Generator(nn.Module):
    def __init__(self, input_dim):
        super(Generator, self).__init__() 
        self.input_dim = input_dim
        self.main = nn.Sequential(
            # input is Z, going into a convolution
            nn.ConvTranspose2d(input_dim, ngf * 8, 4, 1, 0, bias=False),
            nn.BatchNorm2d(ngf * 8),
            nn.ReLU(True),
            # state size. (ngf*8) x 4 x 4
            nn.ConvTranspose2d(ngf * 8, ngf * 4, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ngf * 4),
            nn.ReLU(True),
            # state size. (ngf*4) x 8 x 8
            nn.ConvTranspose2d( ngf * 4, ngf * 2, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ngf * 2),
            nn.ReLU(True),
            # state size. (ngf*2) x 16 x 16
            nn.ConvTranspose2d( ngf * 2, ngf, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ngf),
            nn.ReLU(True),
            # state size. (ngf) x 32 x 32
            nn.ConvTranspose2d(ngf, nc, 4, 2, 1, bias=False),
            nn.Tanh()
            # state size. (nc) x 64 x 64
        )
    
    def forward(self, x): 
        return self.main(x)
    
    def rsample(self, batch_size: int = 1):
        """Get a differentiable sample of the generator model."""
        noise = torch.randn(batch_size, self.input_dim, 1, 1).to(device=device)
        return self(noise)
    
    def sample(self, batch_size: int=1):
        """Get a sample of the generator model."""
        return self.rsample(batch_size).detach()
    
    
class Discriminator(nn.Module):
    def __init__(self, input_dim):
        super(Discriminator, self).__init__()
        self.main = nn.Sequential(
            # input is (nc) x 64 x 64
            nn.Conv2d(nc, ndf, 4, 2, 1, bias=False),
            nn.LeakyReLU(0.2, inplace=True),
            # state size. (ndf) x 32 x 32
            nn.Conv2d(ndf, ndf * 2, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ndf * 2),
            nn.LeakyReLU(0.2, inplace=True),
            # state size. (ndf*2) x 16 x 16
            nn.Conv2d(ndf * 2, ndf * 4, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ndf * 4),
            nn.LeakyReLU(0.2, inplace=True),
            # state size. (ndf*4) x 8 x 8
            nn.Conv2d(ndf * 4, ndf * 8, 4, 2, 1, bias=False),
            nn.BatchNorm2d(ndf * 8),
            nn.LeakyReLU(0.2, inplace=True),
            # state size. (ndf*8) x 4 x 4
            nn.Conv2d(ndf * 8, 1, 4, 1, 0, bias=False),
            nn.Sigmoid()
        )

    def forward(self, x): 
        return self.main(x).view(-1)
    
# custom weights initialization called on netG and netD
def weights_init(m):
    classname = m.__class__.__name__
    if classname.find('Conv') != -1:
        nn.init.normal_(m.weight.data, 0.0, 0.02)
    elif classname.find('BatchNorm') != -1:
        nn.init.normal_(m.weight.data, 1.0, 0.02)
        nn.init.constant_(m.bias.data, 0)
        
def plot_mnist_gan(fake_data, true_data, epoch, n_epoch):
        IPython.display.clear_output(wait=True)
        plt.close()

        fig = plt.figure(constrained_layout=True)
        gs = fig.add_gridspec(2, 2)
        ax1 = fig.add_subplot(gs[0, 0])  # Fake Data
        ax2 = fig.add_subplot(gs[0, 1])  # Real Data
        ax3 = fig.add_subplot(gs[1, :])  # Monitor losses

        w = fake_data[:50].detach().cpu()
        grid = torchvision.utils.make_grid(w, nrow=10, padding=5)
        ax1.imshow(np.transpose(grid.numpy(), (1,2,0)), interpolation='nearest')
        if epoch == n_epoch:
            ax1.set_title(f"Fake Data Final.")
        else:
            ax1.set_title(f"Fake Data Epoch: {epoch}/{n_epoch}.")
        ax1.set_xticks([])
        ax1.set_yticks([])

        w = true_data[:50].detach().cpu()
        grid = torchvision.utils.make_grid(w, nrow=10, padding=5)
        ax2.imshow(np.transpose(grid.numpy(), (1,2,0)), interpolation='nearest')
        if epoch == n_epoch:
            ax2.set_title(f"True Data Final.")
        else:
            ax2.set_title(f"True Data Epoch: {epoch}/{n_epoch}.")

        ax2.set_xticks([])
        ax2.set_yticks([])

        ax3.plot(generator_loss, label='Generator Loss')
        ax3.plot(discriminator_loss, label='Discriminator Loss')
        ax3.set_xlabel('Iteration Number')
        ax3.set_ylabel('Loss')
        ax3.legend(loc='best')
        fig.tight_layout()

        if epoch == n_epoch:
            plt.savefig('final.png')
        else:
            plt.savefig(f'epoch{epoch}.png')

        plt.show()
# Here we do not have a generative model, but only data samples. 
rcParams['figure.figsize'] = (16, 10)
rcParams['font.size'] = 16

torch.manual_seed(0)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False

z_dim = 100
generator_lr = 0.0002
discriminator_lr = 0.0005 
beta1 = 0.5

batch_size = 128
dataset_size = 60000
t2pil = transforms.ToPILImage()

img_size = 64
transform = transforms.Compose([
        transforms.Scale(img_size),
        transforms.ToTensor(),
        transforms.Normalize((0.5,), (0.5,))
])

train_dataset = datasets.MNIST('../data', train=True, download=True, transform=transform)
train_loader = torch.utils.data.DataLoader(dataset=train_dataset, batch_size=batch_size, shuffle=True, drop_last=True)

generator = Generator(input_dim=z_dim).to(device)
generator = torch.jit.script(generator)

generator.apply(weights_init)
discriminator = Discriminator(img_size).to(device)
discriminator = torch.jit.script(discriminator)
discriminator.apply(weights_init)


generator_optimizer = torch.optim.Adam(generator.parameters(), lr=generator_lr, betas=(0.5, 0.999))
discriminator_optimizer = torch.optim.Adam(discriminator.parameters(), lr=discriminator_lr, betas=(0.5, 0.999))


loss = nn.BCELoss()

n_epoch = 20
discriminator_loss, generator_loss = [], []
try:
    for epoch in range(n_epoch):           
        for batch_idx, (x, _) in enumerate(train_loader):
            true_data = x.to(device)  # only this line changes.
            noise = torch.randn(batch_size, z_dim, 1, 1).to(device)
            fake_data = generator(noise)

            true_label = torch.full((batch_size,), 1.).to(device)
            fake_label = torch.full((batch_size,), 0.).to(device)

            ###################################################################################
            # Update G network: maximize log(D(G(z)))                                         #
            ###################################################################################
            generator_optimizer.zero_grad()
            loss_g = loss(discriminator(fake_data), true_label)  # true label.
            loss_g.backward()
            generator_optimizer.step()

            generator_loss.append(loss_g.item())

            ###################################################################################
            # Update D network: maximize log(D(x)) + log(1 - D(G(z)))                         #
            ###################################################################################
            discriminator_optimizer.zero_grad()

            # train on true data.
            loss_d_true = loss(discriminator(true_data), true_label)
            loss_d_true.backward()

            # train on fake data.
            loss_d_fake = loss(discriminator(fake_data.detach()), fake_label)
            loss_d_fake.backward()

            discriminator_optimizer.step()

            loss_d = loss_d_true + loss_d_fake
            discriminator_loss.append(loss_d.item())
            
            if batch_idx == 0:
                plot_mnist_gan(fake_data, true_data, epoch, n_epoch)

except KeyboardInterrupt:
    pass
plot_mnist_gan(fake_data, true_data, n_epoch, n_epoch)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16394751853964631, 'fraction_numerical': 0.024358243011979464, 'mean_word_length': 2.971681014952424, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '617181', 'license': 'MIT', 'n_tokens_mistral': 6015, 'n_tokens_neox': 5663, 'n_words': 1125, 'path': 'iml2020/examples/GANs.ipynb', 'repo_name': 'AriJordan/IML_project', 'size': '17530'}