<jupyter_start><jupyter_text>Let's see a bounded CNN in action<jupyter_code>%matplotlib notebook
from mpl_toolkits import mplot3d
import matplotlib.pyplot as plt 
from pprint import pprint
import numpy as np
from sklearn.datasets import load_digits
from sklearn.mixture import GaussianMixture
import time
import torch
from torch import nn
from torch.nn import functional as F
import torch.optim as optim
from torch.autograd import Variable
import random
import sys

def timeit(function, description="Please provide a description | it "):
    start = time.time()
    values = function()
    end = time.time()
    print(description + " took " + str(np.round(end - start, 4)) + " seconds")
    return values

def hello_person(person):
    print("Hello " + person)
    return "Passes along the return values"

print(timeit(lambda: hello_person("John"), description="Saying hello to John"))<jupyter_output>Hello John
Saying hello to John took 0.0001 seconds
Passes along the return values<jupyter_text>The task: classify these tiny images of digits<jupyter_code>digits = load_digits()

fig = plt.figure(figsize=(8, 1))
for i in range(10):
    ax = fig.add_subplot(1, 10, i+1)
    ax.get_yaxis().set_visible(False)
    ax.get_xaxis().set_visible(False)
    ax.imshow(digits.images[i])
fig.canvas.draw()
time.sleep(0.2)
plt.close(fig)
print(len(digits.images))
print(digits.images[0].shape)
n_layers = 3
n_classes = 10
class Baseline(nn.Module):
    
    def __init__(self, n_layers=n_layers, n_classes=n_classes):
        super(Baseline, self).__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, n_layers, kernel_size=2, stride=2),
            nn.LeakyReLU(inplace=True),
            nn.Conv2d(n_layers, n_layers, kernel_size=2, stride=2),
            nn.LeakyReLU(inplace=True),
        )
        
        self.classifier = nn.Sequential(
            nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2),
            nn.LeakyReLU(inplace=True),
            nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2),
            nn.LeakyReLU(inplace=True),
            nn.Linear(n_layers * 2 * 2, n_classes),
        )
        return
    
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), n_layers * 2 * 2)
        x = self.classifier(x)
        return x<jupyter_output><empty_output><jupyter_text>Make sure our base line works well<jupyter_code>baseline = Baseline()

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(baseline.parameters(), lr=0.001)

data, labels = digits.images[:].reshape(-1, 1, 8, 8), digits.target
data, labels = Variable(torch.tensor(data).float()), Variable(torch.tensor(labels))

