新颖性搜索（Novelty Search，NS）算法实践——利用NS算法解决迷宫导航问题

新颖性搜索（Novelty Search，NS）算法介绍
NS实现基础
- NovItem
- NoveltyArchive
利用新颖性度量的适应度函数
- 新颖性得分
- 新颖性指标
- 适应度函数
- 种群适应度评价函数
- 个体适应度评价函数
迷宫导航模拟
- 迷宫环境与迷宫求解器智能体介绍
- 超参数选择
- 实验运行函数
- 运行实验
- 智能体记录可视化
完整代码

新颖性搜索（Novelty Search，NS）算法介绍

NS算法介绍详见链接。

NS实现基础

要实现新颖性搜索（Novelty Search，NS），应包括保存有关所探查的新颖项信息的数据结构，以及用于维护和管理新颖项列表的结构。利用以下三个Python类进行封装：

NoveltyItem：其中包含有关在进化过程中评估的个体新颖性得分的所有相关信息。
NoveltyArchive：维护相关NoveltyItem实例列表的类。它提供了与已收集的NoveltyItem实例和当前种群相比评估个体基因组的新颖性得分的方法。
ItemsDistance：辅助结构，用于保存两个NoveltyItem实例之间的距离（新颖性）度量值。它可用于计算平均k最近邻距离，将其用作新颖性得分值。

NovItem

此类是保存有关在进化过程中评估的每个个体的新颖性得分信息的主要结构。它具有几个存储相关信息的字段：

def __init__(self, generation=-1, genomeId=-1, fitness=-1, novelty=-1):# 创建实例的代际的IDself.generation = generation# 被评估基因组的IDself.genomeId = genomeId# 所评估基因组的面向目标的适应性得分（与迷宫出口的接近度）self.fitness = fitness# 评估的基因组的新颖性评分self.novelty = noveltyself.in_archive = False# 数据点的列表，这些数据点表示迷宫求解器智能体在模拟过程中访问的特定迷宫位置的坐标。# 该数据列表用于估计当前新奇项和其他新奇项之间的距离。self.data = []

NoveltyArchive

该类维护了一系列相关的新奇项，并提供了评估单个基因组以及整个基因组总体的新颖性评分的方法。它在构造函数中定义了以下字段：

def __init__(self, threshold, metric):# 用于估计新颖性度量的函数引用self.novelty_metric = metric# 添加到此存档中的NoveltyItem的当前最小新奇得分self.novelty_threshold = threshold# novelty_threshold的最小可能值self.novelty_floor = 0.25# 用于novelty_threshold值的动态更改self.items_added_in_generation = 0self.time_out = 0# k个最近邻居的默认数量，可用于新奇评分估计self.neighbors = KNNNoveltyScore# 当前的进化世代self.generation = 0# 当前收集的所有相关NoveltyItem实例的列表self.novel_items = []# 所有对象中具有最大的面向目标适应性得分的新颖项目的列表self.fittest_items = []

novelty_threshold字段的动态变化取决于以下源代码：

    def _adjust_archive_settings(self):"""The function to adjust the dynamic novelty threshold depending on how many have NoveltyItem objects have been added to the archive recently"""if self.items_added_in_generation == 0:self.time_out += 1else:self.time_out = 0# 如果在10代中未找到新的路径，则将novelty_threshold值降低5％if self.time_out >= 10:self.novelty_threshold *= 0.95if self.novelty_threshold < self.novelty_floor:self.novelty_threshold = self.novelty_floorself.time_out = 0# 如果在上一代中添加了四个以上的新奇项，则将novelty_threshold值提高20％if self.items_added_in_generation >= 4:self.novelty_threshold *= 1.2# reset countersself.items_added_in_generation = 0

在每代进化完成后调用此函数，以调整下一代的novelty_threshold字段值。该值确定了下一代应向archive中添加多少个新颖性项，以适应随着时间的推移使用NS方法查找新颖解的难度。
以下源代码显示了如何使用novelty_threshold值来确定要添加哪个NoveltyItem：

    def evaluate_individual_novelty(self, genome, genomes, n_items_map, only_fitness=False):if genome.key not in n_items_map:print("WARNING! Found Genome without novelty point associated: %s" +"\nNovelty evaluation will be skipped for it. Probably winner found!" % genome.key)returnitem = n_items_map[genome.key]if item.fitness == -1.0:return -1.0result = 0.0if only_fitness:result = self._novelty_avg_knn(item=item, genomes=genomes, n_items_map=n_items_map)else:result = self._novelty_avg_knn(item=item, neighbors=1, n_items_map=n_items_map)if result > self.novelty_threshold or len(self.novel_items) < ArchiveSeedAmount:self._add_novelty_item(item)item.novelty = resultitem.generation = self.generationreturn result

前面的代码使用函数来评估新颖性得分，以估计所提供基因组的新颖性。如果在更新存档模式下调用此函数(only_fitness = False)，则将获得的新奇得分（result）与novelty_threshold字段的当前值进行比较。根据比较结果，是否将NoveltyItem对象添加到NoveltyArchive对象。此外，引入了ArchiveSeedAmount常量，以在存档仍然为空时在演化开始时使用NoveltyItem实例对存档进行初始化。

利用新颖性度量的适应度函数

新颖性得分

智能体的行为空间由运行迷宫求解模拟时通过迷宫的轨迹定义。因此，任何行为空间访问点密集的区域都不那么新颖，对求解器代理程序的奖励也更少。
最简单的衡量点稀疏度的方法是从它到k个近邻的平均距离。稀疏区域具有较高的距离值，而较稠密的区域具有较低的距离值。以下公式给出了行为空间点的稀疏性：
ρ=1k∑i=0kdist(x,ui)\rho = \frac 1k\sum_{i=0}^kdist(x,u_i)ρ=k1i=0∑kdist(x,ui)
其中，uiu_iui是距离（新颖性）dist(x,y)dist(x,y)dist(x,y)度量计算的第i个最接近的邻居。
通过上述公式在行为空间中特定点的稀疏度计算出的新颖性分数可以由适应度函数使用。
以下函数定义了用于查找新颖性分数的Python代码：

    def _novelty_avg_knn(self, item, n_items_map, genomes=None, neighbors=None):distances = None# 判断是否包含当前种群中所有基因组的列表，若包含，则首先填充种群中所有基因组的行为特征之间的距离列表，包括来自NoveltyArchive的所有NoveltyItem对象。否则，将使用提供的新颖项（item）从NoveltyArchive查找其与所有NoveltyItem对象之间的距离if genomes is not None:distances = self._map_novelty_in_population(item=item, genomes=genomes, n_items_map=n_items_map)else:distances = self._map_novelty(item=item)# 按照从小到大的顺序对距离列表进行排序distances.sort()# 初始化计算k最近邻居分数所需的所有中间变量，并测试上一步中收集的距离值的数量是否大于ArchiveSeedAmount常数值if neighbors is None:neighbors = self.neighborsdensity, weight, distance_sum = 0.0, 0.0, 0.0length = len(distances)# 检查找到的距离列表的长度是否小于针对其进行测试的邻居的数量（neighbors）if length >= ArchiveSeedAmount:length = neighborsif len(distances) < length:# the number of mapped distances is less than number of neighborslength = len(distances)# 循环获取所有距离和权重之和i = 0while weight < float(neighbors) and i < length:distance_sum += distances[i].distanceweight += 1.0i += 1# 由于计算出的权重值超过了指定的邻居数而导致前一个循环退出时，或者如果已经对distances列表中的所有距离值进行了迭代，则可以将给定项目的新颖性得分计算为平均距离到k个最近的邻居if weight > 0:density = distance_sum / weightreturn density

新颖性指标

新颖性指标是衡量当前解决方案与已知解决方案有多不同的一种度量。当估计从行为空间中的当前点到它的k个最近邻居的距离时，它用于计算新奇分数。
在实验中，通过两个轨迹向量（每个智能体一个向量）之间的逐项距离来确定测量行为差异的新颖性度量，轨迹矢量包含迷宫导航代理在仿真过程中访问的位置的坐标：
dist(x,μ)=1n∑j=0n∣xj−uj∣dist(x,\mu)=\frac 1n\sum_{j=0}^n|x_j-u_j|dist(x,μ)=n1j=0∑n∣xj−uj∣
其中n是轨迹向量μj\mu_jμj和xjx_jxj的大小。
在迷宫导航实验中，主要对智能体的最终位置感兴趣。因此，最终，轨迹矢量可能仅包含智能体的最终坐标。
用于新颖性度量值估计的Python代码如下：

def maze_novelty_metric(first_item, second_item):diff_accum = 0.0size = len(first_item.data)for i in range(size):diff = abs(first_item.data[i] - second_item.data[i])diff_accum += diff
return diff_accum / float(size)

适应度函数

对于不同任务，使用各种适合度因子：

新颖性分数用于指导神经进化过程。它被指定为每个基因组的适应度值，并在进化世代中用于基因组评估。
从迷宫模拟器获得的面向目标的适应度得分，用于测试是否已实现最终目标（即，已找到迷宫出口），记录此值以高效评估每个智能体。

适应度值评估的源代码通过两个函数提供：

回调函数，用于评估整个种群的适应性得分（eval_genomes）
通过迷宫求解过程评估单个基因组的函数（eval_individual）

种群适应度评价函数

适应度评估函数是在NEAT-Python库中注册的回调函数，允许该库针对需要解决的特定任务的特定条件对种群基因组进行评估。
NEAT-Python库不允许通过回调函数传递有关任务完成的任何信号，除非通过指定获胜者基因组的特定适应性得分值。该适应性值必须高于NEAT-Python超参数配置中的适应性阈值。但是，使用NS算法，不可能准确估计获胜者基因组可以达到的新奇分数的上限。此外，优胜者基因组的新颖性得分值可以低于在进化过程中较早时基因组获得的值。
因此，鉴于将新奇评分指定为基因组的适应度值，需要一种解决方法，以使用由NEAT-Python库定义的标准终止条件。通过使用一个特定的评分值来做到这一点，此值确定通过NEAT-Python超参数配置提供的终止条件。使用800000作为新奇得分的指示性度量，并将其自然对数（约13.59）作为适应度阈值。
该函数的完整源代码如下：

def eval_genomes(genomes, config):# 创建字典以存储种群中每个基因组的评估新颖性项（n_items_map）n_items_map = {}solver_genome = None# 遍历种群中的所有基因组，评估其迷宫求解性能for genome_id, genome in genomes:found = eval_individual(genome_id=genome_id,genome=genome,genomes=genomes,n_items_map=n_items_map,config=config)if found:solver_genome = genometrial_sim.archive.end_of_generation()# 遍历种群中的所有基因组，以使用估计的新颖性分数为基因组分配适应性分数。# 在迷宫求解模拟过程中，新颖性分数估算过程使用n_items_map中收集的NoveltyItem对象：for genome_id, genome in genomes:fitness = trial_sim.archive.evaluate_individual_novelty(genome=genome,genomes=genomes,n_items_map=n_items_map,only_fitness=True)if fitness > 1:fitness = math.log(fitness)else:fitness = 0genome.fitness = fitness# 如果找到了成功的求解器基因组，为其分配的适应度值等于之前所述的指示性适应度得分if solver_genome is not None:solver_genome.fitness = math.log(800000) # ~=13.59

个体适应度评价函数

此函数是种群适应性评估的重要组成部分，可以从前面讨论的eval_genomes函数中调用该功能，以评估种群中每个基因组的迷宫解决性能。
通过迷宫导航模拟对单个基因组作为迷宫导航器的评估如下：

def eval_individual(genome_id, genome, genomes, n_items_map, config):# 创建NoveltyItem对象，以保存有关与特定基因组相关的新颖性评分的信息，# 并将其保存在n_items_map字典的全基因组ID下n_item = archive.NoveltyItem(generation=trial_sim.population.generation,genomeId=genome_id)n_items_map[genome_id] = n_item# 创建原始迷宫环境的副本，以避免在仿真过程中产生副作用，然后从提供的基因组创建ANNmaze_env = copy.deepcopy(trial_sim.orig_maze_environment)control_net = neat.nn.FeedForwardNetwork.create(genome, config)# 使用迷宫环境和ANN的副本，针对给定数量的模拟步骤执行迷宫求解模拟goal_fitness = maze.maze_simulation_evaluate(env=maze_env, net=control_net, time_steps=SOLVER_TIME_STEPS,n_item=n_item,mcns=MCNS)if goal_fitness == -1:# The individual doesn't meet the minimal fitness criterionprint("Individ with ID: %d marked for extiction, MCNS: %f" % (genome_id, MCNS))return False# 返回的基于目标的适应度评分以及其他基因组参数存储在AgentRecord中，然后将其添加到记录存储中record = agent.AgentRecord(generation=trial_sim.population.generation,agent_id=genome_id)record.fitness = goal_fitnessrecord.x = maze_env.agent.location.xrecord.y = maze_env.agent.location.yrecord.hit_exit = maze_env.exit_foundrecord.species_id = trial_sim.population.species.get_species_id(genome_id)record.species_age = record.generation - \trial_sim.population.species.get_species(genome_id).created# add record to the storetrial_sim.record_store.add_record(record)# 如果给定的基因组不是最终结果，估计其新奇分数，# 用当前基因组的NoveltyItem更新NoveltyArchive中高适应度值的基因组列表：if not maze_env.exit_found:# evaluate genome novelty and add it to the archive if appropriaterecord.novelty = trial_sim.archive.evaluate_individual_novelty(
genome=genome, genomes=genomes,n_items_map=n_items_map)# update fittest organisms listtrial_sim.archive.update_fittest_with_genome(genome=genome, n_items_map=n_items_map)return maze_env.exit_found

实验中，基因组的适应度评分定义为两个单独的值，每个值具有不同的用途。面向目标的适应性评分有助于测试是否已找到解，并收集统计信息。基于新颖性的适应度评分指导神经进化过程。

迷宫导航模拟

迷宫环境与迷宫求解器智能体介绍

该部分详细介绍参考链接。

超参数选择

实验中使用的目标函数基于没有明确上限值的新颖性度量。因此，不能精确地估计适应度阈值，为了表明找到了胜出的解决方案，使用一个指示值。
选择800000作为指示性新颖性得分。但是，为了在绘制实验结果时保持适应度得分的直观呈现，使用自然对数缩小了求解器的新颖性得分。因此，配置文件中使用的适应性阈值变为13.5，这比最大可能适应度分数（13.59）小一点，以避免出现舍入浮点数的问题：

[NEAT]
fitness_criterion = max
fitness_threshold = 13.5
pop_size = 500
reset_on_extinction = False

物种在进化停滞时的的生存时间更长：

[DefaultStagnation]
max_stagnation = 100

实验运行函数

加载NEAT算法配置并创建初始的基因组种群：

config = neat.Config(neat.DefaultGenome,neat.DefaultReproduction,neat.DefaultSpeciesSet,neat.DefaultStagnation,config_file)
p = neat.Population(config)

为了在每次评估之后保留中间结果，使用MazeSimulationTrial对象初始化trial_sim全局变量。
使用一个全局变量，可以通过传递给NEAT-Python框架的适应度评估回调函数(eval_genomes(genomes,config))进行访问：
```
global trial_sim
trial_sim = MazeSimulationTrial(maze_env=maze_env,population=p,archive=novelty_archive)
```

同样，向Population对象注册数个报告者，以输出算法结果并收集统计信息：

p.add_reporter(neat.StdOutReporter(True))
stats = neat.StatisticsReporter()
p.add_reporter(stats)

在指定的世代中运行NEAT算法并评估结果：

start_time = time.time()
best_genome = p.run(eval_genomes,n=n_generations)
elapsed_time = time.time() - start_time
# Display the best genome among generations.
print('\nBest genome: \n%s' % (best_genome))
solution_found = (best_genome.fitness >= config.fitness_threshold)
if solution_found:print("SUCCESS: The stable maze solver controller was found!!!")
else:print("FAILURE: Failed to find the stable maze solver controller!!!")

之后，可以将收集到的统计数据和新颖性档案记录可视化并保存到文件系统中：

