Deepmind 开放了自己的星际争霸机器学习环境，现在只需要安装对应的 pysc2 包，就可以自己写一个星际2 AI啦！本文是Rest探路者分享在博客园的一个相关项目，里面有非常详尽的代码，有兴趣的小伙伴可以参考。
文 | Rest探路者
出处 | 博客园
准备
我的环境是python3.6，sc2包0.11.1，机器学习包：pysc2，地图下载链接maps：https://github
.com/Blizzard/s2client-proto#downloads。
pysc2是DeepMind开发的星际争霸Ⅱ学习环境。它是封装星际争霸Ⅱ机器学习API，同时也提供Python增强学习环境。以神族为例编写代码，神族建筑科技图如下：
采矿
# -*- encoding: utf-8 -*-'''@File : __init__.py.py@Modify Time @Author @Desciption------------ ------- -----------2019/11/3 12:32 Jonas None'''import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerclass SentdeBot(sc2.BotAI):async def on_step(self, iteration: int):await self.distribute_workersrun_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)],realtime = True)
注意game_data.py的assert self.id != 0注释掉 pixel_map.py的assert self.bits_per_pixel % 8 == 0, "Unsupported pixel density"注释掉，否则会报错
运行结果如下，农民开始采矿
可以正常采矿
建造农民和水晶塔
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *class SentdeBot(sc2.BotAI):async def on_step(self, iteration: int):await self.distribute_workersawait self.build_workersawait self.build_pylons# 建造农民async def build_workers(self):# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且水晶不是正在建造if self.supply_left<5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.e
xists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON,near=nexuses.first)## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)],realtime = True)
运行结果如下，基地造农民，农民造水晶
收集气体和开矿
代码如下
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *class SentdeBot(sc2.BotAI):async def on_step(self, iteration: int):await self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expand# 建造农民async def build_workers(self):# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))## 开矿async def expand(self):if self.units(UnitTypeId.NEXUS).amount<3 and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)], realtime=False)
run_game的realtime设置成False，可以在加速模式下运行游戏。
运行效果如下：
可以建造吸收厂和开矿
建造军队
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *class SentdeBot(sc2.BotAI):async def on_step(self, iteration: int):await self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expandawait self.offensive_force_buildingsawait self.build_offensive_force# 建造农民async def build_workers(self):# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))## 开矿async def expand(self):if self.units(UnitTypeId.NEXUS).amount<2 and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 建造进攻性建筑async def offensive_force_buildings(self):if self.units(UnitTypeId.PYLON).ready.exists:pylon = self.units(UnitTypeId.PYLON).ready.randomif self.units(UnitTypeId.PYLON).ready.exists:# 根据神族建筑科技图，折跃门建造过后才可以建造控制核心if self.units(UnitTypeId.GATEWAY).ready.exists:if not self.units(UnitTypeId.CYBERNETICSCORE):if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):await self.build(UnitTypeId.CYBERNETICSCORE,near = pylon)# 否则建造折跃门else:if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):await self.build(UnitTypeId.GATEWAY,near=pylon)# 造兵async def build_offensive_force(self):# 无队列化建造for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:if self.can_afford(UnitTypeId.STALKER) and self.supply_left>0:await self.do(gw.train(UnitTypeId.STALKER))## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)], realtime=False)
运行结果如下：
可以看到，我们建造了折跃门和控制核心并训练了追猎者
控制部队进攻
代码如下
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *import randomclass SentdeBot(sc2.BotAI):async def on_step(self, iteration: int):await self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expandawait self.offensive_force_buildingsawait self.build_offensive_forceawait self.attack# 建造农民async def build_workers(self):# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))## 开矿async def expand(self):if self.units(UnitTypeId.NEXUS).amount<3 and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 建造进攻性建筑async def offensive_force_buildings(self):if self.units(UnitTypeId.PYLON).ready.exists:pylon = self.units(UnitTypeId.PYLON).ready.random# 根据神族建筑科技图，折跃门建造过后才可以建造控制核心if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):await self.build(UnitTypeId.CYBERNETICSCORE,near = pylon)# 否则建造折跃门elif len(self.units(UnitTypeId.GATEWAY))<=3:if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):await self.build(UnitTypeId.GATEWAY,near=pylon)## 造兵async def build_offensive_force(self):# 无队列化建造for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:if self.can_afford(UnitTypeId.STALKER) and self.supply_left>0:await self.do(gw.train(UnitTypeId.STALKER))## 寻找目标def find_target(self,state):if len(self.known_enemy_units)>0:# 随机选取敌方单位return random.choice(self.known_enemy_units)elif len(self.known_enemy_units)>0:# 随机选取敌方建筑return random.choice(self.known_enemy_structures)else:# 返回敌方出生点位return self.enemy_start_locations[0]## 进攻async def attack(self):# 追猎者数量超过15个开始进攻if self.units(UnitTypeId.STALKER).amount>15:for s in self.units(UnitTypeId.STALKER).idle:await self.do(s.attack(self.find_target(self.state)))# 防卫模式：视野范围内存在敌人，开始攻击if self.units(UnitTypeId.STALKER).amount>5:if len(self.known_enemy_units)>0:for s in self.units(UnitTypeId.STALKER).idle:await self.do(s.attack(random.choice(self.known_enemy_units)))## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Medium)], realtime=False)
运行结果如下
可以看到，4个折跃门训练追猎者并发动进攻。
击败困难电脑
我们目前的代码只能击败中等和简单电脑，那么如何击败困难电脑呢？代码如下
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *import randomclass SentdeBot(sc2.BotAI):def __init__(self):# 经过计算，每分钟大约165迭代次数self.ITERATIONS_PER_MINUTE =165# 最大农民数量self.MAX_WORKERS =65async def on_step(self, iteration: int):self.iteration = iterationawait self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expandawait self.offensive_force_buildingsawait self.build_offensive_forceawait self.attack# 建造农民async def build_workers(self):# 星灵枢钮*16（一个基地配备16个农民）大于农民数量并且现有农民数量小于MAX_WORKERSif len(self.units(UnitTypeId.NEXUS))*16>len(self.units(UnitTypeId.PROBE)) and len(self.units(UnitTypeId.PROBE))<self.MAX_WORKERS:# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿建造农民if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))## 开矿async def expand(self):# (self.iteration / self.ITERATIONS_PER_MINUTE)是一个缓慢递增的值,动态开矿if self.units(UnitTypeId.NEXUS).amount<self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 建造进攻性建筑async def offensive_force_buildings(self):print(self.iteration / self.ITERATIONS_PER_MINUTE)if self.units(UnitTypeId.PYLON).ready.exists:pylon = self.units(UnitTypeId.PYLON).ready.random# 根据神族建筑科技图，折跃门建造过后才可以建造控制核心if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)# 否则建造折跃门# (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一个缓慢递增的值elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):await self.build(UnitTypeId.GATEWAY, near=pylon)# 控制核心存在的情况下建造星门if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):await self.build(UnitTypeId.STARGATE, near=pylon)## 造兵async def build_offensive_force(self):# 无队列化建造for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:await self.do(gw.train(UnitTypeId.STALKER))for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:await self.do(sg.train(UnitTypeId.VOIDRAY))## 寻找目标def find_target(self,state):if len(self.known_enemy_units)>0:# 随机选取敌方单位return random.choice(self.known_enemy_units)elif len(self.known_enemy_units)>0:# 随机选取敌方建筑return random.choice(self.known_enemy_structures)else:# 返回敌方出生点位return self.enemy_start_locations[0]## 进攻async def attack(self):# {UNIT: [n to fight, n to defend]}aggressive_units = {UnitTypeId.STALKER: [15, 5],UnitTypeId.VOIDRAY: [8, 3]}for UNIT in aggressive_units:# 攻击模式if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][1]:for s in self.units(UNIT).idle:await self.do(s.attack(self.find_target(self.state)))# 防卫模式elif self.units(UNIT).amount > aggressive_units[UNIT][1]:if len(self.known_enemy_units) > 0:for s in self.units(UNIT).idle:await self.do(s.attack(random.choice(self.known_enemy_units)))## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)], realtime=False)
运行结果如下
可以看到，击败了困难人族电脑，但是电脑选择了rush战术，我们写得AI脚本会输掉游戏。显然，这不是最佳方案。“只有AI才能拯救我的胜率”，请看下文。
采集地图数据
这次我们只造一个折跃门，全力通过星门造虚空光辉舰 修改offensive_force_buildings(self)方法的判断
elif len(self.units(GATEWAY)) < 1:if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):await self.build(GATEWAY, near=pylon)
注释或者删除build_offensive_force(self)的建造追猎者的代码
## 造兵async def build_offensive_force(self):# 无队列化建造# for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:# if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:## if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:# await self.do(gw.train(UnitTypeId.STALKER))for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:await self.do(sg.train(UnitTypeId.VOIDRAY))
attack(self)中的aggressive_units注释掉Stalker 导入numpy和cv2库
game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)
建立以地图Heigt为行，Width为列的三维矩阵
for nexus in self.units(NEXUS):nex_pos = nexus.positionprint(nex_pos)cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1) # BGR
遍历星灵枢纽，获取下一个位置，画圆，circle(承载圆的img, 圆心, 半径, 颜色, thickness=-1表示填充)，接下来我们要垂直翻转三维矩阵，因为我们建立的矩阵左上角是原点(0,0)，纵坐标向下延申，横坐标向右延申。翻转之后就成了正常的坐标系。
flipped = cv2.flip(game_data, 0)
图像缩放，达到可视化最佳。
resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)cv2.imshow('Intel', resized)cv2.waitKey(1)
至此，完整代码如下
import sc2from sc2 import run_game, maps, Race, Difficultyfrom sc2.player import Bot, Computerfrom sc2.constants import *import randomimport numpy as npimport cv2class SentdeBot(sc2.BotAI):def __init__(self):# 经过计算，每分钟大约165迭代次数self.ITERATIONS_PER_MINUTE =165# 最大农民数量self.MAX_WORKERS =65async def on_step(self, iteration: int):self.iteration = iterationawait self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expandawait self.offensive_force_buildingsawait self.build_offensive_forceawait self.intelawait self.attackasync def intel(self):# 根据地图建立的三维矩阵game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)for nexus in self.units(UnitTypeId.NEXUS):nex_pos = nexus.position# circle(承载圆的img, 圆心, 半径, 颜色, thickness=-1表示填充)# 记录星灵枢纽的位置cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)# 图像翻转垂直镜像flipped = cv2.flip(game_data, 0)# 图像缩放# cv2.resize(原图像,输出图像的大小,width方向的缩放比例,height方向缩放的比例)resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)cv2.imshow('Intel', resized)# cv2.waitKey(每Xms刷新图像)cv2.waitKey(1)# 建造农民async def build_workers(self):# 星灵枢钮*16（一个基地配备16个农民）大于农民数量并且现有农民数量小于MAX_WORKERSif len(self.units(UnitTypeId.NEXUS)) * 16 > len(self.units(UnitTypeId.PROBE)) and len(self.units(UnitTypeId.PROBE)) < self.MAX_WORKERS:# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿建造农民if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0, vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR, vaspene))## 开矿async def expand(self):# (self.iteration / self.ITERATIONS_PER_MINUTE)是一个缓慢递增的值,动态开矿if self.units(UnitTypeId.NEXUS).amount < self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 建造进攻性建筑async def offensive_force_buildings(self):print(self.iteration / self.ITERATIONS_PER_MINUTE)if self.units(UnitTypeId.PYLON).ready.exists:pylon = self.units(UnitTypeId.PYLON).ready.random# 根据神族建筑科技图，折跃门建造过后才可以建造控制核心if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)# 否则建造折跃门# (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一个缓慢递增的值# elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):elif len(self.units(UnitTypeId.GATEWAY)) < 1:if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):await self.build(UnitTypeId.GATEWAY, near=pylon)# 控制核心存在的情况下建造星门if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):await self.build(UnitTypeId.STARGATE, near=pylon)## 造兵async def build_offensive_force(self):# 无队列化建造for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:await self.do(sg.train(UnitTypeId.VOIDRAY))## 寻找目标def find_target(self, state):if len(self.known_enemy_units) > 0:# 随机选取敌方单位return random.choice(self.known_enemy_units)elif len(self.known_enemy_units) > 0:# 随机选取敌方建筑return random.choice(self.known_enemy_structures)else:# 返回敌方出生点位return self.enemy_start_locations[0]## 进攻async def attack(self):# {UNIT: [n to fight, n to defend]}aggressive_units = {UnitTypeId.VOIDRAY: [8, 3]}for UNIT in aggressive_units:# 攻击模式if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][1]:for s in self.units(UNIT).idle:await self.do(s.attack(self.find_target(self.state)))# 防卫模式elif self.units(UNIT).amount > aggressive_units[UNIT][1]:if len(self.known_enemy_units) > 0:for s in self.units(UNIT).idle:await self.do(s.attack(random.choice(self.known_enemy_units)))## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)], realtime=False)
运行结果如下
采集到了地图位置。
侦察
在intel(self)里创建一个字典draw_dict，UnitTypeId作为key,半径和颜色是value
draw_dict = {UnitTypeId.NEXUS: [15, (0, 255, 0)],UnitTypeId.PYLON: [3, (20, 235, 0)],UnitTypeId.PROBE: [1, (55, 200, 0)],UnitTypeId.ASSIMILATOR: [2, (55, 200, 0)],UnitTypeId.GATEWAY: [3, (200, 100, 0)],UnitTypeId.CYBERNETICSCORE: [3, (
150, 150, 0)],UnitTypeId.STARGATE: [5, (255, 0, 0)],UnitTypeId.ROBOTICSFACILITY: [5, (215, 
155, 0)],UnitTypeId.VOIDRAY: [3, (255, 100, 0)]}
迭代同上
for unit_type in draw_dict:for unit in self.units(unit_type).ready:pos = unit.positioncv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1)
存储三族的主基地名称（星灵枢纽，指挥中心，孵化场），刻画敌方建筑。
# 主基地名称main_base_names = ["nexus", "supplydepot", "hatchery"]# 记录敌方基地位置for enemy_building in self.known_enemy_structures:pos = enemy_building.positionif enemy_building.name.lower not in main_base_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 5, (200, 50, 212), -1)for enemy_building in self.known_enemy_structures:pos = enemy_building.positionif enemy_building.name.lower in main_base_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 15, (0, 0, 255), -1)
刻画敌方单位，如果是农民画得小些，其他单位则画大些。
for enemy_unit in self.known_enemy_units:if not enemy_unit.is_structure:worker_names = ["probe", "scv", "drone"]# if that unit is a PROBE, SCV, or DRONE... it's a workerpos = enemy_unit.positionif enemy_unit.name.lower in worker_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (55, 0, 155), -1)else:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (50, 0, 215), -1)
在offensive_force_buildings(self)方法中添加建造机械台
if self.units(CYBERNETICSCORE).ready.exists:if len(self.units(ROBOTICSFACILITY)) < 1:if self.can_afford(ROBOTICSFACILITY) and not self.already_pending(ROBOTICSFACILITY):await self.build(ROBOTICSFACILITY, near=pylon)
创建scout，训练Observer
async def scout(self):if len(self.units(OBSERVER)) > 0:scout = self.units(OBSERVER)[0]if scout.is_idle:enemy_location = self.enemy_start_locations[0]move_to = self.random_location_variance(enemy_location)print(move_to)await self.do(scout.move(move_to))else:for rf in self.units(ROBOTICSFACILITY).ready.noqueue:if self.can_afford(OBSERVER) and self.supply_left > 0:await self.do(rf.train(OBSERVER))
生成随机位置，很简单。意思是横坐标累计递增-0.2和0.2倍的横坐标，限制条件为如果x超过横坐标，那么就是横坐标最大值。
纵坐标同理。
def random_location_variance(self, enemy_start_location):x = enemy_start_location[0]y = enemy_start_location[1]x = ((random.randrange(-20, 20))/100) * enemy_start_location[0]y = ((random.randrange(-20, 20))/100) * enemy_start_location[1]if x < 0:x = 0if y < 0:y = 0if x > self.game_info.map_size[0]:x = self.game_info.map_size[0]if y > self.game_info.map_size[1]:y = self.game_info.map_size[1]go_to = position.Point2(position.Pointlike((x,y)))return go_to
完整代码如下
# -*- encoding: utf-8 -*-'''@File : demo.py@Modify Time @Author @Desciption------------ ------- -----------2019/11/3 12:32 Jonas None'''import sc2from sc2 import run_game, maps, Race, Difficulty, positionfrom sc2.player import Bot, Computerfrom sc2.constants import *import randomimport numpy as npimport cv2class SentdeBot(sc2.BotAI):def __init__(self):# 经过计算，每分钟大约165迭代次数self.ITERATIONS_PER_MINUTE =165# 最大农民数量self.MAX_WORKERS =50async def on_step(self, iteration: int):self.iteration = iterationawait self.scoutawait self.distribute_workersawait self.build_workersawait self.build_pylonsawait self.build_assimilatorsawait self.expandawait self.offensive_force_buildingsawait self.build_offensive_forceawait self.intelawait self.attack## 侦察async def scout(self):if len(self.units(UnitTypeId.OBSERVER)) > 0:scout = self.units(UnitTypeId.OBSERVER)[0]if scout.is_idle:enemy_location = self.enemy_start_locations[0]move_to = self.random_location_variance(enemy_location)print(move_to)await self.do(scout.move(move_to))else:for rf in self.units(UnitTypeId.ROBOTICSFACILITY).ready.noqueue:if self.can_afford(UnitTypeId.OBSERVER) and self.supply_left > 0:await self.do(rf.train(UnitTypeId.OBSERVER))async def intel(self):game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)# UnitTypeId作为key,半径和颜色是valuedraw_dict = {UnitTypeId.NEXUS: [15, (0, 255, 0)],UnitTypeId.PYLON: [3, (20, 235, 0)],UnitTypeId.PROBE: [1, (55, 200, 0)],UnitTypeId.ASSIMILATOR: [2, (55, 200, 0)],UnitTypeId.GATEWAY: [3, (200, 100, 0)],UnitTypeId.CYBERNETICSCORE: [3, (150, 150, 0)],UnitTypeId.STARGATE: [5, (255, 0, 0)],UnitTypeId.ROBOTICSFACILITY: [5, (215, 155, 0)],UnitTypeId.VOIDRAY: [3, (255, 100, 0)],# OBSERVER: [3, (255, 255, 255)],}for unit_type in draw_dict:for unit in self.units(unit_type).ready:pos = unit.positioncv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1)# 主基地名称main_base_names = ["nexus", "supplydepot", "hatchery"]# 记录敌方基地位置for enemy_building in self.known_enemy_structures:pos = enemy_building.position# 不是主基地建筑，画小一些if enemy_building.name.lower not in main_base_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 5, (200, 50, 212), -1)for enemy_building in self.known_enemy_structures:pos = enemy_building.positionif enemy_building.name.lower in main_base_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 15, (0, 0, 255), -1)for enemy_unit in self.known_enemy_units:if not enemy_unit.is_structure:worker_names = ["probe", "scv", "drone"]# if that unit is a PROBE, SCV, or DRONE... it's a workerpos = enemy_unit.positionif enemy_unit.name.lower in worker_names:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (55, 0, 155), -1)else:cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (50, 0, 215), -1)for obs in self.units(UnitTypeId.OBSERVER).ready:pos = obs.positioncv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (255, 255, 255), -1)# flip horizontally to make our final fix in visual representation:flipped = cv2.flip(game_data, 0)resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)cv2.imshow('Intel', resized)cv2.waitKey(1)def random_location_variance(self, enemy_start_location):x = enemy_start_location[0]y = enemy_start_location[1]x = ((random.randrange(-20, 20)) / 100) * enemy_start_location[0]y = ((random.randrange(-20, 20)) / 100) * enemy_start_location[1]if x < 0:x =0if y < 0:y =0if x > self.game_info.map_size[0]:x = self.game_info.map_size[0]if y > self.game_info.map_size[1]:y = self.game_info.map_size[1]go_to = position.Point2(position.Pointlike((x, y)))return go_to# 建造农民async def build_workers(self):# 星灵枢钮*16（一个基地配备16个农民）大于农民数量并且现有农民数量小于MAX_WORKERSif len(self.units(UnitTypeId.NEXUS)) * 16 > len(self.units(UnitTypeId.PROBE)) and len(self.units(UnitTypeId.PROBE)) < self.MAX_WORKERS:# 星灵枢纽(NEXUS)无队列建造，可以提高晶体矿的利用率，不至于占用资源for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:# 是否有50晶体矿建造农民if self.can_afford(UnitTypeId.PROBE):await self.do(nexus.train(UnitTypeId.PROBE))## 建造水晶async def build_pylons(self):## 供应人口和现有人口之差小于5且建筑不是正在建造if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):nexuses = self.units(UnitTypeId.NEXUS).readyif nexuses.exists:if self.can_afford(UnitTypeId.PYLON):await self.build(UnitTypeId.PYLON, near=nexuses.first)## 建造吸收厂async def build_assimilators(self):for nexus in self.units(UnitTypeId.NEXUS).ready:# 在瓦斯泉上建造吸收厂vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)for vaspene in vaspenes:if not self.can_afford(UnitTypeId.ASSIMILATOR):breakworker = self.select_build_worker(vaspene.position)if worker is None:breakif not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0, vaspene).exists:await self.do(worker.build(UnitTypeId.ASSIMILATOR, vaspene))## 开矿async def expand(self):# (self.iteration / self.ITERATIONS_PER_MINUTE)是一个缓慢递增的值,动态开矿if self.units(UnitTypeId.NEXUS).amount < self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(UnitTypeId.NEXUS):await self.expand_now## 建造进攻性建筑async def offensive_force_buildings(self):print(self.iteration / self.ITERATIONS_PER_MINUTE)if self.units(UnitTypeId.PYLON).ready.exists:pylon = self.units(UnitTypeId.PYLON).ready.random# 根据神族建筑科技图，折跃门建造过后才可以建造控制核心if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)# 否则建造折跃门# (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一个缓慢递增的值# elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):elif len(self.units(UnitTypeId.GATEWAY)) < 1:if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):await self.build(UnitTypeId.GATEWAY, near=pylon)# 控制核心存在的情况下建造机械台if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:if len(self.units(UnitTypeId.ROBOTICSFACILITY)) < 1:if self.can_afford(UnitTypeId.ROBOTICSFACILITY) and not self.already_pending(UnitTypeId.ROBOTICSFACILITY):await self.build(UnitTypeId.ROBOTICSFACILITY, near=pylon)# 控制核心存在的情况下建造星门if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):await self.build(UnitTypeId.STARGATE, near=pylon)## 造兵async def build_offensive_force(self):# 无队列化建造# for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:# if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:## if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:# await self.do(gw.train(UnitTypeId.STALKER))for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:await self.do(sg.train(UnitTypeId.VOIDRAY))## 寻找目标def find_target(self, state):if len(self.known_enemy_units) > 0:# 随机选取敌方单位return random.choice(self.known_enemy_units)elif len(self.known_enemy_units) > 0:# 随机选取敌方建筑return random.choice(self.known_enemy_structures)else:# 返回敌方出生点位return self.enemy_start_locations[0]## 进攻async def attack(self):# {UNIT: [n to fight, n to defend]}aggressive_units = {UnitTypeId.VOIDRAY: [8, 3]}for UNIT in aggressive_units:# 攻击模式if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][1]:for s in self.units(UNIT).idle:await self.do(s.attack(self.find_target(self.state)))# 防卫模式elif self.units(UNIT).amount > aggressive_units[UNIT][1]:if len(self.known_enemy_units) > 0:for s in self.units(UNIT).idle:await self.do(s.attack(random.choice(self.known_enemy_units)))## 启动游戏run_game(maps.get("AcidPlantLE"), [Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)], realtime=False)
运行结果如下，红色和粉红色是敌方单位。
作者：Rest探路者 出处：http://
www.cnblogs.com/Java-Starter/
回复下方「关键词」，获取优质资源
回复关键词「 pybook03」，立即获取主页君与小伙伴一起翻译的《Think Python 2e》电子版
回复关键词「入门资料」，立即获取主页君整理的 10 本 Python 入门书的电子版
回复关键词「m」，立即获取Python精选优质文章合集
回复关键词「book 数字」，将数字替换成 0 及以上数字，有惊喜好礼哦~
题图：pexels，CC0 授权。
,
