Automatic_Evolution.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import random\n",
    "import torch\n",
    "import copy\n",
    "import torch.optim as optim\n",
    "import torch.utils.data\n",
    "import torchvision.transforms as transforms\n",
    "import torchvision.utils as vutils\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import matplotlib.animation as animation\n",
    "from IPython.display import HTML\n",
    "from numpy.random import randint\n",
    "import scipy.stats as stats\n",
    "\n",
    "# save images\n",
    "from torchvision.utils import save_image\n",
    "img_save_path = './plots/'\n",
    "os.makedirs(img_save_path, exist_ok=True)\n",
    "\n",
    "from CAN.parameters import *\n",
    "from CAN.model_CAN_16_9 import * \n",
    "\n",
    "from deap import base, creator, tools, algorithms\n",
    "\n",
    "from keras.models import Model\n",
    "from keras.layers import Dense, Dropout\n",
    "from keras.applications.inception_resnet_v2 import InceptionResNetV2\n",
    "from keras.applications.inception_resnet_v2 import preprocess_input\n",
    "from keras.utils.image_utils import load_img, img_to_array\n",
    "import tensorflow as tf\n",
    "from PIL import Image\n",
    "\n",
    "from NIMA.utils.score_utils import mean_score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set random seed for reproducibility\n",
    "# used seeds for simulation: 3,5,7,9,11\n",
    "manualSeed = 7\n",
    "#manualSeed = random.randint(1, 10000)\n",
    "print(\"Random Seed: \", manualSeed)\n",
    "random.seed(manualSeed)\n",
    "torch.manual_seed(manualSeed)\n",
    "np.random.seed(manualSeed)\n",
    "\n",
    "\n",
    "# Decide which device we want to run on\n",
    "device = torch.device(\"cuda:0\" if (torch.cuda.is_available() and ngpu > 0) else \"cpu\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### load Generator model\n",
    "\n",
    "# Create the generator\n",
    "netG = Generator(ngpu).to(device)\n",
    "\n",
    "# Setup Adam optimizers for G\n",
    "optimizerG = optim.Adam(netG.parameters(), lr=lr, betas=(beta1, 0.999))\n",
    "\n",
    "# Path to model parameters\n",
    "PATH_GEN = './CAN/models/GEN_16_9.pth'\n",
    "\n",
    "# load checkpoint\n",
    "checkpoint = torch.load(PATH_GEN, map_location=torch.device('cpu'))\n",
    "netG.load_state_dict(checkpoint['model_state_dict'])\n",
    "optimizerG.load_state_dict(checkpoint['optimizer_state_dict'])\n",
    "\n",
    "# set model to evaluation\n",
    "netG.eval()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### load Evaluation model\n",
    "\n",
    "netEval = InceptionResNetV2(input_shape=(None, None, 3), include_top=False, pooling='avg', weights=None)\n",
    "x = Dropout(0.75)(netEval.output)\n",
    "x = Dense(10, activation='softmax')(x)\n",
    "netEval = Model(netEval.input, x)\n",
    "netEval.load_weights('./NIMA/weights/inception_resnet_weights.h5')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Fitness functions\n",
    "\n",
    "# simple random fitness\n",
    "def fitnessRandom(ind):\n",
    "    return np.random.uniform(), # deap needs the comma\n",
    "\n",
    "\n",
    "# automatic evaluation\n",
    "\n",
    "def AutomaticEvaluation(img_path):\n",
    "    \n",
    "    with tf.device('/CPU:0'):\n",
    "        # load image, unfortunately from path due to problems in converting torch.tensor to PilImage\n",
    "        target_size = (224, 224)\n",
    "        image = load_img(img_path, target_size=target_size)\n",
    "        image = img_to_array(image)\n",
    "        image = np.expand_dims(image, axis=0)\n",
    "        image = preprocess_input(image)\n",
    "\n",
    "        # evaluate the image\n",
    "        score = netEval.predict(image, batch_size=1, verbose=0)[0]\n",
    "        score = mean_score(score)\n",
    "\n",
    "    return score\n",
    "\n",
    "def fitnessAutomaticEvaluation(ind):\n",
    "\n",
    "    # part 1: create image\n",
    "    \n",
    "    with torch.no_grad():\n",
    "        ind = torch.tensor(ind)\n",
    "        ind = ind.view(1,nz,1,1)\n",
    "        image = netG(ind.float()).detach().cpu()\n",
    "\n",
    "        # not really sexy, but when directly converting to PilImage, image is slightly changed!\n",
    "        # delete after reading it in, no need to keep\n",
    "        save_image(image.data, \"current_individual.png\", normalize=True)\n",
    "    \n",
    "    # part 2: evluate image and delete it\n",
    "\n",
    "    score = AutomaticEvaluation(\"current_individual.png\")\n",
    "    os.remove(\"current_individual.png\")\n",
    "\n",
    "\n",
    "    return score, # deap needs the comma\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### DEAP setup\n",
    "\n",
    "# maximation, one objective\n",
    "creator.create(\"FitnessMax\", base.Fitness, weights=(1.0,))\n",
    "creator.create(\"Individual\", list, fitness=creator.FitnessMax)\n",
    "\n",
    "# set up individuals, random normal distribution\n",
    "toolbox = base.Toolbox()\n",
    "toolbox.register(\"attribute\", np.random.normal)\n",
    "toolbox.register(\"individual\", tools.initRepeat, creator.Individual,\n",
    "                 toolbox.attribute, n=nz)\n",
    "toolbox.register(\"population\", tools.initRepeat, list, toolbox.individual)\n",
    "\n",
    "### set up mating, mutation, selection\n",
    "\n",
    "# from experiments with recombination, uniform with indpb = .25 promising\n",
    "toolbox.register(\"mate\", tools.cxUniform, indpb = 0.25)\n",
    "# only used in local search, else overwritten\n",
    "toolbox.register(\"mutate\", tools.mutGaussian, mu=0, sigma=1, indpb=1/100)\n",
    "toolbox.register(\"select\", tools.selStochasticUniversalSampling)\n",
    "toolbox.register(\"evaluate\", fitnessAutomaticEvaluation) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### save some statistics of each generation\n",
    "stats = tools.Statistics(key=lambda ind: ind.fitness.values)\n",
    "stats.register(\"max\", np.max)\n",
    "stats.register(\"avg\", np.mean)\n",
    "stats.register(\"min\", np.min)\n",
    "stats.register(\"std\", np.std)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_Evolution(MU = 10, crossP = 0.5, mutP = 0.5, NGEN = 10):\n",
    "\n",
    "    # set fitness function\n",
    "    toolbox.register(\"evaluate\", fitnessRandom)\n",
    "    toolbox.register(\"mutate\", tools.mutGaussian, mu=0, sigma=0.1, indpb=0.5)\n",
    "\n",
    "    img_list = []\n",
    "    genes = torch.randn(MU, nz, 1, 1, device=device)\n",
    "\n",
    "\n",
    "    with torch.no_grad():\n",
    "        fake = netG(genes).detach().cpu()\n",
    "        img_list.append(vutils.make_grid(fake, padding=2, normalize=True))\n",
    "        save_image(fake.data, img_save_path + '/evolution_random_%d.png' % (0), normalize=True)\n",
    "\n",
    "\n",
    "    #log and hof\n",
    "    log = tools.Logbook()\n",
    "    log.header = \"gen\", \"new\", \"max\", \"mean\", \"min\", \"std\"\n",
    "    hof = tools.HallOfFame(maxsize=20)\n",
    "    \n",
    "    # population\n",
    "    pop = toolbox.population(n=MU)\n",
    "    # genes replacing the pop\n",
    "    for i in range(MU):\n",
    "        ind = genes[i].tolist()\n",
    "        #flatten lists\n",
    "        ind = [item for sublist in ind for item in sublist]\n",
    "        ind = [item for sublist in ind for item in sublist]\n",
    "        pop[i] = (creator.Individual(ind))\n",
    "\n",
    "    results = []\n",
    "\n",
    "    for i in range(NGEN):\n",
    "        # let the algorithm run for one round only, then plot the results again\n",
    "        pop, log = algorithms.eaSimple(pop, toolbox, cxpb=crossP, mutpb=mutP, ngen=1, stats=stats, halloffame=hof)\n",
    "        results.append(log)\n",
    "\n",
    "        with torch.no_grad():\n",
    "            new_genes = torch.tensor(pop)\n",
    "            new_genes = new_genes.view(MU,nz,1,1)\n",
    "            fake = netG(new_genes.float()).detach().cpu()\n",
    "            img_list.append(vutils.make_grid(fake, padding=2, normalize=True))\n",
    "            save_image(fake.data, img_save_path + '/evolution_random_%d.png' % (i+1), normalize=True)\n",
    "\n",
    "\n",
    "    return pop, results, hof, img_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## local search instead of random mutation\n",
    "def local_Search(genes, NGEN = 100):\n",
    "\n",
    "    pop = toolbox.population(n=1)\n",
    "    pop[0] = (creator.Individual(genes))\n",
    "    ref = copy.deepcopy(pop[0])\n",
    "\n",
    "    # switch to selBest, mutation is only used here and can be kept\n",
    "    toolbox.register(\"select\", tools.selBest)\n",
    "\n",
    "    offspring = algorithms.eaMuPlusLambda(pop, toolbox, mu=1, lambda_=1, \n",
    "        cxpb=0, mutpb=1, ngen=NGEN, stats=stats, verbose=False)\n",
    "    \n",
    "    # switch back\n",
    "    toolbox.register(\"select\", tools.selStochasticUniversalSampling)\n",
    "\n",
    "    same = False\n",
    "    if all(x == y for x, y in zip(ref, offspring[0][0])):\n",
    "        same = True\n",
    "\n",
    "    return offspring[0][0], same"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# diversity preservation metric to avoid too close individuals:\n",
    "\n",
    "def preserve_Diversity(pop, MU, threshold = 25):\n",
    "    invalid_inds = []\n",
    "    dist = []\n",
    "    # Threshold ~ 5-30, arbitrary\n",
    "\n",
    "    for i in range(MU):\n",
    "        for j in range(MU):\n",
    "            if j > i:\n",
    "                dist.append(sum([abs(x-y) for x, y in zip(pop[i], pop[j])]))\n",
    "                if sum([abs(x-y) for x, y in zip(pop[i], pop[j])]) < threshold:\n",
    "                    invalid_inds.append(i)\n",
    "\n",
    "    return np.unique(invalid_inds).tolist(), dist "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def complex_Evolution(MU = 15, crossP = 0.5, mutP = 0.5, NGEN = 25):\n",
    "\n",
    "    # create first generation\n",
    "    img_list = []\n",
    "    genes = torch.randn(MU, nz, 1, 1, device=device)\n",
    "\n",
    "    #log and hof\n",
    "    logbook = tools.Logbook()\n",
    "    logbook.header = ['gen', 'nevals', 'imm'] + (stats.fields if stats else [])\n",
    "    hof = tools.HallOfFame(maxsize=20)\n",
    "    \n",
    "    # population\n",
    "    pop = toolbox.population(n=MU)\n",
    "    # genes replacing the pop\n",
    "    for i in range(MU):\n",
    "        ind = genes[i].tolist()\n",
    "        #flatten lists\n",
    "        ind = [item for sublist in ind for item in sublist]\n",
    "        ind = [item for sublist in ind for item in sublist]\n",
    "        pop[i] = (creator.Individual(ind))\n",
    "\n",
    "\n",
    "    ## evaluation of first generation\n",
    "    invalid_ind = [ind for ind in pop if not ind.fitness.valid]\n",
    "    fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)\n",
    "    for ind, fit in zip(invalid_ind, fitnesses):\n",
    "        ind.fitness.values = fit\n",
    "\n",
    "    # get stats from first generation\n",
    "    record = stats.compile(pop)\n",
    "    logbook.record(gen=0, nevals=len(invalid_ind), imm=0, **record)\n",
    "    hof.update(pop)\n",
    "    print(logbook.stream)\n",
    "\n",
    "\n",
    "    for i in range(NGEN):\n",
    "\n",
    "        # save images\n",
    "        with torch.no_grad():\n",
    "            new_genes = torch.tensor(pop)\n",
    "            new_genes = new_genes.view(MU,nz,1,1)\n",
    "            fake = netG(new_genes.float()).detach().cpu()\n",
    "            img_list.append(vutils.make_grid(fake, padding=2, normalize=True, nrow = 5))\n",
    "            save_image(fake.data, img_save_path + '/evolution_automatic_%d.png' % (i), normalize=True, nrow = 5)\n",
    "\n",
    "\n",
    "        # let the algorithm run for one round only, then plot the results again\n",
    "        \n",
    "        # Select the next generation individuals\n",
    "        offspring = toolbox.select(pop, len(pop))\n",
    "        # Clone the selected individuals\n",
    "        offspring = list(map(toolbox.clone, offspring))\n",
    "\n",
    "        # Apply crossover on the offspring\n",
    "        for child1, child2 in zip(offspring[::2], offspring[1::2]):\n",
    "            if random.random() < crossP:\n",
    "                toolbox.mate(child1, child2)\n",
    "                del child1.fitness.values\n",
    "                del child2.fitness.values\n",
    "\n",
    "        # Apply mutation on the offspring: Here memetic algorithm using local search!\n",
    "        for mutant in range(len(offspring)):\n",
    "            if random.random() < mutP:\n",
    "                offspring[mutant], same = local_Search(offspring[mutant])\n",
    "                if same == False:\n",
    "                    del offspring[mutant].fitness.values\n",
    "\n",
    "\n",
    "        # Preserve diversity, for novelty search + exploration + against user fatigue\n",
    "        boring_inds, _ = preserve_Diversity(offspring, MU)\n",
    "        for ind in boring_inds:\n",
    "\n",
    "            ## to check the measure:\n",
    "            with torch.no_grad():\n",
    "                new_genes = torch.tensor(offspring[ind])\n",
    "                new_genes = new_genes.view(1,nz,1,1)\n",
    "                fake = netG(new_genes.float()).detach().cpu()\n",
    "                save_image(fake.data, img_save_path + '/evolution_automatic_%d_replacedInd_%d.png' % (i+1, ind), normalize=True, nrow = 5)\n",
    "\n",
    "            # replace by random immigrant if to close to other individuals\n",
    "            offspring[ind] = toolbox.population(n=1)[0]\n",
    "            # and go through local search directly\n",
    "            offspring[ind],_ = local_Search(offspring[ind])\n",
    "            del offspring[ind].fitness.values\n",
    "\n",
    "        # Evaluate the individuals with an invalid fitness\n",
    "        invalid_ind = [ind for ind in offspring if not ind.fitness.valid]\n",
    "        fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)\n",
    "        for ind, fit in zip(invalid_ind, fitnesses):\n",
    "            ind.fitness.values = fit\n",
    "\n",
    "        # The population is entirely replaced by the offspring\n",
    "        pop[:] = offspring\n",
    "\n",
    "        # statistics and hall of fame\n",
    "        record = stats.compile(pop)\n",
    "        logbook.record(gen=i+1, nevals=len(invalid_ind), imm=len(boring_inds), **record)\n",
    "        hof.update(pop)\n",
    "        print(logbook.stream)\n",
    "\n",
    "        ## upscale hall of fame after last generation\n",
    "        if i == 24:\n",
    "            \n",
    "            from high_resolution import upscale\n",
    "\n",
    "            for j, img in enumerate(hof):\n",
    "\n",
    "                # save images\n",
    "                with torch.no_grad():\n",
    "                    new_genes = torch.tensor(img)\n",
    "                    new_genes = new_genes.view(1,nz,1,1)\n",
    "                    fake = netG(new_genes.float()).detach().cpu()\n",
    "                    save_image(fake.data, img_save_path + '/evolution_automatic_hall_of_fame_%d.png' % (j), normalize=True)\n",
    "\n",
    "\n",
    "                upscale(img_path = img_save_path + '/evolution_automatic_hall_of_fame_%d.png' % (j), scale = 8)\n",
    "\n",
    "    # finally also save last generation    \n",
    "    with torch.no_grad():\n",
    "        new_genes = torch.tensor(pop)\n",
    "        new_genes = new_genes.view(MU,nz,1,1)\n",
    "        fake = netG(new_genes.float()).detach().cpu()\n",
    "        img_list.append(vutils.make_grid(fake, padding=2, normalize=True, nrow = 5))\n",
    "        save_image(fake.data, img_save_path + '/evolution_automatic_%d.png' % (i+1), normalize=True, nrow = 5)\n",
    "\n",
    "    return pop, logbook, hof, img_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pop, log, hof, img_list = complex_Evolution()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check whether after evolution genes still fit expected distribution\n",
    "avg = []\n",
    "std = []\n",
    "for i,ind in enumerate(pop):\n",
    "    avg.append(np.mean(ind))\n",
    "    std.append(np.std(ind))\n",
    "print(np.mean(avg))\n",
    "print(np.mean(std))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# write Logbook as csv\n",
    "gen = []\n",
    "nevals = []\n",
    "imm = []\n",
    "maxi = []\n",
    "avg = []\n",
    "mini = []\n",
    "std = []\n",
    "\n",
    "for entry in log:\n",
    "    gen.append(entry['gen'])\n",
    "    nevals.append(entry['nevals'])\n",
    "    imm.append(entry['imm'])\n",
    "    maxi.append(entry['max'])\n",
    "    avg.append(entry['avg'])\n",
    "    mini.append(entry['min'])\n",
    "    std.append(entry['std'])\n",
    "\n",
    "import csv\n",
    "from itertools import zip_longest\n",
    "d = [gen, nevals, imm, maxi, avg, mini, std]\n",
    "export_data = zip_longest(*d, fillvalue = '')\n",
    "with open('evolution_automatic_results.csv', 'w', encoding=\"ISO-8859-1\", newline='') as myfile:\n",
    "    wr = csv.writer(myfile)\n",
    "    wr.writerow((\"gen\", \"nevals\", \"immigrants\",\"max\", \"avg\", \"min\", \"std\"))\n",
    "    wr.writerows(export_data)\n",
    "myfile.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## plot results\n",
    "\n",
    "# set upper and lower bound\n",
    "se = ([x/np.sqrt(15) for x in std])\n",
    "\n",
    "lower = [a_i - b_i for a_i, b_i in zip(avg, se)]\n",
    "upper = [a_i + b_i for a_i, b_i in zip(avg, se)]\n",
    "\n",
    "\n",
    "fig, ax = plt.subplots()\n",
    "ax.plot(gen, avg, color = 'blue', label = 'mean')\n",
    "ax.fill_between(gen, lower, upper, alpha=0.2)\n",
    "ax.plot(gen, maxi, color = 'darkblue', label = 'max', linestyle='dashed', linewidth = 0.5)\n",
    "ax.plot(gen, mini, color = 'lightblue', label = 'min', linestyle='dashed', linewidth = 0.5)\n",
    "\n",
    "ax.set_ylabel('Fitness')\n",
    "ax.set_xlabel(\"Generation\")\n",
    "ax.legend(loc = 'upper left')\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Visualize Evolution\n",
    "fig = plt.figure(figsize=(8,6))\n",
    "plt.axis(\"off\")\n",
    "ims = [[plt.imshow(np.transpose(i,(1,2,0)), animated=True)] for i in img_list]\n",
    "ani = animation.ArtistAnimation(fig, ims, interval=1000, repeat_delay=1000, blit=True)\n",
    "\n",
    "HTML(ani.to_jshtml())"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Thesis",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}