node_names = {-1:'RF_R',-2:'RF_RF',-3:'RF_F',-4:'RF_FL',-5:'RF_L',-6:'RF_B',-7:'RAD_F',-8:'RAD_L',-9:'RAD_B',-10:'RAD_R',0:'ANG_VEL',1:'VEL'
}
visualize.draw_net(config,best_genome,view=show_results,node_names=node_names,directory=trial_out_dir,fmt='svg')
if args is None:visualize.draw_maze_records(maze_env,trial_sim.record_store.records,view=show_results)
else:visualize.draw_maze_records(maze_env,trial_sim.record_store.records,view=show_results,width=args.width,height=args.height,filename=os.path.join(trial_out_dir,'maze_records.svg'))
visualize.plot_stats(stats,ylog=False,filename=os.path.join(trial_out_dir,'avg_fitness.svg'))
visualize.plot_species(stats,view=show_results,filename=os.path.join(trial_out_dir,'speciation.svg'))
# store NoveltyItems archive data
trial_sim.archive.write_fittest_to_file(path=os.path.join(trial_out_dir,'ns_items_fittest.txt'))
trial_sim.archive.write_to_file(path=os.path.join(trial_out_dir,'ns_items_all.txt')
)

最后，执行其他可视化例程，以可视化迷宫求解器智能体通过迷宫的路径。
为此，对进化过程中发现的最佳智能体的决策器ANN进行了迷宫导航仿真。在此模拟运行期间，将收集智能体访问的所有路径点，以供通过draw_agent_path函数进行渲染：

maze_env = copy.deepcopy(trial_sim.orig_maze_environment)
control_net = neat.nn.FeedForwardNetwork.create(best_genome,config)
path_points = []
evaluate_fitness = maze.maze_simulation_evaluate(env=maze_env,net=control_net,time_steps=SOLVER_TIME_STEPS,path_points=path_points)
print("Evaluated fitness of best agent: %f" % evaluate_fitness)
visualize.draw_agent_path(trial_sim.orig_maze_environment,path_points,best_genome,view=show_results,width=args.width,height=args.height,filename=os.path.join(trial_out_dir,'best_solver_path.svg'))

运行实验

在终端中执行以下命令：

$ python3 maze_experiment.py -g 500 -t 10 -m medium --width 300 --height 150

显示获胜者基因组的配置和有关试验的一般统计数据：

Best genome:
Key: 36170
Fitness: 13.592367006650065
Nodes:0 DefaultNodeGene(key=0, bias=0.9583749969785536, response=1.0, activation=sigmoid, aggregation=sum)1 DefaultNodeGene(key=1, bias=-1.3352111211865185, response=1.0, activation=sigmoid, aggregation=sum)
Connections:DefaultConnectionGene(key=(-10, 0), weight=2.012697148953962, enabled=True)DefaultConnectionGene(key=(-10, 1), weight=2.3586743900645715, enabled=True)DefaultConnectionGene(key=(-9, 0), weight=0.5133819837545476, enabled=False)DefaultConnectionGene(key=(-9, 1), weight=-1.3453064468779043, enabled=True)DefaultConnectionGene(key=(-8, 0), weight=-1.3151248904230235, enabled=True)DefaultConnectionGene(key=(-6, 1), weight=-1.50551995321142, enabled=True)DefaultConnectionGene(key=(-5, 0), weight=-3.020445866909174, enabled=False)DefaultConnectionGene(key=(-5, 1), weight=-2.090540743662507, enabled=True)DefaultConnectionGene(key=(-4, 0), weight=-1.8754146567384993, enabled=True)DefaultConnectionGene(key=(-4, 1), weight=2.0773106904549614, enabled=True)DefaultConnectionGene(key=(-3, 0), weight=2.6406887829938044, enabled=True)DefaultConnectionGene(key=(-3, 1), weight=0.4049529471735065, enabled=True)DefaultConnectionGene(key=(-2, 1), weight=0.5571713919237005, enabled=True)DefaultConnectionGene(key=(-1, 0), weight=1.5212036155782374, enabled=True)DefaultConnectionGene(key=(-1, 1), weight=0.7204766260373855, enabled=True)DefaultConnectionGene(key=(0, 0), weight=1.1105019563826226, enabled=True)
SUCCESS: The stable maze solver controller was found!!!

控制器ANN的最终配置如下所示：
进化过程适应度得分曲线图：

进化过程中物种图：

智能体记录可视化

智能体记录的可视化：

$ python3 visualize.py -m medium -r out/maze_ns/medium/0/data.pickle --width 300 --height 150

查看成功的智能体的路径，该智能体能够找到迷宫出口：

完整代码

geometry.py、medium_maze.txt、utils.py
代码链接。
agent.py

import pickleclass Agent:"""This is maze navigating agent."""def __init__(self,location,heading=0,speed=0,angular_vel=0,radius=8.0,range_finder_range=100.0):"""Creates new Agent with spcified parameters.Arguments:location: The agent initial position within maze.heading: The heading direction in degrees.speed: The linear velocity of the agent.angular_vel: The angular velocity of the agent.radius: The agent's body radius.range_finder_range: The maximal detection range for radar sensor."""self.heading = headingself.speed = speedself.angular_vel = angular_velself.radius = radiusself.range_finder_range = range_finder_rangeself.location = location# defining the range finder sensorsself.range_finder_angles = [-90.0, -45.0, 0.0, 45.0, 90.0, -180.0]# defining the range radar sensorsself.radar_angles = [(315.0, 405.0), (45.0, 135.0), (135.0, 225.0), (225.0, 315.0)]# the list to hold range finders activationsself.range_finders = [None] * len(self.range_finder_angles)# the list to hold pie-slice radar activationsself.radar = [None] * len(self.radar_angles)class AgentRecord:"""The class to hold results of maze navigation simulation for specificsolver agent. It provides all statistics about the agent at the endof navigation run."""def __init__(self,generation,agent_id):"""Creates new record for specific agent at the specific generationof the evolutionary process."""self.generation = generationself.agent_id = agent_id# initialize agent's propertiesself.x = -1self.y = -1self.fitness = -1self.novelty = -1# The flag to indicate whether this agent was able to find maze exitself.hit_exit = False# The ID of species this agent belong toself.species_id = -1# The age of agent's species at the time of recordingself.species_age = -1class AgentRecordStore:"""The class to control agents record store."""def __init__(self):"""Creates new instance."""self.records = []def add_record(self,record):"""The function to add specified record to this store.Arguments:record: The record to be added."""self.records.append(record)def load(self, file):"""The function to load records list from the specied file into this class.Arguments:file: The path to the file to read agents records from."""with open(file, 'rb') as dump_file:self.records = pickle.load(dump_file)def dump(self, file):"""The function to dump records list to the specified file from this class.Arguments:file: The path to the file to hold data dump."""with open(file, 'wb') as dump_file:pickle.dump(self.records, dump_file)

maze_config.ini

[NEAT]
fitness_criterion = max
fitness_threshold = 13.5
pop_size = 500
reset_on_extinction = True[DefaultGenome]
# node activataion options
activation_default = sigmoid
activation_mutate_rate = 0.0
activation_options = sigmoid# node aggregation options
aggregation_default = sum
aggregation_mutate_rate = 0.0
aggregation_options = sum# node bias options
bias_init_mean = 0.0
bias_init_stdev = 1.0
bias_max_value = 30.0
bias_min_value = -30.0
bias_mutate_power = 0.5
bias_mutate_rate = 0.7
bias_replace_rate = 0.1# genome compatibility options
compatibility_disjoint_coefficient = 1.1
compatibility_weight_coefficient = 0.5# connection add/reomve rates
conn_add_prob = 0.5
conn_delete_prob = 0.1# connection enable options
enabled_default = True
enabled_mutate_rate = 0.01feed_forward = False
initial_connection = partial_direct 0.5# node add/reomve rates
node_add_prob = 0.1
node_delete_prob = 0.1# network parameters
num_hidden = 1
num_inputs = 10
num_outputs = 2# node responde options
response_init_mean = 1.0
response_init_stdev = 0.0
response_max_value = 30.0
response_min_value = -30.0
response_mutate_power = 0.0
response_mutate_rate = 0.0
response_replace_rate = 0.0# connection weight options
weight_init_mean = 0.0
weight_init_stdev = 1.0
weight_max_value = 30.0
weight_min_value = -30.0
weight_mutate_power = 0.5
weight_mutate_rate = 0.8
weight_replace_rate = 0.1[DefaultSpeciesSet]
compatibility_threshold = 3.0[DefaultStagnation]
species_fitness_func = max
max_stagnation = 100
species_elitism = 1[DefaultReproduction]
elitism = 2
survival_threshold = 0.1
min_species_size = 2