print_every = (4, 9, 99, 499, 999,)
loss_history = list()
for epoch in range(1000):
    running_loss = 0.0
    
    optimizer.zero_grad()
    
    outputs = baseline(data)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()
    
    running_loss = loss.item()
    loss_history.append(running_loss)
    
    if epoch in print_every:
        fig = plt.figure()
        ax = fig.add_subplot(1, 1, 1)
        ax.plot(loss_history)
        fig.subplots_adjust(bottom=0.15)
        fig.canvas.draw()
        print("Loss: " + str(loss_history[-1]))
        _, predictions = F.softmax(baseline(data), dim=1).max(1)
        accuracy = 1 - (torch.nonzero(predictions - labels).size(0)) / predictions.size(0)
        print("Accuracy: " + str(accuracy))
        time.sleep(0.2)
        plt.close(fig)<jupyter_output><empty_output><jupyter_text>Build our fitted Neural Network<jupyter_code>class DomainNet(nn.Module):
    
    def __init__(self, n_layers=n_layers, n_classes=n_classes):
        super(DomainNet, self).__init__()
        
        self.n_layers = n_layers
        self.n_classes = n_classes
        
        ##### layers #####
        self.conv1 = nn.Conv2d(1, n_layers, kernel_size=2, stride=2)
        self.conv2 = nn.Conv2d(n_layers, n_layers, kernel_size=2, stride=2)
        
        self.linear1 = nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2)
        self.linear2 = nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2)
        self.linear3 = nn.Linear(n_layers * 2 * 2, n_classes)
        #### layers #####
        
        ### Gaussian Mixture Models ###
        self.gmm_inputs = list()
        self.gmm_conv1 =  list()
        self.gmm_conv2 =  None
        
        self.gmm_linear1 = None
        self.gmm_linear2 = None
        self.gmm_linear3 = None
        ### Gaussian Mixture Models ###
        return
    
    def forward(self, x):
        outputs = F.leaky_relu(self.conv1(x))
        outputs = F.leaky_relu(self.conv2(outputs))
        
        outputs = outputs.view(outputs.shape[0], self.n_layers * 2 * 2)
        
        outputs = F.leaky_relu(self.linear1(outputs))
        outputs = F.leaky_relu(self.linear2(outputs))
        outputs = self.linear3(outputs)
        return outputs
    
    def _fit_domain_of_convolutional_layer(self, outputs, val=None, n_components=50):
        gmms = list()
        for j in range(0, outputs.shape[2] - 1, 2):
            for k in range(0, outputs.shape[3] - 1, 2):
                kernel = outputs[:, :, j:j+2, k:k+2]
                gmm = GaussianMixture(covariance_type='full',
                                    n_components=n_components).fit(
                                    kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                if val is not None:
                    kernel = val[:, :, j:j+2, k:k+2]
                probabilities = gmm.score_samples(kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                gmm.threshold = np.amin(probabilities)
                gmms.append(gmm)
        return gmms
    
    def _fit_domain_of_linear_layer(self, outputs, val=None, n_components=50):
        gmm = GaussianMixture(covariance_type='full',
                                         n_components=n_components).fit(outputs.detach().numpy())
        if val is None:
            probabilities = gmm.score_samples(outputs.detach().numpy())
        else:
            probabilities = gmm.score_samples(val.detach().numpy())
        gmm.threshold = np.amin(probabilities)
        return gmm
        
    def fit_domain(self, x, val=None, n_components=50):
        self.gmm_inputs = self._fit_domain_of_convolutional_layer(x, val=val, n_components=n_components)
        outputs = self.conv1(x)
        if val is not None:
            val = self.conv1(val)
        
        self.gmm_conv1 = self._fit_domain_of_convolutional_layer(outputs, val=val, n_components=n_components)
        outputs = self.conv2(F.leaky_relu(outputs))
        if val is not None:
            val = self.conv2(F.leaky_relu(val))
        
        outputs = outputs.view(outputs.size(0), self.n_layers * 2 * 2)
        if val is not None:
            val = val.view(val.size(0), self.n_layers * 2 * 2)
    
        self.gmm_conv2 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear1(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear1(F.leaky_relu(val))
        
        self.gmm_linear1 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear2(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear2(F.leaky_relu(val))
        
        self.gmm_linear2 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear3(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear3(F.leaky_relu(val))
        
        self.gmm_linear3 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        return
    
    def _domain_check_convolutional_layer(self, outputs, gmms):
        i = 0
        for j in range(0, outputs.shape[2] - 1, 2):
            for k in range(0, outputs.shape[3] - 1, 2):
                kernel = outputs[:, :, j:j+2, k:k+2]
                probabilities = gmms[i].score_samples(
                    kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                if probabilities[0] < gmms[i].threshold:
                    return False
                i +=1
        return True
    
    def _domain_check_linear_layer(self, outputs, gmm):
        probabilities = gmm.score_samples(outputs.detach().numpy())
        if probabilities[0] < gmm.threshold:
            return False
        return True
    
    def domain_check(self, x):
        """
        Currently only supports checking one image at a time
        """
        passing = self._domain_check_convolutional_layer(x, self.gmm_inputs)
        if not passing:
            return -1
        outputs = self.conv1(x)
        
        passing = self._domain_check_convolutional_layer(outputs, self.gmm_conv1)
        if not passing:
            return -2
        outputs = self.conv2(F.leaky_relu(outputs))
        
        outputs = outputs.view(outputs.size(0), self.n_layers * 2 * 2)
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_conv2)
        if not passing:
            return -3
        outputs = self.linear1(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear1)
        if not passing:
            return -4
        outputs = self.linear2(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear2)
        if not passing:
            return -5
        outputs = self.linear3(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear3)
        if not passing:
            return -6
        
        # It is in domain
        return 0
    
    def sample_from_linear(self, gmm, target, target_gmm, linear_transformation, sample_size=100, temp=1):
        max_iterations = 1000
        for i in range(max_iterations):
            if max_iterations - i < 5:
                print("sampling iteration: " + str(i))
            samples, _ = gmm.sample(sample_size)
            scores = gmm.score_samples(samples)
            idxs = np.where(scores >= gmm.threshold)[0]
            if idxs.shape[0] != 0:
                samples = samples[idxs]
                squashed_sample = linear_transformation(F.leaky_relu(torch.tensor(samples).float())).detach().numpy()
                scores = target_gmm.score_samples(squashed_sample)
                idxs = np.where(scores >= target_gmm.threshold)[0]
                if idxs.shape[0] != 0:
                    samples = samples[idxs]
                    squashed_sample = squashed_sample[idxs]
                    break
            if i == (max_iterations - 1):
                sys.exit("Sampling took to long we timed out")       
        negative_distances = -1 * np.linalg.norm(squashed_sample - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        new_target = samples[idx]
        return new_target
    
    def sample_from_outputs(self, gmm, target, sample_size=100, temp=1):
        max_iterations = 1000
        for i in range(max_iterations):
            if max_iterations - i < 5:
                print("sampling iteration: " + str(i))
            samples, _ = gmm.sample(sample_size)
            scores = gmm.score_samples(samples)
            idxs = np.where(scores >= gmm.threshold)[0]
            if idxs.shape[0] != 0:
                samples = samples[idxs]
                break
            if i == (max_iterations - 1):
                sys.exit("Sampling took to long we timed out")
        
        squashed_samples = F.softmax(torch.tensor(samples).float(), dim=1).detach().numpy()
        negative_distances = -1 * np.linalg.norm(squashed_samples - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        return samples[idx]
    
    def generate_one_sample_from_convolutional(self, gmms, target,
                                               conv_filter, sample_size=100, temp=1, is_inputs=False):
        max_iterations = 1000
        # create an array filled with nans that has the shape of a sample
        sample = np.zeros((1, conv_filter.weight.shape[1], target.shape[2] * 2, target.shape[3] * 2))
        sample.fill(np.nan)
        
        # create an array filled with nans that has the shape of target
        transformed_sample = np.zeros(target.shape)
        transformed_sample.fill(np.nan)
        
        # create a list of all the kernel locations and shuffle it
        kernel_idxs = list()
        i = 0
        for j in range(target.shape[2]):
            for k in range(target.shape[3]):
                kernel_idxs.append((i, j, k))
                i += 1
        random.shuffle(kernel_idxs)
        
        # sample from each kernel
        # make sure each sample is in domain. If you can't generate an in domain sample give up after a while
        # and return None
        for (i, j, k) in kernel_idxs:
            kernel_target = target[:, :, j, k].reshape(-1, self.n_layers)
            for m in range(max_iterations):
                if max_iterations - m < 5:
                    print("sampling iteration: " + str(m))
                kernel_sample, _ = gmms[i].sample(sample_size)
                scores = gmms[i].score_samples(kernel_sample)
                print
                idxs = np.where(scores >= gmms[i].threshold)[0]
                if idxs.shape[0] != 0:
                    kernel_sample = kernel_sample[idxs]
                    kernel_sample = kernel_sample.reshape(-1, conv_filter.weight.shape[1], 2, 2)
                    break
                if m == (max_iterations - 1):
                    print("Gave up on this iteration")
                    return None, None
                
            if is_inputs:
                # If its the input there shouldn't be any non-linearity
                transformed_kernel_sample = conv_filter(
                    torch.tensor(kernel_sample).float()).detach().numpy().reshape(-1, self.n_layers)
            else:
                transformed_kernel_sample = conv_filter(F.leaky_relu(
                    torch.tensor(kernel_sample).float())).detach().numpy().reshape(-1, self.n_layers)
            
            negative_distances = -1 * np.linalg.norm(transformed_kernel_sample - kernel_target, axis=1) / temp
            probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
            idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
            sample_j, sample_k = (j * 2), (k *2)
            sample[0, :, sample_j:sample_j+2, sample_k:sample_k+2] = kernel_sample[idx, :, :, :]
            transformed_sample[0, :, j, k] = transformed_kernel_sample[idx, :]
        return sample, transformed_sample
        
    def sample_from_convolutional(self, gmms, target
                                  , target_gmms, conv_filter, sample_size=100
                                  , temp=1, is_inputs=False, is_last_conv_layer=False):
        samples = list()
        transformed_samples = list()
        
        for i in range(int(sample_size/10)):
            sample, transformed_sample = self.generate_one_sample_from_convolutional(gmms, target
                                                                 , conv_filter, sample_size=100
                                                                 , temp=temp, is_inputs=is_inputs)
            if sample is not None:
                samples.append(sample)
                transformed_samples.append(transformed_sample)
        
        if len(samples) == 0:
            sys.exit("Uanble to generate a single in domain sample")
        
        if not is_last_conv_layer:
            in_domain = np.array([self._domain_check_convolutional_layer(torch.tensor(transformed_sample).float()
                                                        , target_gmms) for transformed_sample in transformed_samples])
        else:
            in_domain = np.array([self._domain_check_linear_layer(torch.tensor(transformed_sample.reshape(-1, self.n_layers * 2 * 2)).float()
                                                                  , target_gmms) for transformed_sample in transformed_samples])
        idxs = np.where(in_domain == True)[0]
        if idxs.shape[0] == 0:
            sys.exit("No samples were in domain")
        
        samples = np.concatenate(samples, axis=0)
        transformed_samples = np.concatenate(transformed_samples, axis=0)
        
        samples = samples[idxs]
        transformed_samples = transformed_samples[idxs].reshape(samples.shape[0], -1)
        target = target.reshape(1, -1)
        negative_distances = -1 * np.linalg.norm(transformed_samples - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        return np.expand_dims(samples[idx], axis=0)
        
    def sample(self, target=None, sample_size=1000, temp=1):
        if target is None:
            target, _ = self.gmm_linear3.sample()
        else:
            target = self.sample_from_outputs(self.gmm_linear3, target, sample_size=sample_size, temp=temp)
        
        target = self.sample_from_linear(self.gmm_linear2, target, self.gmm_linear3, self.linear3, sample_size=sample_size, temp=temp)
        target = self.sample_from_linear(self.gmm_linear1, target, self.gmm_linear2, self.linear2, sample_size=sample_size, temp=temp)
        target = self.sample_from_linear(self.gmm_conv2, target, self.gmm_linear1, self.linear1, sample_size=sample_size, temp=temp)
        
        target = target.reshape(-1, self.n_layers, 2, 2)
                
        target = self.sample_from_convolutional(self.gmm_conv1, target,  self.gmm_conv2, self.conv2, sample_size=sample_size, temp=temp
                                               , is_last_conv_layer=True)
        target = self.sample_from_convolutional(self.gmm_inputs, target, self.gmm_conv1, self.conv1, sample_size=sample_size, temp=temp
                                                , is_inputs=True)
        return target

domain_net = DomainNet()

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(domain_net.parameters(), lr=0.001)

data, labels = digits.images[:].reshape(-1, 1, 8, 8), digits.target
idxs = np.arange(data.shape[0])
np.random.shuffle(idxs)
val_data, val_labels = data[idxs[1000:]], labels[idxs[1000:]]
data, labels = data[idxs[:1000]], labels[idxs[:1000]]
data, labels = Variable(torch.tensor(data).float()), Variable(torch.tensor(labels))
val_data, val_labels = Variable(torch.tensor(val_data).float()), Variable(torch.tensor(val_labels))

print_every = (4, 9, 99, 499)
loss_history = list()
for epoch in range(500):
    running_loss = 0.0
    
    optimizer.zero_grad()
    
    outputs = domain_net(data)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()
    
    running_loss = loss.item()
    loss_history.append(running_loss)
    
    if epoch in print_every:
        fig = plt.figure()
        ax = fig.add_subplot(1, 1, 1)
        ax.plot(loss_history)
        fig.subplots_adjust(bottom=0.15)
        fig.canvas.draw()
        print("Epoch: " + str(epoch))
        print("Loss: " + str(loss_history[-1]))
        _, predictions = F.softmax(domain_net(data), dim=1).max(1)
        accuracy = 1 - (torch.nonzero(predictions - labels).size(0)) / predictions.size(0)
        print("Accuracy: " + str(accuracy))
        _, predictions = F.softmax(domain_net(val_data), dim=1).max(1)
        accuracy = 1 - (torch.nonzero(predictions - val_labels).size(0)) / predictions.size(0)
        print("Validation Accuracy: " + str(accuracy))
        time.sleep(0.2)
        plt.close(fig)
        timeit(lambda: domain_net.fit_domain(data, val=val_data, n_components=50), description="Fit domain")
        img = timeit(lambda: domain_net.sample(target=np.array([1,0,0,0,0,0,0,0,0,0]), temp=0.0001, sample_size=500)
                    , description="Sampling an image from the domain")
        fig = plt.figure(figsize=(1, 1))
        ax = fig.add_subplot(1, 1, 1)
        ax.get_yaxis().set_visible(False)
        ax.get_xaxis().set_visible(False)
        ax.imshow(img[0, 0, :, :])
        fig.canvas.draw()
        time.sleep(0.2)
        plt.close(fig)
        print("0 means in domain " + str(timeit(lambda: domain_net.domain_check(torch.tensor(img).float())
                                               , description="Checking whether an image is in domain")))
        print()
        print()
img = timeit(lambda: domain_net.sample(target=np.array([[0.1,0.1,0.1,0.1,0.1,0.1,0.1,0.1,0.1,0.1]]), temp=1, sample_size=500), description="Sampling")
img = np.clip(img, 0, 16).astype(int)
# print(img)
fig = plt.figure(figsize=(1, 1))
ax = fig.add_subplot(1, 1, 1)
ax.get_yaxis().set_visible(False)
ax.get_xaxis().set_visible(False)
ax.imshow(img[0, 0, :, :])
fig.canvas.draw()
time.sleep(0.2)
plt.close(fig)

print("0 means in domain " + str(timeit(lambda: domain_net.domain_check(torch.tensor(img).float())
                                               , description="Checking whether an image is in domain")))

a = np.round(F.softmax(domain_net(torch.tensor(img).float()), dim=1).detach().numpy(), 2)
print(a[0])
print(np.argmax(a[0]))
import torchvision<jupyter_output><empty_output><jupyter_text>We can use fashion MNIST for out of domain detection<jupyter_code>trainset = torchvision.datasets.FashionMNIST(root='./data/fashion', train=True, download=True)
trainset.train_data = F.pad(trainset.train_data, (2, 2, 2, 2), 'constant', 100)
testset = torchvision.datasets.FashionMNIST(root='./data/fashion', train=False, download=True)
testset.test_data = F.pad(testset.test_data, (2, 2, 2, 2), 'constant', 100)
print(testset.test_data[0].shape)
fig = plt.figure(figsize=(8, 1))
for i in range(10):
    ax = fig.add_subplot(1, 10, i+1)
    ax.get_yaxis().set_visible(False)
    ax.get_xaxis().set_visible(False)
    ax.imshow(testset.test_data[i])
fig.canvas.draw()
time.sleep(0.2)
plt.close(fig)<jupyter_output><empty_output><jupyter_text>We can also use EMNIST for out of domain detection<jupyter_code>trainset = torchvision.datasets.EMNIST(root='./data/emnist', split='letters', train=True, download=True)
trainset.train_data = F.pad(trainset.train_data, (2, 2, 2, 2), 'constant', 100)
testset = torchvision.datasets.EMNIST(root='./data/emnist', split='letters', train=False, download=True)
testset.test_data = F.pad(testset.test_data, (2, 2, 2, 2), 'constant', 100)
print(testset.test_data[0].shape)
len(testset)
random_idxs = np.random.randint(1, 20800, 20)
fig = plt.figure(figsize=(8, 2))
for i in range(20):
    ax = fig.add_subplot(2, 10, i+1)
    ax.get_yaxis().set_visible(False)
    ax.get_xaxis().set_visible(False)
    ax.imshow(testset.test_data[random_idxs[i]])
fig.canvas.draw()
time.sleep(0.2)
plt.close(fig)<jupyter_output><empty_output><jupyter_text>Do the same thing but with MNIST<jupyter_code>trainset = torchvision.datasets.MNIST(root='./data/mnist', train=True, download=True)
trainset.train_data = F.pad(trainset.train_data, (2, 2, 2, 2), 'constant', 100)
testset = torchvision.datasets.MNIST(root='./data/mnist', train=False, download=True)
testset.test_data = F.pad(testset.test_data, (2, 2, 2, 2), 'constant', 100)
fig = plt.figure(figsize=(8, 1))
for i in range(10):
    ax = fig.add_subplot(1, 10, i+1)
    ax.get_yaxis().set_visible(False)
    ax.get_xaxis().set_visible(False)
    ax.imshow(testset.test_data[i])
fig.canvas.draw()
time.sleep(0.2)
plt.close(fig)
class DomainNet(nn.Module):
    
    def __init__(self, n_layers=n_layers, n_classes=n_classes):
        super(DomainNet, self).__init__()
        
        self.n_layers = n_layers
        self.n_classes = n_classes
        
        ##### layers #####
        self.conv1 = nn.Conv2d(1, n_layers, kernel_size=2, stride=2)
        self.conv2 = nn.Conv2d(n_layers, n_layers, kernel_size=2, stride=2)
        
        self.linear1 = nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2)
        self.linear2 = nn.Linear(n_layers * 2 * 2, n_layers * 2 * 2)
        self.linear3 = nn.Linear(n_layers * 2 * 2, n_classes)
        #### layers #####
        
        ### Gaussian Mixture Models ###
        self.gmm_inputs = list()
        self.gmm_conv1 =  list()
        self.gmm_conv2 =  None
        
        self.gmm_linear1 = None
        self.gmm_linear2 = None
        self.gmm_linear3 = None
        ### Gaussian Mixture Models ###
        return
    
    def forward(self, x):
        outputs = F.leaky_relu(self.conv1(x))
        outputs = F.leaky_relu(self.conv2(outputs))
        
        outputs = outputs.view(outputs.shape[0], self.n_layers * 2 * 2)
        
        outputs = F.leaky_relu(self.linear1(outputs))
        outputs = F.leaky_relu(self.linear2(outputs))
        outputs = self.linear3(outputs)
        return outputs
    
    def _fit_domain_of_convolutional_layer(self, outputs, val=None, n_components=50):
        gmms = list()
        for j in range(0, outputs.shape[2] - 1, 2):
            for k in range(0, outputs.shape[3] - 1, 2):
                kernel = outputs[:, :, j:j+2, k:k+2]
                gmm = GaussianMixture(covariance_type='full',
                                    n_components=n_components).fit(
                                    kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                if val is not None:
                    kernel = val[:, :, j:j+2, k:k+2]
                probabilities = gmm.score_samples(kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                gmm.threshold = np.amin(probabilities)
                gmms.append(gmm)
        return gmms
    
    def _fit_domain_of_linear_layer(self, outputs, val=None, n_components=50):
        gmm = GaussianMixture(covariance_type='full',
                                         n_components=n_components).fit(outputs.detach().numpy())
        if val is None:
            probabilities = gmm.score_samples(outputs.detach().numpy())
        else:
            probabilities = gmm.score_samples(val.detach().numpy())
        gmm.threshold = np.amin(probabilities)
        return gmm
        
    def fit_domain(self, x, val=None, n_components=50):
        self.gmm_inputs = self._fit_domain_of_convolutional_layer(x, val=val, n_components=n_components)
        outputs = self.conv1(x)
        if val is not None:
            val = self.conv1(val)
        
        self.gmm_conv1 = self._fit_domain_of_convolutional_layer(outputs, val=val, n_components=n_components)
        outputs = self.conv2(F.leaky_relu(outputs))
        if val is not None:
            val = self.conv2(F.leaky_relu(val))
        
        outputs = outputs.view(outputs.size(0), self.n_layers * 2 * 2)
        if val is not None:
            val = val.view(val.size(0), self.n_layers * 2 * 2)
    
        self.gmm_conv2 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear1(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear1(F.leaky_relu(val))
        
        self.gmm_linear1 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear2(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear2(F.leaky_relu(val))
        
        self.gmm_linear2 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        outputs = self.linear3(F.leaky_relu(outputs))
        if val is not None:
            val = self.linear3(F.leaky_relu(val))
        
        self.gmm_linear3 = self._fit_domain_of_linear_layer(outputs, val=val, n_components=n_components)
        return
    
    def _domain_check_convolutional_layer(self, outputs, gmms):
        i = 0
        for j in range(0, outputs.shape[2] - 1, 2):
            for k in range(0, outputs.shape[3] - 1, 2):
                kernel = outputs[:, :, j:j+2, k:k+2]
                probabilities = gmms[i].score_samples(
                    kernel.detach().numpy().reshape(-1, kernel.shape[1] * 2 * 2))
                if probabilities[0] < gmms[i].threshold:
                    return False
                i +=1
        return True
    
    def _domain_check_linear_layer(self, outputs, gmm):
        probabilities = gmm.score_samples(outputs.detach().numpy())
        if probabilities[0] < gmm.threshold:
            return False
        return True
    
    def domain_check(self, x):
        """
        Currently only supports checking one image at a time
        """
        passing = self._domain_check_convolutional_layer(x, self.gmm_inputs)
        if not passing:
            return -1
        outputs = self.conv1(x)
        
        passing = self._domain_check_convolutional_layer(outputs, self.gmm_conv1)
        if not passing:
            return -2
        outputs = self.conv2(F.leaky_relu(outputs))
        
        outputs = outputs.view(outputs.size(0), self.n_layers * 2 * 2)
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_conv2)
        if not passing:
            return -3
        outputs = self.linear1(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear1)
        if not passing:
            return -4
        outputs = self.linear2(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear2)
        if not passing:
            return -5
        outputs = self.linear3(F.leaky_relu(outputs))
        
        passing = self._domain_check_linear_layer(outputs, self.gmm_linear3)
        if not passing:
            return -6
        
        # It is in domain
        return 0
    
    def sample_from_linear(self, gmm, target, target_gmm, linear_transformation, sample_size=100, temp=1):
        max_iterations = 1000
        for i in range(max_iterations):
            if max_iterations - i < 5:
                print("sampling iteration: " + str(i))
            samples, _ = gmm.sample(sample_size)
            scores = gmm.score_samples(samples)
            idxs = np.where(scores >= gmm.threshold)[0]
            if idxs.shape[0] != 0:
                samples = samples[idxs]
                squashed_sample = linear_transformation(F.leaky_relu(torch.tensor(samples).float())).detach().numpy()
                scores = target_gmm.score_samples(squashed_sample)
                idxs = np.where(scores >= target_gmm.threshold)[0]
                if idxs.shape[0] != 0:
                    samples = samples[idxs]
                    squashed_sample = squashed_sample[idxs]
                    break
            if i == (max_iterations - 1):
                sys.exit("Sampling took to long we timed out")       
        negative_distances = -1 * np.linalg.norm(squashed_sample - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        new_target = samples[idx]
        return new_target
    
    def sample_from_outputs(self, gmm, target, sample_size=100, temp=1):
        max_iterations = 1000
        for i in range(max_iterations):
            if max_iterations - i < 5:
                print("sampling iteration: " + str(i))
            samples, _ = gmm.sample(sample_size)
            scores = gmm.score_samples(samples)
            idxs = np.where(scores >= gmm.threshold)[0]
            if idxs.shape[0] != 0:
                samples = samples[idxs]
                break
            if i == (max_iterations - 1):
                sys.exit("Sampling took to long we timed out")
        
        squashed_samples = F.softmax(torch.tensor(samples).float(), dim=1).detach().numpy()
        negative_distances = -1 * np.linalg.norm(squashed_samples - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        return samples[idx]
    
    def generate_one_sample_from_convolutional(self, gmms, target,
                                               conv_filter, sample_size=100, temp=1, is_inputs=False):
        max_iterations = 1000
        # create an array filled with nans that has the shape of a sample
        sample = np.zeros((1, conv_filter.weight.shape[1], target.shape[2] * 2, target.shape[3] * 2))
        sample.fill(np.nan)
        
        # create an array filled with nans that has the shape of target
        transformed_sample = np.zeros(target.shape)
        transformed_sample.fill(np.nan)
        
        # create a list of all the kernel locations and shuffle it
        kernel_idxs = list()
        i = 0
        for j in range(target.shape[2]):
            for k in range(target.shape[3]):
                kernel_idxs.append((i, j, k))
                i += 1
        random.shuffle(kernel_idxs)
        
        # sample from each kernel
        # make sure each sample is in domain. If you can't generate an in domain sample give up after a while
        # and return None
        for (i, j, k) in kernel_idxs:
            kernel_target = target[:, :, j, k].reshape(-1, self.n_layers)
            for m in range(max_iterations):
                if max_iterations - m < 5:
                    print("sampling iteration: " + str(m))
                kernel_sample, _ = gmms[i].sample(sample_size)
                scores = gmms[i].score_samples(kernel_sample)
                print
                idxs = np.where(scores >= gmms[i].threshold)[0]
                if idxs.shape[0] != 0:
                    kernel_sample = kernel_sample[idxs]
                    kernel_sample = kernel_sample.reshape(-1, conv_filter.weight.shape[1], 2, 2)
                    break
                if m == (max_iterations - 1):
                    print("Gave up on this iteration")
                    return None, None
                
            if is_inputs:
                # If its the input there shouldn't be any non-linearity
                transformed_kernel_sample = conv_filter(
                    torch.tensor(kernel_sample).float()).detach().numpy().reshape(-1, self.n_layers)
            else:
                transformed_kernel_sample = conv_filter(F.leaky_relu(
                    torch.tensor(kernel_sample).float())).detach().numpy().reshape(-1, self.n_layers)
            
            negative_distances = -1 * np.linalg.norm(transformed_kernel_sample - kernel_target, axis=1) / temp
            probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
            idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
            sample_j, sample_k = (j * 2), (k *2)
            sample[0, :, sample_j:sample_j+2, sample_k:sample_k+2] = kernel_sample[idx, :, :, :]
            transformed_sample[0, :, j, k] = transformed_kernel_sample[idx, :]
        return sample, transformed_sample
        
    def sample_from_convolutional(self, gmms, target
                                  , target_gmms, conv_filter, sample_size=100
                                  , temp=1, is_inputs=False, is_last_conv_layer=False):
        samples = list()
        transformed_samples = list()
        
        for i in range(int(sample_size/10)):
            sample, transformed_sample = self.generate_one_sample_from_convolutional(gmms, target
                                                                 , conv_filter, sample_size=100
                                                                 , temp=temp, is_inputs=is_inputs)
            if sample is not None:
                samples.append(sample)
                transformed_samples.append(transformed_sample)
        
        if len(samples) == 0:
            sys.exit("Uanble to generate a single in domain sample")
        
        if not is_last_conv_layer:
            in_domain = np.array([self._domain_check_convolutional_layer(torch.tensor(transformed_sample).float()
                                                        , target_gmms) for transformed_sample in transformed_samples])
        else:
            in_domain = np.array([self._domain_check_linear_layer(torch.tensor(transformed_sample.reshape(-1, self.n_layers * 2 * 2)).float()
                                                                  , target_gmms) for transformed_sample in transformed_samples])
        idxs = np.where(in_domain == True)[0]
        if idxs.shape[0] == 0:
            sys.exit("No samples were in domain")
        
        samples = np.concatenate(samples, axis=0)
        transformed_samples = np.concatenate(transformed_samples, axis=0)
        
        samples = samples[idxs]
        transformed_samples = transformed_samples[idxs].reshape(samples.shape[0], -1)
        target = target.reshape(1, -1)
        negative_distances = -1 * np.linalg.norm(transformed_samples - target, axis=1) / temp
        probability_weights = F.softmax(torch.tensor(negative_distances).float(), dim=0).detach().numpy()
        idx = np.random.choice(np.arange(probability_weights.shape[0]), p=probability_weights)
        print("Distance between sampled point and target point " + str(-1 * negative_distances[idx] * temp))
        return np.expand_dims(samples[idx], axis=0)
        
    def sample(self, target=None, sample_size=1000, temp=1):
        if target is None:
            target, _ = self.gmm_linear3.sample()
        else:
            target = self.sample_from_outputs(self.gmm_linear3, target, sample_size=sample_size, temp=temp)
        
        target = self.sample_from_linear(self.gmm_linear2, target, self.gmm_linear3, self.linear3, sample_size=sample_size, temp=temp)
        target = self.sample_from_linear(self.gmm_linear1, target, self.gmm_linear2, self.linear2, sample_size=sample_size, temp=temp)
        target = self.sample_from_linear(self.gmm_conv2, target, self.gmm_linear1, self.linear1, sample_size=sample_size, temp=temp)
        
        target = target.reshape(-1, self.n_layers, 2, 2)
                
        target = self.sample_from_convolutional(self.gmm_conv1, target,  self.gmm_conv2, self.conv2, sample_size=sample_size, temp=temp
                                               , is_last_conv_layer=True)
        target = self.sample_from_convolutional(self.gmm_inputs, target, self.gmm_conv1, self.conv1, sample_size=sample_size, temp=temp
                                               , is_inputs=True)
        return target
a = min(np.array([5]), np.array([1]))
a
np.exp(1)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup