ensemble_learning_code.py

from models import CustomModel
from train import TrainCommands
from torchvision import transforms
from PIL import Image
import numpy as np
import itertools
import requests
import torch
import json
import math
import time
import os

def calculate_fully_connected(layers, size):
	for layer in layers:
		k = 1
		p = 0
		s = 1
		d = 1
		if "kernel_size" in layer:
			k = layer["kernel_size"]
		if "padding" in layer:
			p = layer["padding"]
		if "stride" in layer:
			s = layer["stride"]
		if "dilation" in layer:
			d = layer["dilation"]
		size = math.floor((size + 2*p - d*(k-1) - 1)/s + 1)
	
	return size

model = [
	{
		"name": "Conv2d",
		"in_channels": 3,
		"out_channels": 6,
		"kernel_size": 5
	},
	{
		"name": "ReLU",
		"inplace": True
	},
	{
		"name": "MaxPool2d",
		"kernel_size": 2,
		"stride": 2
	},
	{
		"name": "Conv2d",
		"in_channels": 6,
		"out_channels": 16,
		"kernel_size": 5
	},
	{
		"name": "ReLU",
		"inplace": True
	},
	{
		"name": "MaxPool2d",
		"kernel_size": 2,
		"stride": 2
	},
	{
		"name": "Flatten",
		"start_dim": 1
	},
	{
		"name": "Linear",
		"in_features": "change_with_calculated_fn_size",
		"out_features": 120
	},
	{
		"name": "ReLU",
		"inplace": True
	},
	{
		"name": "Linear",
		"in_features": 120,
		"out_features": 84
	},
	{
		"name": "ReLU",
		"inplace": True
	},
	{
		"name": "Linear",
		"in_features": 84,
		"out_features": "n_labels"
	}
]

optimizer = "SGD"
lr = 0.1
momentum = 0.9
loss_function = "CrossEntropyLoss"
output_size = 16
image_size = 32

labels = [
	"American Hairless Terrier imagesize:500x500",
	"Alaskan Malamute imagesize:500x500",
	"American Eskimo Dog imagesize:500x500",
	#"Australian Shepherd imagesize:500x500",
	#"Boston Terrier imagesize:500x500",
	#"Boykin Spaniel imagesize:500x500",
	#"Chesapeake Bay Retriever imagesize:500x500",
	#"Catahoula Leopard Dog imagesize:500x500",
	#"Toy Fox Terrier imagesize:500x500"
]

label_combinations = []
index_list = list(range(0,len(labels)))
index_list = list(itertools.combinations(index_list,2))
index_list = list(set(index_list))

for indexes in index_list:
	label_combinations = label_combinations + [[labels[indexes[0]], labels[indexes[1]]]]


training_dicts = []
for binary_labels in label_combinations:
	model_name = binary_labels[0].split(" imagesize")[0].replace(" ","_").lower()
	model_name = model_name + "_vs_" + binary_labels[1].split(" imagesize")[0].replace(" ","_").lower()
	calculated_fc_size = calculate_fully_connected(model,image_size)
	for layer in model:
		if (layer["name"] == "Linear") and (layer["in_features"] == "change_with_calculated_fn_size"):
			model[model.index(layer)]['in_features'] = calculated_fc_size * calculated_fc_size * output_size ## Assuming image shape and kernel are squares
			break
	training_dict = {
		"model_name": model_name,
		"criterion": {
			"name": loss_function
		},
		"optimizer": {
			"name": optimizer,
			"lr": lr,
			"momentum": momentum
		},
		"batch_size": 4,
		"n_epoch": 10,
		"n_labels": 0,
		"image_ops": [
			{
				"resize": {
					"size": [
						image_size,
						image_size
					],
					"resample": "Image.ANTIALIAS"
				}
			},
			{
				"convert": {
					"mode": "'RGB'"
				}
			}
		],
		"transform": {
			"ToTensor": True,
			"Normalize": {
				"mean": [
					0.5,
					0.5,
					0.5
				],
				"std": [
					0.5,
					0.5,
					0.5
				]
			}
		},
		"target_transform": {
			"ToTensor": True
		},
		"label_names": binary_labels,
		"model": {
			"name": "",
			"layers": model
		}
	}
	training_dicts = training_dicts + [training_dict]

trained_models = os.listdir("models")

for training_dict in training_dicts:
	if (training_dict['model_name']+".pt") not in trained_models:
		print("---")
		print("Training Model: {}".format(training_dict['model_name']))
		body = json.dumps(training_dict)
		response = requests.post("http://localhost:8000/train", headers = {"Content-Type": "application/json"}, data=body, allow_redirects = True)
		if response.status_code == 200:
			while True:
				response = requests.post("http://localhost:8000/find_attempt/?name={}".format(training_dict["model_name"]), headers = {"Content-Type": "application/json"}, allow_redirects = True)
				if response.json() != None and response.json()['status'] == "Trained":
					break
				time.sleep(1)
		testing_dict = training_dict
		testing_dict['limit'] = 100
		body = json.dumps(testing_dict)
		response = requests.post("http://localhost:8000/test", headers = {"Content-Type": "application/json"}, data=body, allow_redirects = True)
		if response.status_code == 200:
			while True:
				response = requests.post("http://localhost:8000/find_attempt/?name={}".format(training_dict["model_name"]), headers = {"Content-Type": "application/json"}, allow_redirects = True)
				if response.json() != None and response.json()['status'] == "Complete":
					break
				time.sleep(1)
		print("Accuracy: {}".format(response.json()['accuracy']))
		print("---")
	else:
		print("---")
		print("Skipping Already Existing Model: {}".format(training_dict['model_name']))
		print("---")

device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
img = Image.open("examples/malamute_example.jpg")
transform = transforms.Compose([transforms.Resize((32,32), antialias=True), transforms.ToTensor(), transforms.Normalize(mean=(0.5,0.5,0.5), std=(0.5,0.5,0.5))])
img = transform(img)
img = [img.numpy()]
img = np.asarray(img, dtype='float64')
img = torch.from_numpy(img).float()
img = img.to(device)

predictions = []
for example_dict in training_dicts:
	model_path = "models/" + example_dict['model_name'] + ".pt"
	example_dict['n_labels'] = 2
	label_names = example_dict['label_names']
	example_dict = TrainCommands(**example_dict)
	model = CustomModel(example_dict)
	model.load_state_dict(torch.load(model_path))
	if torch.cuda.is_available():
		model.cuda()
		pred = model(img).to(device)[0]
	else:
		prediction = model(img)[0]
	pred = label_names[prediction.argmax()]
	predictions = predictions + [pred]

final_prediction = max(set(predictions), key = predictions.count)
if " imagesize:" in final_prediction:
	final_prediction = final_prediction.split(" imagesize")[0]

print("Prediction is {}".format(final_prediction))