maze_environment.py

import mathimport agent
import geometryfrom novelty_archive import NoveltyItem# The maximal allowed speed for the maze solver agent
MAX_AGENT_SPEED = 3.0def maze_novelty_metric(first_item, second_item):"""The function to calculate the novelty metric score as a distancebetween two data vectors in provided NoveltyItemsArguments:first_item: The first NoveltyItemsecond_item: The second NoveltyItemReturns:The novelty metric as a distance between two data vectorsin provided noveltyItems"""if not (hasattr(first_item, 'data') or hasattr(second_item, 'data')):return NotImplementedif len(first_item.data) != len(second_item.data):# can not be comparedreturn 0.0diff_accum = 0.0size = len(first_item.data)for i in range(size):diff = abs(first_item.data[i] - second_item.data[i])diff_accum += diffreturn diff_accum / float(size)def maze_novelty_metric_euclidean(first_item, second_item):"""The function to calculate the novelty metric score as a distancebetween two data vectors in provided NoveltyItems.Arguments:first_item: The first NoveltyItemsecond_item: The second NoveltyItemReturns:The novelty metric as a distance between two data vectorsin provided noveltyItems"""if not (hasattr(first_item, 'data') or hasattr(second_item, 'data')):return NotImplementedif len(first_item.data) != len(second_item.data):# can't be comparedreturn 0.0diff_accum = 0.0size = len(first_item.data)for i in range(size):diff = (first_item.data[i] - second_item.data[i])diff_accum += (diff * diff)return math.sqrt(diff_accum)class MazeEnvironment:"""This class encapsulates the maze simulation environment."""def __init__(self, agent, walls, exit_point, exit_range=5.0):"""Creates new maze environment with specified walls and exit point.Arguments:agent: The maze novigating agentwalls: The maze wallsexit_point: The maze exit pointexit_range: The range arround exit point marking exit area"""self.walls = wallsself.exit_point = exit_pointself.exit_range = exit_range# The maze navigating agentself.agent = agent# The flag to indicate if exit was foundself.exit_found = False# The initial distance of agent from exitself.initial_distance = self.agent_distance_to_exit()# The sample rate of agent position points saving during simulation steps.self.location_sample_rate = -1# Update sensorsself.update_rangefinder_sensors()self.update_radars()def agent_distance_to_exit(self):"""The function to estimate distance from maze solver agent to the maze exit.Returns:The distance from maze solver agent to the maze exit"""return self.agent.location.distance(self.exit_point)def test_wall_collision(self,loc):"""The function to test if agent at specified location collides with any ofthe maze wallsArgument:loc: The new agent location to test for collision.Returns:The True if agent at new location will collide with any of the maze walls."""for w in self.walls:if w.distance(loc) < self.agent.radius:return Truereturn Falsedef create_net_inputs(self):"""The function to create the ANN input values from the simulaiton environment.Returns:The list of ANN inputs consist of values get from solver agent sensors."""inputs = []# The range findersfor ri in self.agent.range_finders:inputs.append(ri)# The radar sensors:for rs in self.agent.radar:inputs.append(rs)return inputsdef apply_control_signals(self, control_signals):"""The function to apply control signals received from control ANN to themaze solver agent.Arguments:control_signals: The control received from the control ANN."""self.agent.angular_vel += (control_signals[0] - 0.5)self.agent.speed += (control_signals[1] - 0.5)# constrain the speed & angular velocityif self.agent.speed > MAX_AGENT_SPEED:self.agent.speed = MAX_AGENT_SPEEDif self.agent.speed < -MAX_AGENT_SPEED:self.agent.speed = -MAX_AGENT_SPEEDif self.agent.angular_vel > MAX_AGENT_SPEED:self.agent.angular_vel = MAX_AGENT_SPEEDif self.agent.angular_vel < -MAX_AGENT_SPEED:self.agent.angular_vel = -MAX_AGENT_SPEEDdef update_rangefinder_sensors(self):"""The function to update the agent range finder sensors."""for i, angle in enumerate(self.agent.range_finder_angles):rad = geometry.deg_to_rad(angle)# project a point from agent location outwardsprojection_point = geometry.Point(x = self.agent.location.x + math.cos(rad) * self.agent.range_finder_range,y = self.agent.location.y + math.sin(rad) * self.agent.range_finder_range)# rotate the projection point by the agent's heading angle to# algin it with heading direction.projection_point.rotate(self.agent.heading, self.agent.location)# create the line segment from the agent location to the projected pointprojection_line = geometry.Line(a = self.agent.location,b = projection_point)# set range to maximum detection rangemin_range = self.agent.range_finder_range# now test against maze walls to see if projection line hits any wall# and find the closest hitfor wall in self.walls:found, intersection = wall.intersection(projection_line)if found:found_range = intersection.distance(self.agent.location)# we are interested in the closest hitif found_range < min_range:min_range = found_range# Update sendor valueself.agent.range_finders[i] = min_rangedef update_radars(self):"""The function to update the agent radar sensors."""target = geometry.Point(self.exit_point.x, self.exit_point.y)# rotate target with respect to the agent's heading to align it # with haeading diretiontarget.rotate(self.agent.heading, self.agent.location)# translate with respect to the agent's locationtarget.x -= self.agent.location.xtarget.y -= self.agent.location.y# the angle between maze eixt point and the agent's heading directionangle = target.angle()# find the appropriate radar sensor to be firedfor i, r_angles in enumerate(self.agent.radar_angles):self.agent.radar[i] = 0.0 # reset specific radarif (angle >= r_angles[0] and angle < r_angles[1]) or (angle + 360 >= r_angles[0] and angle + 360 < r_angles[1]):self.agent.radar[i] = 1.0 # fire teh radardef update(self,control_signals):"""The function to update the solver agent position within maze. After agent position updated it will be checked to find out if maze exit wasreached after that.Arguments:control_signals: The control signals received from control ANNReturns:The True if maze exit was found after update or maze exit was alreadyfound in previous simulation cycles."""if self.exit_found:return True# Apply control signalsself.apply_control_signals(control_signals)# get X and Y velocity conponentsvx = math.cos(geometry.deg_to_rad(self.agent.heading)) * self.agent.speedvy = math.sin(geometry.deg_to_rad(self.agent.heading)) * self.agent.speed# Update current Agent's heading (we consider the simulation time # step size equal to 1s and the angular velocity as degrees# per second)self.agent.heading += self.agent.angular_vel# Enforce angular velocity bounds by wrappingif self.agent.heading > 360:self.agent.heading -= 360elif self.agent.heading < 0:self.agent.heading += 360# find the next loaction of the agentnew_loc = geometry.Point(x = self.agent.location.x + vx,y = self.agent.location.y + vy)if not self.test_wall_collision(new_loc):self.agent.location = new_loc# update agent's sensorsself.update_rangefinder_sensors()self.update_radars()# check if agent reached exit pointdistance = self.agent_distance_to_exit()self.exit_found = (distance < self.exit_range)return self.exit_founddef __str__(self):"""Returns the nicely formatted string representation of this environment."""str = "MAZE\nAgent at: (%.1f, %.1f)" % (self.agent.location.x, self.agent.location.y)str += "\nExit  at: (%.1f, %.1f), exit range: %.1f" % (self.exit_point.x, self.exit_point.y, self.exit_range)str += "\nWalls [%d]" % len(self.walls)for w in self.walls:str += "\n\t%s" % wreturn strdef read_environment(file_path):"""The function to read maze environment configuration from provided file.Argument:file_path: The path to the file read maze configuration from.Returns:The initialized maze environment."""num_lines, index = -1, 0walls = []maze_agent, maze_exit = None, Nonewith open(file_path, 'r') as f:for line in f.readlines():line = line.strip()if len(line) == 0:# skip empty linescontinueif index == 0:# read the number of lines segmentsnum_lines = int(line)elif index == 1:# read the agent's positionloc = geometry.read_point(line)maze_agent = agent.Agent(location=loc)elif index == 2:# read the agent's initial headingmaze_agent.heading = float(line)elif index == 3:# read the maze exit locationmaze_exit = geometry.read_point(line)else:# read the wallswall = geometry.read_line(line)walls.append(wall)# increment cursorindex += 1assert len(walls) == num_linesprint("Maze environment configured successfully from the file: %s" % file_path)# create and return the maze environmentreturn MazeEnvironment(agent=maze_agent, walls=walls, exit_point=maze_exit)def maze_simulation_evaluate(env, net, time_steps,mcns=0.0, n_item=None,path_points=None):"""The function to evaluate maze simulation for specific environmentand controller ANN provided. The results will be saved into providedagent record holder.Arguments:env: The maze configuration environment.net: The maze solver agent's controll ANN.time_steps: The number of time steps for maze simulation.mcns: The minimal criteria fitness value.n_item: The NovelryItem to store evaluation results.path_points: The holder for path points collected during simulation.If provided None the nothing will be collected.Returns:The goal-oriented fitness value, i.e., how close is agent to the exitat the end of simulation."""exit_found = Falsefor i in range(time_steps):if maze_simulation_step(env, net):print("Maze solved in %d steps" % (i + 1))exit_found = Truebreakif path_points is not None:# collect current positionpath_points.append(geometry.Point(env.agent.location.x, env.agent.location.y))# store agent path points at a given sample size rateif (time_steps - i) % env.location_sample_rate == 0 and n_item is not None:n_item.data.append(env.agent.location.x)n_item.data.append(env.agent.location.y)# store final agent coordinates as genome's novelty characteristicsif n_item is not None:n_item.data.append(env.agent.location.x)n_item.data.append(env.agent.location.y)# Calculate the fitness score based on distance from exitfitness = 0.0if exit_found:fitness = 1.0else:# Noralize distance to range (0,1]distance = env.agent_distance_to_exit()fitness = (env.initial_distance - distance) / env.initial_distanceif fitness <= 0:fitness = 0.01# Use minimal criteria fitness value to signal if genome# should be included into populationif fitness < mcns:fitness = -1 # mark genome to be excludedif n_item is not None:n_item.fitness = fitnessreturn fitnessdef maze_simulation_step(env, net):"""The function to perform one step of maze simulation.Arguments:env: The maze configuration environment.net: The maze solver agent's control ANN.Returns:The True if maze agent solved the maze."""# create inputs from the current state of the environmentinputs = env.create_net_inputs()# load inputs into controll ANN and get resultsoutput = net.activate(inputs)# apply control signal to the environment and updatereturn env.update(output)

