Предыдущая статья — Python AI в StarCraft II. Часть XIV: совершенствуем разведку.

Добро пожаловать в пятнадцатую часть серии статей про использование искусственного интеллекта в игре Starcraft II. Здесь мы увеличим возможности выбора для нашего AI.

Начнем с того, что добавим в метод __init__ следующий код:

        self.choices = {0: self.build_scout,                         1: self.build_zealot,                         2: self.build_gateway,                         3: self.build_voidray,                         4: self.build_stalker,                         5: self.build_worker,                         6: self.build_assimilator,                         7: self.build_stargate,                         8: self.build_pylon,                         9: self.defend_nexus,                         10: self.attack_known_enemy_unit,                         11: self.attack_known_enemy_structure,                         12: self.expand,                         13: self.do_nothing,                         }

Теперь изменим метод on_step:

    async def on_step(self, iteration):          self.time = (self.state.game_loop/22.4) / 60         #print('Time:',self.time)         await self.distribute_workers()         await self.scout()         await self.intel()         await self.do_something()

Далее нам нужно создать метод self.do_something():

    async def do_something(self):          if self.time > self.do_something_after:             if self.use_model:                 prediction = self.model.predict([self.flipped.reshape([-1, 176, 200, 3])])                 choice = np.argmax(prediction[0])             else:                 choice = random.randrange(0, 14)             try:                 await self.choices[choice]()             except Exception as e:                 print(str(e))             y = np.zeros(14)             y[choice] = 1             self.train_data.append([y, self.flipped])

Легче легкого!

Что ж, теперь убедимся, что у нас есть все необходимые методы.

Больше материалов по машинному обучению

Подписывайтесь на нас в Телеграм

Подписаться ×

У нас уже должен быть метод для создания разведчика:

    async def build_scout(self):         for rf in self.units(ROBOTICSFACILITY).ready.noqueue:             print(len(self.units(OBSERVER)), self.time/3)             if self.can_afford(OBSERVER) and self.supply_left > 0:                 await self.do(rf.train(OBSERVER))                 break

Далее создадим Зелот:

    async def build_zealot(self):         gateways = self.units(GATEWAY).ready         if gateways.exists:             if self.can_afford(ZEALOT):                 await self.do(random.choice(gateways).train(ZEALOT))

Затем Врата:

    async def build_gateway(self):         pylon = self.units(PYLON).ready.random         if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):             await self.build(GATEWAY, near=pylon)

Луч Бездны:

    async def build_voidray(self):         stargates = self.units(STARGATE).ready         if stargates.exists:             if self.can_afford(VOIDRAY):                 await self.do(random.choice(stargates).train(VOIDRAY))

Сталкер:

    async def build_stalker(self):         pylon = self.units(PYLON).ready.random         gateways = self.units(GATEWAY).ready         cybernetics_cores = self.units(CYBERNETICSCORE).ready          if gateways.exists and cybernetics_cores.exists:             if self.can_afford(STALKER):                 await self.do(random.choice(gateways).train(STALKER))          if not cybernetics_cores.exists:             if self.units(GATEWAY).ready.exists:                 if self.can_afford(CYBERNETICSCORE) and not self.already_pending(CYBERNETICSCORE):                     await self.build(CYBERNETICSCORE, near=pylon)

Рабочий:

    async def build_worker(self):         nexuses = self.units(NEXUS).ready         if nexuses.exists:             if self.can_afford(PROBE):                 await self.do(random.choice(nexuses).train(PROBE))

Ассимилятор:

    async def build_assimilator(self):         for nexus in self.units(NEXUS).ready:             vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)             for vaspene in vaspenes:                 if not self.can_afford(ASSIMILATOR):                     break                 worker = self.select_build_worker(vaspene.position)                 if worker is None:                     break                 if not self.units(ASSIMILATOR).closer_than(1.0, vaspene).exists:                     await self.do(worker.build(ASSIMILATOR, vaspene))

Звездные Врата:

    async def build_stargate(self):         if self.units(PYLON).ready.exists:             pylon = self.units(PYLON).ready.random             if self.units(CYBERNETICSCORE).ready.exists:                 if self.can_afford(STARGATE) and not self.already_pending(STARGATE):                     await self.build(STARGATE, near=pylon)

Пилоны:

    async def build_pylon(self):             nexuses = self.units(NEXUS).ready             if nexuses.exists:                 if self.can_afford(PYLON):                     await self.build(PYLON, near=self.units(NEXUS).first.position.towards(self.game_info.map_center, 5))

Обратите внимание, что код для Пилонов мы немного изменили, чтобы избежать засорения ресурсных областей.

Расширение:

    async def expand(self):         try:             if self.can_afford(NEXUS):                 await self.expand_now()         except Exception as e:             print(str(e))

