491 lines
21 KiB
Python
491 lines
21 KiB
Python
# pylint: disable=W0201,W0212,R0912
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import time
|
|
import warnings
|
|
from abc import ABC
|
|
from collections import Counter
|
|
from typing import TYPE_CHECKING, Any
|
|
from typing import Dict, Generator, Iterable, List, Set, Tuple, Union, final
|
|
|
|
from s2clientprotocol import sc2api_pb2 as sc_pb
|
|
|
|
from .constants import (
|
|
IS_PLACEHOLDER,
|
|
)
|
|
from .data import Race
|
|
from .game_data import GameData
|
|
from .game_state import Blip, GameState
|
|
from .pixel_map import PixelMap
|
|
from .position import Point2
|
|
from .unit import Unit
|
|
from .units import Units
|
|
|
|
# with warnings.catch_warnings():
|
|
# warnings.simplefilter("ignore")
|
|
# from scipy.spatial.distance import cdist, pdist
|
|
|
|
if TYPE_CHECKING:
|
|
from .client import Client
|
|
from .game_info import GameInfo
|
|
|
|
|
|
class BotAIInternal(ABC):
|
|
"""Base class for bots."""
|
|
|
|
@final
|
|
def _initialize_variables(self):
|
|
""" Called from main.py internally """
|
|
self.cache: Dict[str, Any] = {}
|
|
# Specific opponent bot ID used in sc2ai ladder games http://sc2ai.net/ and on ai arena https://aiarena.net
|
|
# The bot ID will stay the same each game so your bot can "adapt" to the opponent
|
|
if not hasattr(self, "opponent_id"):
|
|
# Prevent overwriting the opponent_id which is set here https://github.com/Hannessa/python-sc2-ladderbot/blob/master/__init__.py#L40
|
|
# otherwise set it to None
|
|
self.opponent_id: str = None
|
|
# Select distance calculation method, see _distances_override_functions function
|
|
if not hasattr(self, "distance_calculation_method"):
|
|
self.distance_calculation_method: int = 2
|
|
# Select if the Unit.command should return UnitCommand objects. Set this to True if your bot uses 'self.do(unit(ability, target))'
|
|
if not hasattr(self, "unit_command_uses_self_do"):
|
|
self.unit_command_uses_self_do: bool = False
|
|
# This value will be set to True by main.py in self._prepare_start if game is played in realtime (if true, the bot will have limited time per step)
|
|
self.realtime: bool = False
|
|
self.base_build: int = -1
|
|
self.all_units: Units = Units([], self)
|
|
self.units: Units = Units([], self)
|
|
self.workers: Units = Units([], self)
|
|
self.larva: Units = Units([], self)
|
|
self.structures: Units = Units([], self)
|
|
self.townhalls: Units = Units([], self)
|
|
self.gas_buildings: Units = Units([], self)
|
|
self.all_own_units: Units = Units([], self)
|
|
self.enemy_units: Units = Units([], self)
|
|
self.enemy_structures: Units = Units([], self)
|
|
self.all_enemy_units: Units = Units([], self)
|
|
self.resources: Units = Units([], self)
|
|
self.destructables: Units = Units([], self)
|
|
self.watchtowers: Units = Units([], self)
|
|
self.mineral_field: Units = Units([], self)
|
|
self.vespene_geyser: Units = Units([], self)
|
|
self.placeholders: Units = Units([], self)
|
|
self.techlab_tags: Set[int] = set()
|
|
self.reactor_tags: Set[int] = set()
|
|
self.minerals: int = 50
|
|
self.vespene: int = 0
|
|
self.supply_army: float = 0
|
|
self.supply_workers: float = 12 # Doesn't include workers in production
|
|
self.supply_cap: float = 15
|
|
self.supply_used: float = 12
|
|
self.supply_left: float = 3
|
|
self.idle_worker_count: int = 0
|
|
self.army_count: int = 0
|
|
self.warp_gate_count: int = 0
|
|
self.blips: Set[Blip] = set()
|
|
self.race: Race = None
|
|
self.enemy_race: Race = None
|
|
self._generated_frame = -100
|
|
self._units_created: Counter = Counter()
|
|
self._unit_tags_seen_this_game: Set[int] = set()
|
|
self._units_previous_map: Dict[int, Unit] = {}
|
|
self._structures_previous_map: Dict[int, Unit] = {}
|
|
self._enemy_units_previous_map: Dict[int, Unit] = {}
|
|
self._enemy_structures_previous_map: Dict[int, Unit] = {}
|
|
self._all_units_previous_map: Dict[int, Unit] = {}
|
|
self._expansion_positions_list: List[Point2] = []
|
|
self._resource_location_to_expansion_position_dict: Dict[Point2, Point2] = {}
|
|
self._time_before_step: float = None
|
|
self._time_after_step: float = None
|
|
self._min_step_time: float = math.inf
|
|
self._max_step_time: float = 0
|
|
self._last_step_step_time: float = 0
|
|
self._total_time_in_on_step: float = 0
|
|
self._total_steps_iterations: int = 0
|
|
# Internally used to keep track which units received an action in this frame, so that self.train() function does not give the same larva two orders - cleared every frame
|
|
self.unit_tags_received_action: Set[int] = set()
|
|
|
|
@final
|
|
@property
|
|
def _game_info(self) -> GameInfo:
|
|
""" See game_info.py """
|
|
warnings.warn(
|
|
"Using self._game_info is deprecated and may be removed soon. Please use self.game_info directly.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return self.game_info
|
|
|
|
@final
|
|
@property
|
|
def _game_data(self) -> GameData:
|
|
""" See game_data.py """
|
|
warnings.warn(
|
|
"Using self._game_data is deprecated and may be removed soon. Please use self.game_data directly.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return self.game_data
|
|
|
|
@final
|
|
@property
|
|
def _client(self) -> Client:
|
|
""" See client.py """
|
|
warnings.warn(
|
|
"Using self._client is deprecated and may be removed soon. Please use self.client directly.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return self.client
|
|
|
|
@final
|
|
def _prepare_start(self, client, player_id, game_info, game_data, realtime: bool = False, base_build: int = -1):
|
|
"""
|
|
Ran until game start to set game and player data.
|
|
|
|
:param client:
|
|
:param player_id:
|
|
:param game_info:
|
|
:param game_data:
|
|
:param realtime:
|
|
"""
|
|
self.client: Client = client
|
|
self.player_id: int = player_id
|
|
self.game_info: GameInfo = game_info
|
|
self.game_data: GameData = game_data
|
|
self.realtime: bool = realtime
|
|
self.base_build: int = base_build
|
|
|
|
self.race: Race = Race(self.game_info.player_races[self.player_id])
|
|
|
|
if len(self.game_info.player_races) == 2:
|
|
self.enemy_race: Race = Race(self.game_info.player_races[3 - self.player_id])
|
|
|
|
|
|
@final
|
|
def _prepare_first_step(self):
|
|
"""First step extra preparations. Must not be called before _prepare_step."""
|
|
if self.townhalls:
|
|
self.game_info.player_start_location = self.townhalls.first.position
|
|
# Calculate and cache expansion locations forever inside 'self._cache_expansion_locations', this is done to prevent a bug when this is run and cached later in the game
|
|
self._time_before_step: float = time.perf_counter()
|
|
|
|
@final
|
|
def _prepare_step(self, state, proto_game_info):
|
|
"""
|
|
:param state:
|
|
:param proto_game_info:
|
|
"""
|
|
# Set attributes from new state before on_step."""
|
|
self.state: GameState = state # See game_state.py
|
|
# update pathing grid, which unfortunately is in GameInfo instead of GameState
|
|
self.game_info.pathing_grid = PixelMap(proto_game_info.game_info.start_raw.pathing_grid, in_bits=True)
|
|
# Required for events, needs to be before self.units are initialized so the old units are stored
|
|
self._units_previous_map: Dict[int, Unit] = {unit.tag: unit for unit in self.units}
|
|
self._structures_previous_map: Dict[int, Unit] = {structure.tag: structure for structure in self.structures}
|
|
self._enemy_units_previous_map: Dict[int, Unit] = {unit.tag: unit for unit in self.enemy_units}
|
|
self._enemy_structures_previous_map: Dict[int, Unit] = {
|
|
structure.tag: structure
|
|
for structure in self.enemy_structures
|
|
}
|
|
self._all_units_previous_map: Dict[int, Unit] = {unit.tag: unit for unit in self.all_units}
|
|
|
|
self._prepare_units()
|
|
self.minerals: int = state.common.minerals
|
|
self.vespene: int = state.common.vespene
|
|
self.supply_army: int = state.common.food_army
|
|
self.supply_workers: int = state.common.food_workers # Doesn't include workers in production
|
|
self.supply_cap: int = state.common.food_cap
|
|
self.supply_used: int = state.common.food_used
|
|
self.supply_left: int = self.supply_cap - self.supply_used
|
|
|
|
if self.race == Race.Zerg:
|
|
# Workaround Zerg supply rounding bug
|
|
pass
|
|
# self._correct_zerg_supply()
|
|
elif self.race == Race.Protoss:
|
|
self.warp_gate_count: int = state.common.warp_gate_count
|
|
|
|
self.idle_worker_count: int = state.common.idle_worker_count
|
|
self.army_count: int = state.common.army_count
|
|
self._time_before_step: float = time.perf_counter()
|
|
|
|
if self.enemy_race == Race.Random and self.all_enemy_units:
|
|
self.enemy_race = Race(self.all_enemy_units.first.race)
|
|
|
|
@final
|
|
def _prepare_units(self):
|
|
# Set of enemy units detected by own sensor tower, as blips have less unit information than normal visible units
|
|
self.blips: Set[Blip] = set()
|
|
self.all_units: Units = Units([], self)
|
|
self.units: Units = Units([], self)
|
|
self.workers: Units = Units([], self)
|
|
self.larva: Units = Units([], self)
|
|
self.structures: Units = Units([], self)
|
|
self.townhalls: Units = Units([], self)
|
|
self.gas_buildings: Units = Units([], self)
|
|
self.all_own_units: Units = Units([], self)
|
|
self.enemy_units: Units = Units([], self)
|
|
self.enemy_structures: Units = Units([], self)
|
|
self.all_enemy_units: Units = Units([], self)
|
|
self.resources: Units = Units([], self)
|
|
self.destructables: Units = Units([], self)
|
|
self.watchtowers: Units = Units([], self)
|
|
self.mineral_field: Units = Units([], self)
|
|
self.vespene_geyser: Units = Units([], self)
|
|
self.placeholders: Units = Units([], self)
|
|
self.techlab_tags: Set[int] = set()
|
|
self.reactor_tags: Set[int] = set()
|
|
|
|
index: int = 0
|
|
for unit in self.state.observation_raw.units:
|
|
if unit.is_blip:
|
|
self.blips.add(Blip(unit))
|
|
else:
|
|
unit_type: int = unit.unit_type
|
|
# Convert these units to effects: reaper grenade, parasitic bomb dummy, forcefield
|
|
unit_obj = Unit(unit, self, distance_calculation_index=index, base_build=self.base_build)
|
|
index += 1
|
|
self.all_units.append(unit_obj)
|
|
if unit.display_type == IS_PLACEHOLDER:
|
|
self.placeholders.append(unit_obj)
|
|
continue
|
|
alliance = unit.alliance
|
|
# Alliance.Neutral.value = 3
|
|
if alliance == 3:
|
|
# XELNAGATOWER = 149
|
|
if unit_type == 149:
|
|
self.watchtowers.append(unit_obj)
|
|
# all destructable rocks
|
|
else:
|
|
self.destructables.append(unit_obj)
|
|
# Alliance.Self.value = 1
|
|
elif alliance == 1:
|
|
self.all_own_units.append(unit_obj)
|
|
if unit_obj.is_structure:
|
|
self.structures.append(unit_obj)
|
|
# Alliance.Enemy.value = 4
|
|
elif alliance == 4:
|
|
self.all_enemy_units.append(unit_obj)
|
|
if unit_obj.is_structure:
|
|
self.enemy_structures.append(unit_obj)
|
|
else:
|
|
self.enemy_units.append(unit_obj)
|
|
|
|
@final
|
|
async def _after_step(self) -> int:
|
|
""" Executed by main.py after each on_step function. """
|
|
# Keep track of the bot on_step duration
|
|
self._time_after_step: float = time.perf_counter()
|
|
step_duration = self._time_after_step - self._time_before_step
|
|
self._min_step_time = min(step_duration, self._min_step_time)
|
|
self._max_step_time = max(step_duration, self._max_step_time)
|
|
self._last_step_step_time = step_duration
|
|
self._total_time_in_on_step += step_duration
|
|
self._total_steps_iterations += 1
|
|
# Clear set of unit tags that were given an order this frame by self.do()
|
|
self.unit_tags_received_action.clear()
|
|
# Commit debug queries
|
|
await self.client._send_debug()
|
|
|
|
return self.state.game_loop
|
|
|
|
@final
|
|
async def _advance_steps(self, steps: int):
|
|
"""Advances the game loop by amount of 'steps'. This function is meant to be used as a debugging and testing tool only.
|
|
If you are using this, please be aware of the consequences, e.g. 'self.units' will be filled with completely new data."""
|
|
await self._after_step()
|
|
# Advance simulation by exactly "steps" frames
|
|
await self.client.step(steps)
|
|
state = await self.client.observation()
|
|
gs = GameState(state.observation)
|
|
proto_game_info = await self.client._execute(game_info=sc_pb.RequestGameInfo())
|
|
self._prepare_step(gs, proto_game_info)
|
|
await self.issue_events()
|
|
|
|
@final
|
|
async def issue_events(self):
|
|
"""This function will be automatically run from main.py and triggers the following functions:
|
|
- on_unit_created
|
|
- on_unit_destroyed
|
|
- on_building_construction_started
|
|
- on_building_construction_complete
|
|
- on_upgrade_complete
|
|
"""
|
|
await self._issue_unit_dead_events()
|
|
await self._issue_unit_added_events()
|
|
await self._issue_building_events()
|
|
await self._issue_upgrade_events()
|
|
await self._issue_vision_events()
|
|
|
|
@final
|
|
async def _issue_unit_added_events(self):
|
|
pass
|
|
# for unit in self.units:
|
|
# if unit.tag not in self._units_previous_map and unit.tag not in self._unit_tags_seen_this_game:
|
|
# self._unit_tags_seen_this_game.add(unit.tag)
|
|
# self._units_created[unit.type_id] += 1
|
|
# await self.on_unit_created(unit)
|
|
# elif unit.tag in self._units_previous_map:
|
|
# previous_frame_unit: Unit = self._units_previous_map[unit.tag]
|
|
# # Check if a unit took damage this frame and then trigger event
|
|
# if unit.health < previous_frame_unit.health or unit.shield < previous_frame_unit.shield:
|
|
# damage_amount = previous_frame_unit.health - unit.health + previous_frame_unit.shield - unit.shield
|
|
# await self.on_unit_took_damage(unit, damage_amount)
|
|
# # Check if a unit type has changed
|
|
# if previous_frame_unit.type_id != unit.type_id:
|
|
# await self.on_unit_type_changed(unit, previous_frame_unit.type_id)
|
|
|
|
@final
|
|
async def _issue_upgrade_events(self):
|
|
pass
|
|
# difference = self.state.upgrades - self._previous_upgrades
|
|
# for upgrade_completed in difference:
|
|
# await self.on_upgrade_complete(upgrade_completed)
|
|
# self._previous_upgrades = self.state.upgrades
|
|
|
|
@final
|
|
async def _issue_building_events(self):
|
|
pass
|
|
# for structure in self.structures:
|
|
# if structure.tag not in self._structures_previous_map:
|
|
# if structure.build_progress < 1:
|
|
# await self.on_building_construction_started(structure)
|
|
# else:
|
|
# # Include starting townhall
|
|
# self._units_created[structure.type_id] += 1
|
|
# await self.on_building_construction_complete(structure)
|
|
# elif structure.tag in self._structures_previous_map:
|
|
# # Check if a structure took damage this frame and then trigger event
|
|
# previous_frame_structure: Unit = self._structures_previous_map[structure.tag]
|
|
# if (
|
|
# structure.health < previous_frame_structure.health
|
|
# or structure.shield < previous_frame_structure.shield
|
|
# ):
|
|
# damage_amount = (
|
|
# previous_frame_structure.health - structure.health + previous_frame_structure.shield -
|
|
# structure.shield
|
|
# )
|
|
# await self.on_unit_took_damage(structure, damage_amount)
|
|
# # Check if a structure changed its type
|
|
# if previous_frame_structure.type_id != structure.type_id:
|
|
# await self.on_unit_type_changed(structure, previous_frame_structure.type_id)
|
|
# # Check if structure completed
|
|
# if structure.build_progress == 1 and previous_frame_structure.build_progress < 1:
|
|
# self._units_created[structure.type_id] += 1
|
|
# await self.on_building_construction_complete(structure)
|
|
|
|
@final
|
|
async def _issue_vision_events(self):
|
|
pass
|
|
# # Call events for enemy unit entered vision
|
|
# for enemy_unit in self.enemy_units:
|
|
# if enemy_unit.tag not in self._enemy_units_previous_map:
|
|
# await self.on_enemy_unit_entered_vision(enemy_unit)
|
|
# for enemy_structure in self.enemy_structures:
|
|
# if enemy_structure.tag not in self._enemy_structures_previous_map:
|
|
# await self.on_enemy_unit_entered_vision(enemy_structure)
|
|
|
|
# # Call events for enemy unit left vision
|
|
# enemy_units_left_vision: Set[int] = set(self._enemy_units_previous_map) - self.enemy_units.tags
|
|
# for enemy_unit_tag in enemy_units_left_vision:
|
|
# await self.on_enemy_unit_left_vision(enemy_unit_tag)
|
|
# enemy_structures_left_vision: Set[int] = (set(self._enemy_structures_previous_map) - self.enemy_structures.tags)
|
|
# for enemy_structure_tag in enemy_structures_left_vision:
|
|
# await self.on_enemy_unit_left_vision(enemy_structure_tag)
|
|
|
|
@final
|
|
async def _issue_unit_dead_events(self):
|
|
pass
|
|
# for unit_tag in self.state.dead_units & set(self._all_units_previous_map):
|
|
# await self.on_unit_destroyed(unit_tag)
|
|
|
|
# DISTANCE CALCULATION
|
|
|
|
@final
|
|
@property
|
|
def _units_count(self) -> int:
|
|
return len(self.all_units)
|
|
|
|
# Helper functions
|
|
|
|
@final
|
|
def square_to_condensed(self, i, j) -> int:
|
|
# Converts indices of a square matrix to condensed matrix
|
|
# https://stackoverflow.com/a/36867493/10882657
|
|
assert i != j, "No diagonal elements in condensed matrix! Diagonal elements are zero"
|
|
if i < j:
|
|
i, j = j, i
|
|
return self._units_count * j - j * (j + 1) // 2 + i - 1 - j
|
|
|
|
# Fast and simple calculation functions
|
|
|
|
@final
|
|
@staticmethod
|
|
def distance_math_hypot(
|
|
p1: Union[Tuple[float, float], Point2],
|
|
p2: Union[Tuple[float, float], Point2],
|
|
) -> float:
|
|
return math.hypot(p1[0] - p2[0], p1[1] - p2[1])
|
|
|
|
@final
|
|
@staticmethod
|
|
def distance_math_hypot_squared(
|
|
p1: Union[Tuple[float, float], Point2],
|
|
p2: Union[Tuple[float, float], Point2],
|
|
) -> float:
|
|
return pow(p1[0] - p2[0], 2) + pow(p1[1] - p2[1], 2)
|
|
|
|
@final
|
|
def _distance_squared_unit_to_unit_method0(self, unit1: Unit, unit2: Unit) -> float:
|
|
return self.distance_math_hypot_squared(unit1.position_tuple, unit2.position_tuple)
|
|
|
|
# Distance calculation using the pre-calculated matrix above
|
|
|
|
@final
|
|
def _distance_squared_unit_to_unit_method1(self, unit1: Unit, unit2: Unit) -> float:
|
|
# If checked on units if they have the same tag, return distance 0 as these are not in the 1 dimensional pdist array - would result in an error otherwise
|
|
if unit1.tag == unit2.tag:
|
|
return 0
|
|
# Calculate index, needs to be after pdist has been calculated and cached
|
|
condensed_index = self.square_to_condensed(unit1.distance_calculation_index, unit2.distance_calculation_index)
|
|
assert condensed_index < len(
|
|
self._cached_pdist
|
|
), f"Condensed index is larger than amount of calculated distances: {condensed_index} < {len(self._cached_pdist)}, units that caused the assert error: {unit1} and {unit2}"
|
|
distance = self._pdist[condensed_index]
|
|
return distance
|
|
|
|
@final
|
|
def _distance_squared_unit_to_unit_method2(self, unit1: Unit, unit2: Unit) -> float:
|
|
# Calculate index, needs to be after cdist has been calculated and cached
|
|
return self._cdist[unit1.distance_calculation_index, unit2.distance_calculation_index]
|
|
|
|
# Distance calculation using the fastest distance calculation functions
|
|
|
|
@final
|
|
def _distance_pos_to_pos(
|
|
self,
|
|
pos1: Union[Tuple[float, float], Point2],
|
|
pos2: Union[Tuple[float, float], Point2],
|
|
) -> float:
|
|
return self.distance_math_hypot(pos1, pos2)
|
|
|
|
@final
|
|
def _distance_units_to_pos(
|
|
self,
|
|
units: Units,
|
|
pos: Union[Tuple[float, float], Point2],
|
|
) -> Generator[float, None, None]:
|
|
""" This function does not scale well, if len(units) > 100 it gets fairly slow """
|
|
return (self.distance_math_hypot(u.position_tuple, pos) for u in units)
|
|
|
|
@final
|
|
def _distance_unit_to_points(
|
|
self,
|
|
unit: Unit,
|
|
points: Iterable[Tuple[float, float]],
|
|
) -> Generator[float, None, None]:
|
|
""" This function does not scale well, if len(points) > 100 it gets fairly slow """
|
|
pos = unit.position_tuple
|
|
return (self.distance_math_hypot(p, pos) for p in points)
|