maze_experiment.py

import os
import shutil
import math
import random
import time
import copy
import argparseimport neat
import visualize
import utilsimport maze_environment as maze
import agent
import novelty_archive as archive# The number of maze solving simulator steps
SOLVER_TIME_STEPS = 400
# The minimal goal fitness criterion
MCNS = 0.01class MazeSimulationTrial:"""The class to hold maze simulator execution parameters and results"""def __init__(self,maze_env,population,archive):"""Creates new instance and initialize fileds.Arguments:maze_env:   The maze environment as loaded from configuration file.population: The population for this trial runarchive:    The archive to hold NoveltyItems"""# The initial maze simulation environmentself.orig_maze_environment = maze_env# The record store for envaluated maze solver agentsself.record_store = agent.AgentRecordStore()# The NEAT population objectself.population = population# The NoveltyItem archiveself.archive = archive# The simulation results holder for a one trial.
# It must be initialized before start of each trial.
trial_sim = Nonedef eval_individual(genome_id, genome, genomes, n_items_map, config):"""Evaluates the individual represented by genome.Arguments:genome_id:      The ID of genome.genome:         The genome to evaluate.genomes:        The genomes population for current generation.n_items_map:    The map to hold novelty items for current generation.config:         The NEAT configuration holder.Return:The True if successful solver found."""# create NoveltyItem for genome and store it into mapn_item = archive.NoveltyItem(generation=trial_sim.population.generation,genomeId=genome_id)n_items_map[genome_id] = n_item# run the simulationmaze_env = copy.deepcopy(trial_sim.orig_maze_environment)control_net = neat.nn.FeedForwardNetwork.create(genome, config)goal_fitness = maze.maze_simulation_evaluate(env=maze_env,net=control_net,time_steps=SOLVER_TIME_STEPS,n_item=n_item,mcns=MCNS)if goal_fitness == -1:# The individual doesn't meet the minimal fitness criterionprint("Individ with ID: %d marked for extiction, MCNS: %f" % (genome_id, MCNS))return False# store simulation results into the agent recordrecord = agent.AgentRecord(generation=trial_sim.population.generation,agent_id=genome_id)record.fitness = goal_fitnessrecord.x = maze_env.agent.location.xrecord.y = maze_env.agent.location.yrecord.hit_exit = maze_env.exit_foundrecord.species_id = trial_sim.population.species.get_species_id(genome_id)record.species_age = record.generation - trial_sim.population.species.get_species(genome_id).created# add record to the storetrial_sim.record_store.add_record(record)# Evaluate the novelty of a genome and add the novelty item to the # archive of Novelty items if appropriateif not maze_env.exit_found:# enaluate genome novelty and add it to the archive if appropriaterecord.novelty = trial_sim.archive.evaluate_individual_novelty(genome=genome,genomes=genomes,n_items_map=n_items_map)# update fittest organisms listtrial_sim.archive.update_fittest_with_genome(genome=genome,n_items_map=n_items_map)return maze_env.exit_founddef eval_genomes(genomes,config):"""The function to evaluate the fitness of each genome in the genomes list.Arguments:genomes: The list of genomes from population in the current generationconfig:  The configuration settings with algorithmhyper-parameters"""# The map to hold the novelty items for current generationn_items_map = {}solver_genome = Nonefor genome_id, genome in genomes:found = eval_individual(genome_id=genome_id,genome=genome,genomes=genomes,n_items_map=n_items_map,config=config)if found:solver_genome = genome# now adjust the arcive settings and evaluate populationtrial_sim.archive.end_of_generation()for genome_id, genome in genomes:# set fitness value as a logarithm of a novelty score of a# genome in the populationfitness = trial_sim.archive.evaluate_individual_novelty(genome=genome,genomes=genomes,n_items_map=n_items_map,only_fitness=True)# To avoid negative genome fitness scores we just set to zero all obtained# fitness scores that is less than 1 (note we use the natural logarithm)if fitness > 1:fitness = math.log(fitness)else:fitness = 0# assign the adjusted fitness score to the genomegenome.fitness = fitness# if successful maze solver was found then adjust its fitness # to signal the finish evolutionif solver_genome is not None:solver_genome.fitness = math.log(800000) # ~=13.59def run_experiment(config_file,maze_env,novelty_archive,trial_out_dir,args=None,n_generations=100,save_results=False,silent=False):"""The function to run the experiment against hyper-parameters defined in the provided configuration file.The winner genome will be rendered as a graph as well as theimportant statistics of neuroevolution process execution.Arguments:config_file:        The path to the file with experiment configurationmaze_env:           The maze environment to use in simulation.novelty_archive:    The archive to work with NoveltyItems.trial_out_dir:      The directory to store outputs for this trialn_generations:      The number of generations to execute.save_results:       The flag to control if intermdiate results will be saved.silent:             If True than no intermediary outputs will bepresented until solution is found.args:               The command line arguments holder.Returns:True if experiment finished with successful solver found. """# set random seed# seed = int(time.time())# random.seed(seed)# print("Selected random seed:", seed)# Load configuration.config = neat.Config(neat.DefaultGenome,neat.DefaultReproduction,neat.DefaultSpeciesSet,neat.DefaultStagnation,config_file)# Create the population, which is the top-level object for a NEAT run.p = neat.Population(config)# Create the trial simulationglobal trial_simtrial_sim = MazeSimulationTrial(maze_env=maze_env,population=p,archive=novelty_archive)# Add a stdout reporter to show progress in the terminal.p.add_reporter(neat.StdOutReporter(True))stats = neat.StatisticsReporter()p.add_reporter(stats)# Run for up to N generations.start_time = time.time()best_genome = p.run(eval_genomes, n=n_generations)elapsed_time = time.time() - start_time# Display the best genome among generations.print('\nBest genome: \n%s' % (best_genome))solution_found = (best_genome.fitness >= config.fitness_threshold)if solution_found:print("SUCCESS: The stable maze solver controller was found!!!")else:print("FAILURE: Failed to find the stable maze solver controller!!!")# write the record store datars_file = os.path.join(trial_out_dir,'data.pickle')trial_sim.record_store.dump(rs_file)print("Record store file: %s" % rs_file)# print("Random seed:", seed)print("Trial elapsed time: %.3f sec" % (elapsed_time))# visualize the experiment resultsshow_results = solution_found or not silentif save_results or show_results:node_names = {-1:'RF_R',-2:'RF_FR',-3:'RF_F',-4:'RF_FL',-5:'RF_L',-6:'RF_B',-7:'RAD_F',-8:'RAD_L',-9:'RAD_B',-10:'RAD_R',0:'ANG_VEL',1:'VEL'}visualize.draw_net(config,best_genome,view=show_results,node_names=node_names,directory=trial_out_dir,fmt='svg')if args is None:visualize.draw_maze_records(maze_env,trial_sim.record_store.records,view=show_results)else:visualize.draw_maze_records(maze_env,trial_sim.record_store.records,view=show_results,width=args.width,height=args.height,filename=os.path.join(trial_out_dir,'maze_records.svg'))visualize.plot_stats(stats,ylog=False,view=show_results,filename=os.path.join(trial_out_dir,'avg_fitness.svg'))visualize.plot_species(stats,view=show_results,filename=os.path.join(trial_out_dir,'speciation.svg'))# store NoveltyItems archive datatrial_sim.archive.write_fittest_to_file(path=os.path.join(trial_out_dir, 'ns_items_fittest.txt'))trial_sim.archive.write_to_file(path=os.path.join(trial_out_dir, 'ns_items_all.txt'))            # store NoveltyItems archive datatrial_sim.archive.write_fittest_to_file(path=os.path.join(trial_out_dir,'ns_items_fittest.txt'))trial_sim.archive.write_to_file(path=os.path.join(trial_out_dir,'ns_items_all.txt'))# create the best genome simulation path and redarmaze_env = copy.deepcopy(trial_sim.orig_maze_environment)control_net = neat.nn.FeedForwardNetwork.create(best_genome, config)path_points = []evaluate_fitness = maze.maze_simulation_evaluate(env=maze_env,net=control_net,time_steps=SOLVER_TIME_STEPS,path_points=path_points)print("Evaluated fitness of best agent: %f" % evaluate_fitness)visualize.draw_agent_path(trial_sim.orig_maze_environment,path_points,best_genome,view=show_results,width=args.width,height=args.height,filename=os.path.join(trial_out_dir,'best_solver_path.svg'))return solution_foundif __name__ == '__main__':# read command line parametersparser = argparse.ArgumentParser(description="The maze experiment runner (Novelty Search).")parser.add_argument('-m', '--maze', default='medium', help='The maze configuration to use.')parser.add_argument('-g', '--generations', default=500, type=int, help='The number of generations for the evolutionary process.')parser.add_argument('-t', '--trials', type=int, default=1, help='The number of trials to run')parser.add_argument('-n', '--ns_threshold', type=float, default=6.0,help="The novelty threshold value for the archive of NoveltyItems.")parser.add_argument('-r', '--location_sample_rate', type=int, default=4000,help="The sample rate of agent position points saving during simulation steps.")parser.add_argument('--width', type=int, default=400, help='The width of the records subplot')parser.add_argument('--height', type=int, default=400, help='The height of the records subplot')args = parser.parse_args()if not (args.maze == 'medium' or args.maze == 'hard'):print('Unsupported maze configuration: %s' % args.maze)exit(1)# The current working directorylocal_dir = os.path.dirname(__file__)# The directory to store outputsout_dir = os.path.join(local_dir, 'out')out_dir = os.path.join(out_dir, 'maze_ns')# Determine path to configuration file.config_path = os.path.join(local_dir, 'maze_config.ini')# Clean results of previous run if any or init the output directoryout_dir = os.path.join(out_dir, args.maze)utils.clear_output(out_dir)# Read the maze environment configurationmaze_env_config = os.path.join(local_dir, '%s_maze.txt' % args.maze)maze_env = maze.read_environment(maze_env_config)maze_env.location_sample_rate = args.location_sample_rate# Run the maze experiment trialsprint("Starting the %s maze experiment (Novelty Search), for %d trials" % (args.maze, args.trials))for t in range(args.trials):print("\n\n----- Starting Trial: %d ------" % (t))#create novelty archivenovelty_archive = archive.NoveltyArchive(threshold=args.ns_threshold,metric=maze.maze_novelty_metric)trial_out_dir = os.path.join(out_dir, str(t))os.makedirs(trial_out_dir, exist_ok=True)solution_found = run_experiment(config_file = config_path, maze_env = maze_env,novelty_archive = novelty_archive, trial_out_dir = trial_out_dir,n_generations = args.generations, args=args,save_results=True,silent=True)print("\n----- Trial %d complete, solution found: %s -------\n" % (t,solution_found))