Наши четыре варианта атаки:

    async def do_nothing(self):         wait = random.randrange(7, 100)/100         self.do_something_after = self.time + wait      async def defend_nexus(self):         if len(self.known_enemy_units) > 0:             target = self.known_enemy_units.closest_to(random.choice(self.units(NEXUS)))             for u in self.units(VOIDRAY).idle:                 await self.do(u.attack(target))             for u in self.units(STALKER).idle:                 await self.do(u.attack(target))             for u in self.units(ZEALOT).idle:                 await self.do(u.attack(target))      async def attack_known_enemy_structure(self):         if len(self.known_enemy_structures) > 0:             target = random.choice(self.known_enemy_structures)             for u in self.units(VOIDRAY).idle:                 await self.do(u.attack(target))             for u in self.units(STALKER).idle:                 await self.do(u.attack(target))             for u in self.units(ZEALOT).idle:                 await self.do(u.attack(target))      async def attack_known_enemy_unit(self):         if len(self.known_enemy_units) > 0:             target = self.known_enemy_units.closest_to(random.choice(self.units(NEXUS)))             for u in self.units(VOIDRAY).idle:                 await self.do(u.attack(target))             for u in self.units(STALKER).idle:                 await self.do(u.attack(target))             for u in self.units(ZEALOT).idle:                 await self.do(u.attack(target))

Мы также внесли изменения, чтобы AI было легче соревноваться с самим собой:

    def __init__(self, use_model=False, title=1):         self.MAX_WORKERS = 50         self.do_something_after = 0         self.use_model = use_model         self.title = title         ...

Добавим self.title в метод intel:

        if not HEADLESS:             cv2.imshow(str(self.title), resized)             cv2.waitKey(1)

Весь код теперь принял следующий вид:

