强化学习实战8.2——用PPO打赢星际争霸【画图和奖励设计】
画图接下来我们需要将战况通过图像返回给Agent游戏的图像太乱了特征是局部的因此要根据已知的信息绘制一张简化的新图发给上位机作为observation。先导入math、cv2绘图import math import cv2基础资源点绘制矿区绘制我们规定已探明矿区的颜色是浅蓝色的未探明矿区的颜色是灰色的。然后颜色会根据储量而线性变化。首先创建一张和原地图分辨率一致的空白图尺寸可以从GameInfo类获取# 1. 初始化空白地图 map np.zeros( (self.game_info.map_size[0], self.game_info.map_size[1], 3), dtypenp.uint8 )然后获取地图所有的矿点并设定基础颜色是黄色。分两种情况如果探明了就按剩余比例调整颜色如果没探明就保持灰色。# 2. 绘制矿产资源水晶矿 for mineral in self.mineral_field: pos mineral.position # 获取矿点的坐标(x,y) c [175, 255, 255] # 基础颜色青蓝色代表水晶矿 # 计算剩余矿量比例当前矿量 / 初始满矿量(2250) fraction mineral.mineral_contents / 2250 if mineral.is_visible: # 可见矿按剩余矿量比例调整颜色亮度矿越多越亮 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c] else: # 不可见/战争迷雾中的矿显示灰色代表未知 map[math.ceil(pos.y)][math.ceil(pos.x)] [50, 50, 50]最后显示地图注意星际 2 的坐标系 和 OpenCV 的坐标系 是完全相反的不翻转地图会上下颠倒。星际 2 游戏坐标系X 轴向右变大Y 轴向上变大原点 (0,0) 在左下角OpenCV 图像坐标系X 轴向右变大Y 轴向下变大原点 (0,0) 在左上角因此只需要把图片上下颠倒就行使用flip函数.如果此时你运行会发现图像很小都是一个pix看不清因此需要放大用resize。地图是每个矿 1 个像素每个基地 1 个像素每个虚空舰 1 个像素但是放大后会让信息点变成模糊的一坨在predator项目我们已经体会过了因此需要INTER_NEAREST最邻近插值保持 “像素块” 风格。# 3. 显示地图缩放翻转适配OpenCV显示 cv2.imshow( map, cv2.flip( cv2.resize( map, None, fx4, fy4, # 放大4倍方便观察 interpolationcv2.INTER_NEAREST # 最近邻插值保留像素块 ), 0 # 0上下翻转修正坐标系 ) )气源绘制类似矿区抄一遍改个名字即可#2:绘制瓦斯资源 for vespene in self.vespene_geyser: pos vespene.position # 获取气矿的坐标(x,y) c [255, 175, 255] # 基础颜色粉紫色代表气矿 # 计算剩余气矿量比例当前气矿量 / 初始满矿量(2250) fraction vespene.vespene_contents / 2250 if vespene.is_visible: # 可见气矿按剩余气矿量比例调整颜色亮度气越多越亮 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c] else: # 不可见/战争迷雾中的气矿显示灰色代表未知 map[math.ceil(pos.y)][math.ceil(pos.x)] [50, 50, 50]完整代码# 画图生成地图状态观测 # 1. 初始化空白地图 map np.zeros( (self.game_info.map_size[0], self.game_info.map_size[1], 3), dtypenp.uint8 ) # 2. 绘制矿产资源水晶矿 for mineral in self.mineral_field: pos mineral.position # 获取矿点的坐标(x,y) c [175, 255, 255] # 基础颜色青蓝色代表水晶矿 # 计算剩余矿量比例当前矿量 / 初始满矿量(2250) fraction mineral.mineral_contents / 2250 if mineral.is_visible: # 可见矿按剩余矿量比例调整颜色亮度矿越多越亮 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c] else: # 不可见/战争迷雾中的矿显示灰色代表未知 map[math.ceil(pos.y)][math.ceil(pos.x)] [50, 50, 50] #3:绘制瓦斯资源 for vespene in self.vespene_geyser: pos vespene.position # 获取气矿的坐标(x,y) c [255, 175, 255] # 基础颜色粉紫色代表气矿 # 计算剩余气矿量比例当前气矿量 / 初始满矿量(2250) fraction vespene.vespene_contents / 2250 if vespene.is_visible: # 可见气矿按剩余气矿量比例调整颜色亮度气越多越亮 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c] else: # 不可见/战争迷雾中的气矿显示灰色代表未知 map[math.ceil(pos.y)][math.ceil(pos.x)] [50, 50, 50] # 4. 显示地图缩放翻转适配OpenCV显示 cv2.imshow( map, cv2.flip( cv2.resize( map, None, fx4, fy4, # 放大3倍方便观察 interpolationcv2.INTER_NEAREST # 最近邻插值保留像素块 ), 0 # 0上下翻转修正坐标系 ) ) cv2.waitKey(1) # 等待1ms刷新窗口必须加否则窗口卡死跑一下测试可以看到气矿是粉红色晶矿是黄色我方基础设施绘制主基地使用浅蓝色色其他设施使用绿色同样根据其血量比例设定亮度。注意防止除零的情况加一个特判。#3:绘制基础设施 for structure in self.structures: pos structure.position # 获取建筑的坐标(x,y) # 区分建筑类型基地(nexus)用特殊颜色其他建筑用另一种颜色 if structure.type_id UnitTypeId.NEXUS: c [255, 255, 175] # 亮黄色代表基地/主基地 else: c [0, 255, 175] # 青绿色代表其他己方建筑如水晶塔、传送门、星门等 # 计算血量比例当前血量 / 最大血量避免除零 fraction structure.health_percentage # 按血量比例缩放颜色绘制到地图上 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c]运行一下中枢是浅蓝色。我方单位绘制我们主要使用虚空星舰作战因此赋予蓝色其他辅助单位PROBE使用亮绿色即可。#5:绘制我方单位 for unit in self.units: pos unit.position # 获取单位的坐标(x,y) # 区分单位类型虚空辉光舰用特殊蓝色其他单位用亮绿色 if unit.type_id UnitTypeId.VOIDRAY: c [255, 0, 0] # 蓝色代表核心作战单位虚空辉光舰 else: c [175, 255, 0] # 亮绿色代表其他己方单位探机等 # 直接获取血量百分比0~1无需手动计算 fraction unit.health_percentage # 按血量比例缩放颜色绘制到地图上 map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c]测试一下开局可以看到很多PROBE在主基地附近一段时间后动作4派出了侦察兵再过一段时间训练出了虚空星舰当虚空星舰超过阈值后执行进攻动作开团绘制敌方单位、敌方初始建筑、敌方基础设施逻辑和我方一样只是改成enemy#6:绘制敌人的起始位置出生点 for enemy_location in self.enemy_start_locations: pos enemy_location # 获取敌人出生点坐标 # 纯红色代表敌人老家显眼 c [0, 0, 255] # 直接赋值不需要遍历i map[math.ceil(pos.y)][math.ceil(pos.x)] c #7:绘制敌人的基础设施建筑 for structure in self.enemy_structures: pos structure.position # 亮红色代表敌人建筑 c [0, 100, 255] # 按血量比例缩放颜色满血最亮残血变暗 fraction structure.health_percentage map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c] #8:绘制敌人的单位兵力 for unit in self.enemy_units: pos unit.position # 橙红色代表敌人活跃的单位/部队 c [100, 0, 255] # 按血量比例缩放颜色 fraction unit.health_percentage map[math.ceil(pos.y)][math.ceil(pos.x)] [int(fraction * i) for i in c]运行一下开局我们的先遣队到敌方老巢就发现敌人了敌方第一波快攻激战成功击退我们的虚空星舰反攻敌方奖励设计攻击奖励我们规定当星舰攻击敌人时每一次给0.015的奖励。索敌条件正在攻击且八格范围内有敌人。# 计算奖励值 reward 0 # 初始化奖励为0 try: # 遍历所有己方的虚空辉光舰 for voidray in self.units(UnitTypeId.VOIDRAY): # 条件1虚空舰正在攻击且目标在攻击范围内有效攻击 if voidray.is_attacking and voidray.target_in_range: # 条件2虚空舰8格范围内有敌人单位/建筑在战场中不是空跑 if self.enemy_structures.closer_than(8, voidray) or self.enemy_units.closer_than(8, voidray): # 满足所有条件给奖励 reward 0.015 except Exception as e: # 捕获异常比如没有虚空舰、敌人不存在避免崩溃 print(freward error:{e}) reward 0 # 异常时奖励归零 # 每10帧打印一次日志方便调试 if iteration % 10 0: print(fiteration:{iteration},RW:{reward},VR:{self.units(UnitTypeId.VOIDRAY).amount})测试一下我方虚空星舰抵达战场激战后续援兵跟上敌方颓势明显大赢特赢我们查看输出日志RW值明显是增加的。全局胜负奖励胜利给500输了扣500。首先先把对局结果存起来resultrun_game(maps.get(2000AtmospheresAIE), [ Bot(Race.Protoss, WorkerRushBot()), Computer(Race.Zerg, Difficulty.Hard) ], realtimeFalse)然后存入到result.txt中。# 1. 记录比赛结果到日志文件 with open(result.txt, a) as f: f.write(f{result}\n)从之前的对局结果可以看到result对象的内容胜利时是Result.Victory因此作为判定条件# 2. 发放终局奖励/惩罚 if str(result) Result.Victory: print(Victory!) rwd 500 # 胜利给500大额奖励 else: rwd -500 # 失败/平局给-500大额惩罚最后一步传递的内容依然要通过transaction.pkl传递因此依然要填写transaction{observation:map,reward:0,action:None,terminated:False,truncated:False}最后一步不用传图了直接赋空图即可。reward就是刚刚得到的rwd。action省略。终止达成。结束后别忘了清理窗口。完整修改# 1. 记录比赛结果到日志文件 with open(result.txt, a) as f: f.write(f{result}\n) # 2. 发放终局奖励/惩罚 if str(result) Result.Victory: print(Victory!) rwd 500 # 胜利给500大额奖励 else: rwd -500 # 失败/平局给-500大额惩罚 # 3. 生成最终观测与交易数据保存为pkl文件 map np.zeros((224,224,3), dtype np.uint8) transaction {observation:map, reward:rwd, action:None, terminated:True,truncated:False} with open(transaction.pkl, wb) as f: pickle.dump(transaction, f) # 4. 清理OpenCV窗口避免残留 cv2.destroyAllWindows() cv2.waitKey(1) time.sleep(1)测试一下在后面先追加print(result)with open(transaction.pkl, rb) as f: transaction1pickle.load(f)print(transaction1)然后这次运行这个resultrun_game(maps.get(2000AtmospheresAIE), [ Bot(Race.Protoss, WorkerRushBot()), Computer(Race.Zerg, Difficulty.Hard) ], realtimeFalse)然后启动上位机for i in range(200): sc2.step(0) time.sleep(0.05) sc2.step(1) time.sleep(0.05) sc2.step(2) time.sleep(0.05) sc2.step(3) time.sleep(0.05) sc2.step(4)对线差一点对方也差不多挂了我们慢了一步打印一下result和最后一次transaction数据都对的上封装脚本然后我们把下位机的代码保存为“WorkerRushBot.py”路径和当前jupyter文件一致。有些许改动可以在文末copy这个.py文件的代码。