novelty_archive.py

from functools import total_ordering# how many nearst neighbors to consider for calculating novelty score
KNNNoveltyScore = 15
# The maximal allowed size for fittest items list
FittestAllowedSize = 5
# The minimal number of items to include in the archive unconditionaly
ArchiveSeedAmount = 1@total_ordering
class NoveltyItem:"""The class to encapsulate information about particular item thatholds information about novelty score associated with specificgenome along with auxiliary information. It is used in combinationwith NoveltyArchive"""def __init__(self,generation=-1,genomeId=-1,fitness=-1,novelty=-1):"""generation: The evolution generation when this item was createdgenomeId:   The ID of genome associated with itfitness:    The goal-oriented fitness score of genome associated with this itemnovelty:    The novelty score of genome"""self.generation = generationself.genomeId = genomeIdself.fitness = fitnessself.novelty = novelty# Indicates whether this item was already added to the archiveself.in_archive = False# The list holding data points associated with this item that will be used# to calculate distance between this item and any other item. This distance# will be used to estimate the novelty score associated with the item.self.data = []def __str__(self):"""The function to create string representation"""return "%s: id: %d, at generation: %d, fitness: %f, novelty: %f\tdata: %s" % \(self.__class__.__name__, self.genomeId, self.generation, self.fitness, self.novelty, self.data)def _is_valid_operand(self,other):return (hasattr(other,'fitness') and hasattr(other,'novelty'))def __lt__(self,other):"""Compare if this item is less than supplied other item bygoal-oriented fitness value."""if not self._is_valid_operand(other):return NotImplementedif self.fitness < other.fitness:return Trueelif self.fitness == other.fitness:# less novel is lessreturn self.novelty < other.noveltyreturn False@total_ordering
class ItemsDistance:"""Holds information about distance between the two NoveltyItem objects basedon the nearest neighbour metric."""def __init__(self,first_item,second_item,distance):"""Creates new instance for two NoveltyItem objectsArguments:first_item:     The item from which distance is measuredsecond_item:    The item to which distance is measureddistance:       The distance value"""self.first_item = first_itemself.second_item = second_itemself.distance = distancedef _is_valid_operand(self,other):return hasattr(other,"distance")def __lt__(self,other):"""Compare if the distance in this object is less that in other."""if not self._is_valid_operand(other):return NotImplementedreturn self.distance < other.distanceclass NoveltyArchive:"""The novelty archive contains all of the novel items we have encountered thus far."""def __init__(self, threshold, metric):"""Creates new instance with specified novelty threshold and functiondefined novelty metric.Arguments:threshold:  The minimal novelty score of the item to be included into this archive.metric:     The function to calculate the novelty score of specific genome."""self.novelty_metric = metricself.novelty_threshold = threshold# the minimal possiable value of novelty thresholdself.novelty_floor = 0.25# the novel items added during current generationself.items_added_in_generation = 0# the counter to keep track of how many generations passed # since we've added to the archiveself.time_out = 0# the parameter specifying how many neighbors to look at for the K-nearest # neighbor distance estimation to be used in novelty scoreself.neighbors = KNNNoveltyScore# the current evolutionary generationself.generation = 0# list with all novel items found so farself.novel_items = []# list with all novel items found that is related to the fittest# genomes (using the goal-oriented fitness score)self.fittest_items = []def evaluate_individual_novelty(self, genome, genomes, n_items_map, only_fitness=False):"""The function to evaluate the novelty score of a single genome withinpopulation and update its fitness if appropriate (only_fitness=True)Arguments:genome:         The genome to evaluategenomes:        The current population of genomesn_items_map:    The map of novelty items for the current population by genome IDonly_fitness:   The flag to indicate if only fitness should be calculated and assigned to genomeusing the novelty score. Otherwise novelty score will be used to acceptgenome into novelty items archive.Returns:The calculated novelty score for individual genome."""if genome.key not in n_items_map:print("WARNING! Found Genome without novelty point associated: %s" +"\nNovelty evaluation will be skipped for it. Probably winner found!" % genome.key)returnitem = n_items_map[genome.key]# Check if individual was marked for extinction due to failure to meet minimal fitness criterionif item.fitness == -1.0:return -1.0result = 0.0if only_fitness:# assign genome fitness according to the average novelty within# archive and populationresult = self._novelty_avg_knn(item=item, genomes=genomes, n_items_map=n_items_map)else:# consider adding a NoveltyItem to the archive based on the # distance to a closest neighborresult = self._novelty_avg_knn(item=item, neighbors=1, n_items_map=n_items_map)if result > self.novelty_threshold or len(self.novel_items) < ArchiveSeedAmount:self._add_novelty_item(item)# store found values to the novelty itemitem.novelty = resultitem.generation = self.generationreturn resultdef update_fittest_with_genome(self,genome,n_items_map):"""The function to update list of NovelItems for the genomes with the higherfitness scores achieved so far during the evolution.Arguments:genome:         The genome to evaluaten_items_map:    The map of novelty items for the current population by genome ID"""assert genome.key in n_items_mapitem = n_items_map[genome.key]if len(self.fittest_items) < FittestAllowedSize:# store novelty item into fittestself.fittest_items.append(item)# sort in descending order by fitnessself.fittest_items.sort(reverse=True)else:last_item = self.fittest_items[-1]if item.fitness > last_item.fitness:# store novelty item into fittestself.fittest_items.append(item)# sort in descending order by fitnessself.fittest_items.sort(reverse=True)# remove the less fit itemdel self.fittest_items[-1]def end_of_generation(self):"""The function to update archive state at the end of the generation."""self.generation += 1self._adjust_archive_settings()def write_to_file(self,path):"""The function to write all NoveltyItems stored in this archive.Arguments:path: The path to the file where to store NoveltyItems"""with open(path, 'w') as f:for ni in self.novel_items:f.write("%s\n" % ni)def write_fittest_to_file(self,path):"""The function to write the list of NoveltyItems of fittests genomesthat was collected during the evolution.Arguments:path: The path to the file where to store NoveltyItems"""with open(path, 'w') as f:for ni in self.fittest_items:f.write("%s\n" % ni)def _add_novelty_item(self,item):"""The function to add specified NoveltyItem to this archive.Arguments:item: The NoveltyItem to be added."""# add itemitem.in_archive = Trueitem.generation = self.generationself.novel_items.append(item)self.items_added_in_generation += 1def _adjust_archive_settings(self):"""The function to adjust the dynamic novelty threshold depending on how many have NoveltyItem objects have been added to the archive recently"""if self.items_added_in_generation == 0:self.time_out += 1else:self.time_out = 0# if no items have been added for the last 10 generations lower the thresholdif self.time_out >= 10:self.novelty_threshold *= 0.95if self.novelty_threshold < self.novelty_floor:self.novelty_threshold = self.novelty_floorself.time_out = 0# if more than four individuals added in last generation then raise thresholdif self.items_added_in_generation >= 4:self.novelty_threshold *= 1.2# reset countersself.items_added_in_generation = 0def _map_novelty(self,item):"""The function to map the novelty metric across the archive against provided itemArguments:item: The NoveltyItem to be used for archive mapping.Returns:The list with distances (novelty scores) of provided item from items stored in this archive."""distances = [None] * len(self.novel_items)for i, n in enumerate(self.novel_items):distances[i] = ItemsDistance(first_item = n,second_item = item,distance = self.novelty_metric(n, item))return distancesdef _map_novelty_in_population(self,item,genomes,n_items_map):"""The function to map the novelty metric across the archive and the current populationagainst the provided item.Arguments:item:        The NoveltyItem to be used for archive mapping.genomes:     The list of genomes from current population.n_items_map: The map of novelty items for the current population by genome ID.Returns:The list with distances (novelty scores) of provided item from items stored in this archiveand from the novelty items associated with genomes in current population."""# first, map item against the archivedistances = self._map_novelty(item)# second, map item against the populationfor genome_id, _ in genomes:if genome_id in n_items_map:gen_item = n_items_map[genome_id]distance = ItemsDistance(first_item = gen_item,second_item = item,distance = self.novelty_metric(gen_item, item))distances.append(distance)return distancesdef _novelty_avg_knn(self,item,n_items_map,genomes=None,neighbors=None):"""The function to calculate the novelty score of a given item within the provided population if anyusing a K-nearest neighbor algorithm.Argumnets:item:        The NoveltyItem to calculate the scoren_items_map: The map of novelty items for the current population by genome IDgenomes:     The list of genomes from population or Noneneighbors:   The number of neighbors to use for calculation (None - to use archive settings)Returns:The density within the vicinity of the provided NoveltyItem calculated using the K-nearest neighboralgorithm. This density can be used either as a novelty score value or as a fitness value."""distances = Noneif genomes is not None:distances = self._map_novelty_in_population(item=item,genomes=genomes,n_items_map=n_items_map)else:distances = self._map_novelty(item=item)# sort by distance (novelty) in ascending order - the minimal firstdistances.sort()# if neighbors size not set - use value from archive parametersif neighbors is None:neighbors = self.neighborsdensity, weight, distance_sum = 0.0, 0.0, 0.0length = len(distances)if length >= ArchiveSeedAmount:length = neighborsif len(distances) < length:# the number of mapped distances id less than number of neighborslength = len(distances)i = 0while weight < float(neighbors) and i < length:distance_sum += distances[i].distanceweight += 1.0i += 1# finding the averageif weight > 0:density = distance_sum / weightreturn density