import sc2 from sc2 import run_game, maps, Race, Difficulty, Result from sc2.player import Bot, Computer from sc2 import position from sc2.constants import NEXUS, PROBE, PYLON, ASSIMILATOR, GATEWAY, CYBERNETICSCORE, STARGATE, VOIDRAY, SCV, DRONE, ROBOTICSFACILITY, OBSERVER, ZEALOT, STALKER import random import cv2 import numpy as np import os import time import math #import keras #os.environ["SC2PATH"] = '/starcraftstuff/StarCraftII/' HEADLESS = False class SentdeBot(sc2.BotAI): def __init__(self, use_model=False, title=1): self.MAX_WORKERS = 50 self.do_something_after = 0 self.use_model = use_model self.title = title self.scouts_and_spots = {} # ADDED THE CHOICES # self.choices = {0: self.build_scout, 1: self.build_zealot, 2: self.build_gateway, 3: self.build_voidray, 4: self.build_stalker, 5: self.build_worker, 6: self.build_assimilator, 7: self.build_stargate, 8: self.build_pylon, 9: self.defend_nexus, 10: self.attack_known_enemy_unit, 11: self.attack_known_enemy_structure, 12: self.expand, 13: self.do_nothing, } self.train_data = [] if self.use_model: print("USING MODEL!") self.model = keras.models.load_model("BasicCNN-30-epochs-0.0001-LR-4.2") def on_end(self, game_result): print('--- on_end called ---') print(game_result, self.use_model) with open("gameout-random-vs-medium.txt","a") as f: if self.use_model: f.write("Model {} - {}n".format(game_result, int(time.time()))) else: f.write("Random {} - {}n".format(game_result, int(time.time()))) async def on_step(self, iteration): self.time = (self.state.game_loop/22.4) / 60 #print('Time:',self.time) await self.distribute_workers() await self.scout() await self.intel() await self.do_something() def random_location_variance(self, location): x = location[0] y = location[1] # FIXED THIS x += random.randrange(-5,5) y += random.randrange(-5,5) if x < 0: print("x below") x = 0 if y < 0: print("y below") y = 0 if x > self.game_info.map_size[0]: print("x above") x = self.game_info.map_size[0] if y > self.game_info.map_size[1]: print("y above") y = self.game_info.map_size[1] go_to = position.Point2(position.Pointlike((x,y))) return go_to async def scout(self): ''' ['__call__', '__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', '_game_data', '_proto', '_type_data', 'add_on_tag', 'alliance', 'assigned_harvesters', 'attack', 'build', 'build_progress', 'cloak', 'detect_range', 'distance_to', 'energy', 'facing', 'gather', 'has_add_on', 'has_buff', 'health', 'health_max', 'hold_position', 'ideal_harvesters', 'is_blip', 'is_burrowed', 'is_enemy', 'is_flying', 'is_idle', 'is_mine', 'is_mineral_field', 'is_powered', 'is_ready', 'is_selected', 'is_snapshot', 'is_structure', 'is_vespene_geyser', 'is_visible', 'mineral_contents', 'move', 'name', 'noqueue', 'orders', 'owner_id', 'position', 'radar_range', 'radius', 'return_resource', 'shield', 'shield_max', 'stop', 'tag', 'train', 'type_id', 'vespene_contents', 'warp_in'] ''' self.expand_dis_dir = {} for el in self.expansion_locations: distance_to_enemy_start = el.distance_to(self.enemy_start_locations[0]) #print(distance_to_enemy_start) self.expand_dis_dir[distance_to_enemy_start] = el self.ordered_exp_distances = sorted(k for k in self.expand_dis_dir) existing_ids = [unit.tag for unit in self.units] # removing of scouts that are actually dead now. to_be_removed = [] for noted_scout in self.scouts_and_spots: if noted_scout not in existing_ids: to_be_removed.append(noted_scout) for scout in to_be_removed: del self.scouts_and_spots[scout] if len(self.units(ROBOTICSFACILITY).ready) == 0: unit_type = PROBE unit_limit = 1 else: unit_type = OBSERVER unit_limit = 15 assign_scout = True if unit_type == PROBE: for unit in self.units(PROBE): if unit.tag in self.scouts_and_spots: assign_scout = False if assign_scout: if len(self.units(unit_type).idle) > 0: for obs in self.units(unit_type).idle[:unit_limit]: if obs.tag not in self.scouts_and_spots: for dist in self.ordered_exp_distances: try: location = next(value for key, value in self.expand_dis_dir.items() if key == dist) # DICT {UNIT_ID:LOCATION} active_locations = [self.scouts_and_spots[k] for k in self.scouts_and_spots] if location not in active_locations: if unit_type == PROBE: for unit in self.units(PROBE): if unit.tag in self.scouts_and_spots: continue await self.do(obs.move(location)) self.scouts_and_spots[obs.tag] = location break except Exception as e: pass for obs in self.units(unit_type): if obs.tag in self.scouts_and_spots: if obs in [probe for probe in self.units(PROBE)]: await self.do(obs.move(self.random_location_variance(self.scouts_and_spots[obs.tag]))) async def intel(self): game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8) draw_dict = { NEXUS: [15, (0, 255, 0)], PYLON: [3, (20, 235, 0)], PROBE: [1, (55, 200, 0)], ASSIMILATOR: [2, (55, 200, 0)], GATEWAY: [3, (200, 100, 0)], CYBERNETICSCORE: [3, (150, 150, 0)], STARGATE: [5, (255, 0, 0)], ROBOTICSFACILITY: [5, (215, 155, 0)], #VOIDRAY: [3, (255, 100, 0)], } for unit_type in draw_dict: for unit in self.units(unit_type).ready: pos = unit.position cv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1) # from Александр Тимофеев via YT main_base_names = ['nexus', 'commandcenter', 'orbitalcommand', 'planetaryfortress', '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.position if 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 worker pos = enemy_unit.position if 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(OBSERVER).ready: pos = obs.position cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (255, 255, 255), -1) for vr in self.units(VOIDRAY).ready: pos = vr.position cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (255, 100, 0), -1) line_max = 50 mineral_ratio = self.minerals / 1500 if mineral_ratio > 1.0: mineral_ratio = 1.0 vespene_ratio = self.vespene / 1500 if vespene_ratio > 1.0: vespene_ratio = 1.0 population_ratio = self.supply_left / self.supply_cap if population_ratio > 1.0: population_ratio = 1.0 plausible_supply = self.supply_cap / 200.0 worker_weight = len(self.units(PROBE)) / (self.supply_cap-self.supply_left) if worker_weight > 1.0: worker_weight = 1.0 cv2.line(game_data, (0, 19), (int(line_max*worker_weight), 19), (250, 250, 200), 3) # worker/supply ratio cv2.line(game_data, (0, 15), (int(line_max*plausible_supply), 15), (220, 200, 200), 3) # plausible supply (supply/200.0) cv2.line(game_data, (0, 11), (int(line_max*population_ratio), 11), (150, 150, 150), 3) # population ratio (supply_left/supply) cv2.line(game_data, (0, 7), (int(line_max*vespene_ratio), 7), (210, 200, 0), 3) # gas / 1500 cv2.line(game_data, (0, 3), (int(line_max*mineral_ratio), 3), (0, 255, 25), 3) # minerals minerals/1500 # flip horizontally to make our final fix in visual representation: self.flipped = cv2.flip(game_data, 0) resized = cv2.resize(self.flipped, dsize=None, fx=2, fy=2) if not HEADLESS: cv2.imshow(str(self.title), resized) cv2.waitKey(1) def find_target(self, state): if len(self.known_enemy_units) > 0: return random.choice(self.known_enemy_units) elif len(self.known_enemy_structures) > 0: return random.choice(self.known_enemy_structures) else: return self.enemy_start_locations[0] async def build_scout(self): for rf in self.units(ROBOTICSFACILITY).ready.noqueue: print(len(self.units(OBSERVER)), self.time/3) if self.can_afford(OBSERVER) and self.supply_left > 0: await self.do(rf.train(OBSERVER)) break async def build_worker(self): nexuses = self.units(NEXUS).ready if nexuses.exists: if self.can_afford(PROBE): await self.do(random.choice(nexuses).train(PROBE)) async def build_zealot(self): gateways = self.units(GATEWAY).ready if gateways.exists: if self.can_afford(ZEALOT): await self.do(random.choice(gateways).train(ZEALOT)) async def build_gateway(self): pylon = self.units(PYLON).ready.random if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY): await self.build(GATEWAY, near=pylon) async def build_voidray(self): stargates = self.units(STARGATE).ready if stargates.exists: if self.can_afford(VOIDRAY): await self.do(random.choice(stargates).train(VOIDRAY)) async def build_stalker(self): pylon = self.units(PYLON).ready.random gateways = self.units(GATEWAY).ready cybernetics_cores = self.units(CYBERNETICSCORE).ready if gateways.exists and cybernetics_cores.exists: if self.can_afford(STALKER): await self.do(random.choice(gateways).train(STALKER)) if not cybernetics_cores.exists: if self.units(GATEWAY).ready.exists: if self.can_afford(CYBERNETICSCORE) and not self.already_pending(CYBERNETICSCORE): await self.build(CYBERNETICSCORE, near=pylon) async def build_assimilator(self): for nexus in self.units(NEXUS).ready: vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus) for vaspene in vaspenes: if not self.can_afford(ASSIMILATOR): break worker = self.select_build_worker(vaspene.position) if worker is None: break if not self.units(ASSIMILATOR).closer_than(1.0, vaspene).exists: await self.do(worker.build(ASSIMILATOR, vaspene)) async def build_stargate(self): if self.units(PYLON).ready.exists: pylon = self.units(PYLON).ready.random if self.units(CYBERNETICSCORE).ready.exists: if self.can_afford(STARGATE) and not self.already_pending(STARGATE): await self.build(STARGATE, near=pylon) async def build_pylon(self): nexuses = self.units(NEXUS).ready if nexuses.exists: if self.can_afford(PYLON): await self.build(PYLON, near=self.units(NEXUS).first.position.towards(self.game_info.map_center, 5)) async def expand(self): try: if self.can_afford(NEXUS): await self.expand_now() except Exception as e: print(str(e)) async def do_nothing(self): wait = random.randrange(7, 100)/100 self.do_something_after = self.time + wait async def defend_nexus(self): if len(self.known_enemy_units) > 0: target = self.known_enemy_units.closest_to(random.choice(self.units(NEXUS))) for u in self.units(VOIDRAY).idle: await self.do(u.attack(target)) for u in self.units(STALKER).idle: await self.do(u.attack(target)) for u in self.units(ZEALOT).idle: await self.do(u.attack(target)) async def attack_known_enemy_structure(self): if len(self.known_enemy_structures) > 0: target = random.choice(self.known_enemy_structures) for u in self.units(VOIDRAY).idle: await self.do(u.attack(target)) for u in self.units(STALKER).idle: await self.do(u.attack(target)) for u in self.units(ZEALOT).idle: await self.do(u.attack(target)) async def attack_known_enemy_unit(self): if len(self.known_enemy_units) > 0: target = self.known_enemy_units.closest_to(random.choice(self.units(NEXUS))) for u in self.units(VOIDRAY).idle: await self.do(u.attack(target)) for u in self.units(STALKER).idle: await self.do(u.attack(target)) for u in self.units(ZEALOT).idle: await self.do(u.attack(target)) async def do_something(self): if self.time > self.do_something_after: if self.use_model: prediction = self.model.predict([self.flipped.reshape([-1, 176, 200, 3])]) choice = np.argmax(prediction[0]) else: choice = random.randrange(0, 14) try: await self.choices[choice]() except Exception as e: print(str(e)) ###### NEW CHOICE HANDLING HERE ######### ###### NEW CHOICE HANDLING HERE ######### y = np.zeros(14) y[choice] = 1 self.train_data.append([y, self.flipped]) if True: run_game(maps.get("AbyssalReefLE"), [ Bot(Race.Protoss, SentdeBot(use_model=False, title=1)), #Bot(Race.Protoss, SentdeBot(use_model=False, title=2)), Computer(Race.Protoss, Difficulty.Medium), ], realtime=False)

Следующая статья — Python AI в StarCraft II. Часть XVI: изменение визуализации.

Больше материалов по машинному обучению

Подписывайтесь на нас в Телеграм

Подписаться ×