-
Notifications
You must be signed in to change notification settings - Fork 3
/
evo_train.py
executable file
·158 lines (126 loc) · 4.66 KB
/
evo_train.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
#!/usr/bin/env python3
import os
import sys
import multiprocessing
import numpy as np
import logging
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
from typing import List, Tuple
from evolution.dna import DNA
from evolution.cell_dna import DNAProperties
from gan_train import train_gan
evo_train_logger = logging.getLogger("evo_train")
def init_logger():
""" Initalize evo_train_logger """
evo_train_logger.setLevel(logging.INFO)
fh = logging.FileHandler('evo_train.log')
fh.setLevel(logging.INFO)
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)
# add the handlers to the logger
evo_train_logger.addHandler(fh)
evo_train_logger.addHandler(ch)
def generate_new_dna(n_dna:int, properties: DNAProperties) -> List[DNA]:
dna_list = []
for i in range(n_dna):
dna_list.append(DNA.gen_random())
return dna_list
def output_dna(dna_list: List[DNA], scores:List[float]) -> None:
""" Print serialized form of a list of DNAs
with their respective inception scores
"""
assert(len(dna_list) == len(scores))
for idx, d in enumerate(dna_list):
print(f"{d.serialize()} : {scores[idx]}")
evo_train_logger.info(f"{d.serialize()} : {scores[idx]}")
def score_dna(dna: DNA) -> float:
""" Create a GAN using the DNA,
train the GAN and return the inception score.
Training uses train_derived from AutoGAN
"""
reward = train_gan(to_arch(dna), 1)
return reward
def scoring_step(dna_list: List[DNA]) -> List[float]:
""" Score each DNA """
# scores = []
# with ThreadPoolExecutor(max_workers=min(multiprocessing.cpu_count(), 4)) as executor:
# futures = []
# for d in dna_list:
# futures.append(executor.submit(score_dna, d))
# futures = wait(futures, timeout=None, return_when=ALL_COMPLETED)[0]
# for f in futures:
# scores.append(f.result())
scores = []
for d in dna_list:
print(to_arch(d))
scores.append(score_dna(d))
return scores
def generation_step(dna_list: List[DNA], scores: List[List[float]], properties: DNAProperties) -> List[DNA]:
""" Set new properties (usually mutation probability)
Evolve and mutate a list of DNAs and return the list
"""
for d in dna_list:
d.set_properties(properties)
# Evoluton:
evo_matrix = generate_evolution_matrix(dna_list, scores)
for d in dna_list:
d.evolve(evo_matrix)
# Mutation
for d in dna_list:
d.mutate()
return dna_list
def to_arch(dna: DNA) -> List[int]:
""" Convert DNA to architecture string
expected by AutoGAN trainer
"""
values = dna.serialize()
skip = values[len(values)-2:]
skip = skip[0] * 2 + skip[1]
values = values[:len(values)-2]
values.append(skip)
return values
def generate_evolution_matrix(dna_list: List[DNA], scores: List[int]) -> List[List[float]]:
""" Generate a probability distribution of evolutopn parameters
based on inception scores from the scoring step
"""
param_options = []
evo_matrix = []
for c in dna_list[0].cells:
for idx, _ in enumerate(c.serialize()):
param_options.append(len(c.parameters.get_field_options(idx)))
for i in range(len(dna_list[0].serialize())):
p = [0 for _ in range(param_options[i])]
for d_idx, d in enumerate(dna_list):
p[d.serialize()[i]] += np.exp(scores[d_idx])
p_sum = sum(p)
p = [x / p_sum for x in p]
evo_matrix.append(p)
return evo_matrix
def main():
# Initialize mutation probability
mut_prob = 0.10
properties = DNAProperties(mutation_probability=mut_prob)
# Initialize dna uniformly
dna_list = generate_new_dna(5, properties)
n_epochs = 10
for epoch in range(n_epochs):
properties = DNAProperties(mutation_probability=mut_prob)
print(f"\n EPOCH: {epoch}\n")
evo_train_logger.info(f"\n EPOCH: {epoch}\n")
inception_scores = scoring_step(dna_list)
output_dna(dna_list, inception_scores)
dna_list = generation_step(dna_list, inception_scores, properties)
mut_prob /= 3
final_dna_list = dna_list
final_scores = scoring_step(final_dna_list)
print("\n FINAL: \n")
evo_train_logger.info("\n FINAL: \n")
output_dna(final_dna_list, final_scores)
if __name__ == "__main__":
init_logger()
main()