visualize.py

from __future__ import print_functionimport copy
import warnings
import random
import argparse
import osimport graphviz
import matplotlib.pyplot as plt
import matplotlib.lines as mlines
import matplotlib.patches as mpatches
import numpy as npimport geometry
import agent
import maze_environment as mazedef plot_stats(statistics, ylog=False, view=False,filename='avg_fitness.svg'):"""Plots the polulation's average and best fitness."""if plt is None:warnings.warn("This display is not available due to a missing optional dependency (matplotlib)")returngeneration = range(len(statistics.most_fit_genomes))best_fitness = [c.fitness for c in statistics.most_fit_genomes]avg_fitness = np.array(statistics.get_fitness_mean())stdev_fitness = np.array(statistics.get_fitness_stdev())plt.plot(generation, avg_fitness, 'b-', label='average')plt.plot(generation, avg_fitness - stdev_fitness, 'g-.', label="-1 sd")plt.plot(generation, avg_fitness + stdev_fitness, 'g-', label="+1 sd")plt.plot(generation, best_fitness, 'r-', label="best")plt.title("Population's average and best fitness")plt.xlabel("Generations")plt.ylabel("Fitness")plt.grid()plt.legend(loc="best")if ylog:plt.gca().set_yscale('symlog')plt.savefig(filename)if view:plt.show()plt.close()def plot_species(statistics, view=False, filename='speciation.svg'):"""Visualizes speciation throught evolution."""if plt is None:warnings.warn("This display is not available due to a missing optional dependency (matplotlib)")returnspecies_sizes = statistics.get_species_sizes()num_generations = len(species_sizes)curves = np.array(species_sizes).Tfig, ax = plt.subplots()ax.stackplot(range(num_generations), *curves)plt.title("Speciation")plt.ylabel("Size per Species")plt.xlabel("Generations")plt.savefig(filename)if view:plt.show()plt.close()def draw_net(config, genome, view=False,filename=None, directory=None, node_names=None, show_disabled=True,prune_unused=False,node_colors=None,fmt='svg'):"""Receives a genome and draws a neural network with arbitrary topology."""# Attributes for network nodesif graphviz is None:warnings.warn("This display is not available due to a missing optional dependency (graphviz)")returnif node_names is None:node_names = {}assert type(node_names) is dictif node_colors is None:node_colors = {}assert type(node_colors) is dictnode_attrs = {'shape': 'circle','fontsize': '9','height': '0.2','width': '0.2'}dot = graphviz.Digraph(format=fmt,node_attr=node_attrs)inputs = set()for k in config.genome_config.input_keys:inputs.add(k)name = node_names.get(k,str(k))input_attrs = {'style': 'filled','shape': 'box','fillcolor': node_colors.get(k, 'lightgray')}dot.node(name, _attributes=input_attrs)outputs = set()for k in config.genome_config.output_keys:outputs.add(k)name = node_names.get(k,str(k))node_attrs = {'style': 'filled','fillcolor': node_colors.get(k, 'lightblue')}dot.node(name, _attributes=node_attrs)if prune_unused:connections = set()for cg in genome.connections.values():if cg.enabled or show_disabled:connections.add((cg.in_node_id,cg.out_node_id))used_nodes = copy.copy(outputs)pending = copy.copy(outputs)while pending:new_pending = set()for a,b in connections:if b in pending and a not in used_nodes:new_pending.add(a)used_nodes.add(a)pending = new_pendingelse:used_nodes = set(genome.nodes.keys())for n in used_nodes:if n in inputs or n in outputs:continueattrs = {'style': 'filled','fillcolor': node_colors.get(n,'white')}dot.node(str(n), _attributes=attrs)for cg in genome.connections.values():if cg.enabled or show_disabled:# if cg.input not in used_nodes or cg.output not in used_nodes:#     continueinput_node, output_node = cg.keya = node_names.get(input_node,str(input_node))b = node_names.get(output_node,str(output_node))style = 'solid' if cg.enabled else 'dotted'color = 'green' if cg.weight > 0 else 'red'width = str(0.1 + abs(cg.weight / 5.0))dot.edge(a, b, _attributes={'style': style, 'color': color, 'penwidth': width})dot.render(filename, directory, view=view)return dotdef draw_agent_path(maze_env, path_points, genome,filename=None, view=False,show_axes=False,width=400,height=400,fig_height=4):"""The function to draw path of the maze solver agent through the maze.Arguments:maze_env: The maze environment configuration.path_points: The list of agent positions during simulation.genome: The genome of solver agent.filename: The name of file to store plot.view: The flag to indicate whether to view plot.width: The width of drawing in pixelsheight: The height of drawing in pixelsfig_height: The plot figure height in inches"""# initialize plottingfig, ax = plt.subplots()fig.set_dpi(100)fig_width = fig_height * (float(width) / float(height)) - 0.2print("Plot figure width: %.1f, height: %.1f" % (fig_width,fig_height))fig.set_size_inches(fig_width, fig_height)ax.set_xlim(0, width)ax.set_ylim(0, height)ax.set_title('Genome ID: %s, Path Length: %d' % (genome.key,len(path_points)))#draw pathfor p in path_points:circle = plt.Circle((p.x,p.y),2.0,facecolor='b')ax.add_patch(circle)# draw maze_draw_maze_(maze_env, ax)# turn off axis renderingif not show_axes:ax.axis('off')# Invert Y axis to have coordinates origin at the top leftax.invert_yaxis()# save figure to fileif filename is not None:plt.savefig(filename)if view:plt.show()plt.close()def draw_maze_records(maze_env, records,best_threshold=0.8,filename=None, view=False, show_axes=False,width=400,height=400,fig_height=7):"""The function to draw maze with recorded agents positions.Arguments:maze_env:       The maze environment configuration.records:        The records of solver agents collected during NEAT execution.best_threshold: The minimal fitness of maze solving agent's species to be included into the best ones.filename:       The name of file to store plot.view:           The flag to indicate whether to view plot.width:          The width of drawing in pixelsheight:         The height of drawing in pixelsfig_height:      The plot figure height in inches"""# find the distance threshold for the best speciesdist_threshold = maze_env.agent_distance_to_exit() * (1.0 - best_threshold)# generate color palette and find the best species IDSmax_sid = 0for r in records:if r.species_id > max_sid:max_sid = r.species_idcolors = [None] * (max_sid + 1)sp_idx = [False] * (max_sid + 1)best_sp_idx = [0] * (max_sid + 1)for r in records:if not sp_idx[r.species_id]:sp_idx[r.species_id] = Truergb = (random.random(),random.random(),random.random())colors[r.species_id] = rgbif maze_env.exit_point.distance(geometry.Point(r.x,r.y)) <= dist_threshold:best_sp_idx[r.species_id] += 1# initialize plottingfig = plt.figure()fig.set_dpi(100)fig_width = fig_height * (float(width)/float(2.0 * height)) - 0.2print("Plot figure width: %.1f, height: %.1f" % (fig_width,fig_height))fig.set_size_inches(fig_width,fig_height)ax1,ax2 = fig.subplots(2,1,sharex=True)ax1.set_xlim(0, width)ax1.set_ylim(0, height)ax2.set_xlim(0, width)ax2.set_ylim(0, height)# draw speciesn_best_species = 0for i,v in enumerate(best_sp_idx):if v > 0:n_best_species += 1_draw_species_(records=records,sid=i,colors=colors,ax=ax1)else:_draw_species_(records=records,sid=i,colors=colors,ax=ax2)ax1.set_title('fitness >= %.1f, species: %d' % (best_threshold, n_best_species))ax2.set_title('fitness < %.1f' % best_threshold)# draw maze_draw_maze_(maze_env, ax1)_draw_maze_(maze_env, ax2)# turn off axis renderingif not show_axes:ax1.axis('off')ax2.axis('off')# Invert Y axis to have coordinates origin at the top leftax1.invert_yaxis()ax2.invert_yaxis()# Save figure to fileif filename is not None:plt.savefig(filename)if view:plt.show()plt.close()def _draw_species_(records, sid, colors, ax):"""The function to draw specific species from the records withparticular color.Arguments:records: The records of solver agents collected during NEAT execution.sid: The species IScolors: The colors table by species IDax: The figure axis instance"""for r in records:if r.species_id == sid:circle = plt.Circle((r.x,r.y),2.0,facecolor=colors[r.species_id])ax.add_patch(circle)def _draw_maze_(maze_env,ax):"""The function to draw maze environmentArguments:maze_env: The maze environment configuration.ax: The figure axis instance"""for wall in maze_env.walls:line = plt.Line2D((wall.a.x,wall.b.x),(wall.a.y,wall.b.y),lw=1.5)ax.add_line(line)# draw start pointstart_circle = plt.Circle((maze_env.agent.location.x,maze_env.agent.location.y),radius=2.5,facecolor=(0.6,1.0,0.6),edgecolor='w')ax.add_patch(start_circle)# draw exit pointexit_circle = plt.Circle((maze_env.exit_point.x,maze_env.exit_point.y),radius=2.5,facecolor=(1.0,0.2,0.0),edgecolor='w')ax.add_patch(exit_circle)if __name__ == '__main__':# read command line parametersparser = argparse.ArgumentParser(description="The maze experiment visualizer.")parser.add_argument('-m', '--maze', default='medium', help='The maze configuration to use.')parser.add_argument('-r', '--records', help='The records file.')parser.add_argument('-o', '--output', help='The file to store the plot.')parser.add_argument('--width', type=int, default=400, help='The width of the subplot')parser.add_argument('--height', type=int, default=400, help='The height of the subplot')parser.add_argument('--fig_height', type=float, default=7, help='The height of the plot figure')parser.add_argument('--show_axes', type=bool, default=False, help='The flag to indicate whether to show plot axes.')args = parser.parse_args()local_dir = os.path.dirname(__file__)if not (args.maze == 'medium' or args.maze == 'hard'):print("Unsupported maze configuration: %s" % args.maze)exit(1)# read maze environmentmaze_env_config = os.path.join(local_dir,'%s_maze.txt' % args.maze)maze_env = maze.read_environment(maze_env_config)# read agents recordsrs = agent.AgentRecordStore()rs.load(args.records)# render visualizationrandom.seed(42)draw_maze_records(maze_env,rs.records,width=args.width,height=args.height,fig_height=args.fig_height,view=True,show_axes=args.show_axes,filename=args.output)