started implementing production buffer

bot/HarstemsAunt/army_group.py

@@ -145,48 +145,57 @@ def _runby_request():
                 if self.bot.structures.filter(lambda struct: struct.type_id in [UnitTypeId.GATEWAY, UnitTypeId.WARPGATE]):
                     if self.bot.units(UnitTypeId.DARKSHRINE).ready and not opponent_has_detection:
                         requested_unit: UnitTypeId = UnitTypeId.DARKTEMPLAR
-                        self.requested_units.append(requested_unit)
+                        if not requested_unit in self.requested_units:
+                            self.requested_units.append(requested_unit)
                         return
                     requested_unit: UnitTypeId = UnitTypeId.ZEALOT
-                    self.requested_units.append(requested_unit)
+                    if not requested_unit in self.requested_units:
+                        self.requested_units.append(requested_unit)
                     return
 
         def _army_request():
             """ request unit for "normal" Army group  """
             gate_aliases:list = [UnitTypeId.GATEWAY, UnitTypeId.WARPGATE]
-            if self.bot.structures.filter(lambda struct: struct.type_id in gate_aliases).idle:
+            if self.bot.structures.filter(lambda struct: struct.type_id in gate_aliases):
                 stalkers:int = len(self.units(UnitTypeId.STALKER))
                 zealots:int = len(self.units(UnitTypeId.ZEALOT)) +1
 
-                if self.bot.structures(UnitTypeId.TEMPLARARCHIVE) and len(self.units(UnitTypeId.HIGHTEMPLAR)) > 2:
+                if self.bot.structures(UnitTypeId.TEMPLARARCHIVE) and len(self.units(UnitTypeId.HIGHTEMPLAR)) < 2:
                     requested_unit: UnitTypeId = UnitTypeId.HIGHTEMPLAR
-                    self.requested_units.append(requested_unit)
-                    return
+                    if not requested_unit in self.requested_units:
+                        self.requested_units.append(requested_unit)
 
-                if not stalkers or stalkers/zealots < 3:
-                    requested_unit: UnitTypeId = UnitTypeId.STALKER
-                else:
+                if not zealots < stalkers:
                     requested_unit: UnitTypeId = UnitTypeId.ZEALOT
+                    if not requested_unit in self.requested_units:
+                        self.requested_units.append(requested_unit)
+                else:
+                    requested_unit: UnitTypeId = UnitTypeId.STALKER
+                    if not requested_unit in self.requested_units:
+                        self.requested_units.append(requested_unit)
 
-            if self.bot.structures(UnitTypeId.ROBOTICSFACILITY).idle:
+            if self.bot.structures(UnitTypeId.ROBOTICSFACILITY):
                 if not self.has_detection:
                     requested_unit: UnitTypeId = UnitTypeId.OBSERVER
                 else:
                     requested_unit: UnitTypeId = UnitTypeId.IMMORTAL
-                self.requested_units.append(requested_unit)
+                if not requested_unit in self.requested_units:
+                    self.requested_units.append(requested_unit)
+
 
-            if self.bot.structures(UnitTypeId.STARGATE).idle:
+            if self.bot.structures(UnitTypeId.STARGATE):
                 if len(self.bot.units.flying) > \
                     len(self.bot.enemy_units.filter(lambda unit: unit.is_flying and unit.can_attack)):
                     requested_unit: UnitTypeId = UnitTypeId.PHOENIX
-                    self.requested_units.append(requested_unit)
+                    if not requested_unit in self.requested_units:
+                        self.requested_units.append(requested_unit)
 
-        match self.GroupTypeId:
-            case GroupTypeId.RUN_BY:
-                logger.info("not Testing Runbys right now")
-                _runby_request()
-            case GroupTypeId.ARMY:
-                _army_request()
+        #match self.GroupTypeId:
+            #case GroupTypeId.RUN_BY:
+             #   logger.info("not Testing Runbys right now")
+#                _runby_request()
+            #case GroupTypeId.ARMY:
+        _army_request()
 
     def remove_unit(self, unit_tag:str) -> bool:
         """ Removes are unit from ArmyGroup 
@@ -285,7 +294,7 @@ def retreat(self) -> None:
     def regroup(self) -> None:
         """ regroup command for the group  """
 
-        if not self.units.filter(lambda unit: unit.distance_to(self.position) > 15):
+        if not self.units.filter(lambda unit: unit.distance_to(self.position) > 5):
             return
 
         self.units.furthest_to(self.position).move(self.position)

bot/HarstemsAunt/build_order.py

@@ -1,4 +1,6 @@
 """DOCSTRING to shut up the Linter """
+from __future__ import annotations
+
 from enum import Enum
 from typing import Union, List
 from functools import cached_property
@@ -14,7 +16,6 @@
 from .army_group import ArmyGroup
 from .common import ALL_STRUCTURES,INITIAL_TECH,DT_TIMING,logger
 
-
 class InstructionType(Enum):
     """Enumeration containing InstructionTypes """
     UNIT_PRODUCTION = 1
@@ -30,7 +31,7 @@ class BuildInstruction:
     """ class representing a Build Instruction """
 
     def __new__(cls,type_id:UnitTypeId,position:Union[Point2,Point3,Unit]=None,\
-        accuracy:int=0, worker_command:UnitCommand=None):
+        accuracy:int=0, worker_command:UnitCommand=None) -> BuildInstruction:
         """ Creates new instance of BuildInstruction
 
         Args:
@@ -121,11 +122,11 @@ def instruction_list(self) -> List[BuildInstruction]:
             BuildInstruction(UnitTypeId.PYLON,wall_pylon_pos),
             BuildInstruction(UnitTypeId.GATEWAY,wall_buildings[0]),
             BuildInstruction(UnitTypeId.ASSIMILATOR,vespene_position_0),
+            BuildInstruction(UnitTypeId.PYLON,tech_pylon_pos),
             BuildInstruction(UnitTypeId.ASSIMILATOR,vespene_position_1),
             BuildInstruction(UnitTypeId.CYBERNETICSCORE,wall_buildings[1]),
             BuildInstruction(UnitTypeId.NEXUS, start_pos),
             BuildInstruction(UnitTypeId.GATEWAY, wall_pylon_pos,5),
-            BuildInstruction(UnitTypeId.PYLON,tech_pylon_pos),
             BuildInstruction(UnitTypeId.STALKER, start_pos),
             BuildInstruction(UnitTypeId.STALKER, start_pos),
             BuildInstruction(UnitTypeId.PYLON, angle_pylon_pos, 1),

bot/HarstemsAunt/common.py

@@ -23,7 +23,8 @@
 
 INITIAL_TECH:Dict = {
     Race.Protoss: [UnitTypeId.TWILIGHTCOUNCIL, UnitTypeId.ROBOTICSFACILITY],
-    Race.Terran: [UnitTypeId.TWILIGHTCOUNCIL, UnitTypeId.TEMPLARARCHIVE],
+    Race.Terran: [UnitTypeId.TWILIGHTCOUNCIL, UnitTypeId.ROBOTICSFACILITY],
+            #[UnitTypeId.TWILIGHTCOUNCIL, UnitTypeId.TEMPLARARCHIVE], -> Terran tech change back later
     Race.Zerg: [UnitTypeId.TWILIGHTCOUNCIL, UnitTypeId.ROBOTICSFACILITY]
 }
 

bot/HarstemsAunt/macro.py

@@ -17,25 +17,6 @@
 from .build_order import BuildOrder, BuildInstruction, InstructionType
 
 
-async def warp_in_unit(bot: BotAI,unit:UnitTypeId,\
-    warp_in_position:Union[Point2,Point3,Unit]) -> bool:
-    pos:Point2= warp_in_position.position.to2.random_on_distance(4)
-    placement = await bot.find_placement(AbilityId.WARPGATETRAIN_STALKER, pos, placement_step=1)
-
-    for gate in bot.structures(UnitTypeId.WARPGATE).idle:
-        if Utils.can_build_unit(bot, unit):
-            gate.warp_in(unit, placement)
-
-async def build_gateway_units(bot:BotAI,unit_type:UnitTypeId):
-    gate_aliases:list = [UnitTypeId.GATEWAY, UnitTypeId.WARPGATE]
-    if Utils.can_build_unit(bot, unit_type):
-        for gate in bot.structures.filter(lambda struct: struct.type_id in gate_aliases):
-            if gate.is_idle and UpgradeId.WARPGATERESEARCH not in bot.researched:
-                gate.train(unit_type)
-            else:
-                warp_in_pos = Utils.get_warp_in_pos(bot)
-                await warp_in_unit(bot, unit_type, warp_in_pos)
-
 class Macro:
     """ Class handling the Marco aspect of the Game """
     def __init__(self,bot:BotAI) -> None:
@@ -134,18 +115,40 @@ async def train_unit(next_step:BuildInstruction):
             Args:
                 next_step (BuildInstruction): next instruction
             """
+
+            async def warp_in_unit(bot: BotAI,unit:UnitTypeId,\
+                warp_in_position:Union[Point2,Point3,Unit]) -> bool:
+                pos:Point2= warp_in_position.position.to2.random_on_distance(4)
+                placement = await bot.find_placement(AbilityId.WARPGATETRAIN_STALKER, pos, placement_step=1)
+
+                for gate in bot.structures(UnitTypeId.WARPGATE).idle:
+                    if Utils.can_build_unit(bot, unit):
+                         gate.warp_in(unit, placement)
+
+            async def build_gateway_units(bot:BotAI,unit_type:UnitTypeId):
+                gate_aliases:list = [UnitTypeId.GATEWAY, UnitTypeId.WARPGATE]
+                if Utils.can_build_unit(bot, unit_type):
+                    for gate in bot.structures.filter(lambda struct: struct.type_id in gate_aliases):
+                        if gate.is_idle and UpgradeId.WARPGATERESEARCH not in bot.researched:
+                            gate.train(unit_type)
+                            self.build_order.increment_step()
+                        else:
+                            warp_in_pos = Utils.get_warp_in_pos(bot)
+                            await warp_in_unit(bot, unit_type, warp_in_pos)
+                            self.build_order.increment_step()
+
             #TODO: ADD WARPPRISM LOGIC, SO REINFORCEMENTS CAN BE WARPED IN CLOSE TO FIGHT
             unit_type: UnitTypeId = next_step.type_id
             production_structure_type = self.get_production_structure(unit_type)
             production_structures: Units = self.bot.structures(production_structure_type)
-            if Utils.can_build_unit(self.bot, next_step.type_id) and production_structures.idle:
+            if Utils.can_build_unit(self.bot, next_step.type_id) and production_structures:
                 if not production_structure_type in [UnitTypeId.WARPGATE, UnitTypeId.GATEWAY]:
-                    production_structures[0].train(unit_type)
-                    self.build_order.increment_step()
-                    return
+                    for struct in production_structures:
+                        if struct.is_idle and not self.bot.already_pending(unit_type):
+                            production_structures[0].train(unit_type)
+                            self.build_order.increment_step()
+                            return
                 await build_gateway_units(self.bot,unit_type)
-                #TODO: #77 STOP COUNTER FROM GOING CRAZY
-                self.build_order.increment_step()
 
 
         next_step: BuildInstruction = self.build_order.next_instruction()
@@ -154,7 +157,7 @@ async def train_unit(next_step:BuildInstruction):
         if not next_step and not self.build_order.buffer:
             return
 
-        if not next_step.type_id == UnitTypeId.ASSIMILATOR:
+        if not next_step.type_id == UnitTypeId.ASSIMILATOR and self.build_order.is_performing_initial_build:
             structure_cost:Cost = self.bot.calculate_cost(next_step.type_id)
             if (self.bot.minerals > (structure_cost.minerals*0.65)):
                 if not Utils.unittype_in_proximity_to_point(self.bot, UnitTypeId.PROBE, next_step.position):

bot/HarstemsAunt/main.py

@@ -308,13 +308,13 @@ async def on_unit_created(self, unit:Unit) -> None:
         Args:
             unit (Unit): Unit that gets created 
         """
-        if self.macro.build_order.is_performing_initial_build \
-            and not unit.type_id == UnitTypeId.PROBE:
+        if not unit.type_id == UnitTypeId.PROBE:
             self.army_groups[0].units_in_transit.append(unit.tag)
 
         for group in self.army_groups:
             if unit.type_id in group.requested_units:
-                group.units_in_transit.append(unit.tag)
+                self.macro.build_order.buffer.remove(unit.type_id)
+                logger.info(f"{unit.type_id} got removed from {group}")
 
     async def on_unit_type_changed(self, unit:Unit, previous_type:UnitTypeId) -> None:
         """ Coroutine that gets called when a unit changes type:

bot/HarstemsAunt/pathing.py

@@ -24,23 +24,23 @@ def __init__(self, bot:BotAI, debug:bool, fade_rate:float = 3.0) -> None:
         self.climber_grid: np.ndarray = self.map_data.get_climber_grid()
         self.units_grid: np.ndarray = self.map_data.get_pyastar_grid()
         self.ground_grid: np.ndarray = self.map_data.get_pyastar_grid()
-        #self.detection_grid: np.ndarray = self.map_data.get_pyastar_grid()
+        self.detection_grid: np.ndarray = self.map_data.get_pyastar_grid()
         self.air_grid: np.ndarray = self.map_data.get_clean_air_grid()
         self.influence_fade_rate: float = fade_rate
 
     def update(self, iteration) -> None:
 
         last_ground_grid:np.ndarray = self.ground_grid
         last_air_grid:np.ndarray = self.air_grid
-        #last_detection_grid: np.ndarray = self.detection_grid
+        last_detection_grid: np.ndarray = self.detection_grid
 
         last_ground_grid[last_ground_grid != 0] /= self.influence_fade_rate
         last_air_grid[last_air_grid != 0] /= self.influence_fade_rate
-        #last_detection_grid[last_detection_grid != 0] /=self.influence_fade_rate
+        last_detection_grid[last_detection_grid != 0] /=self.influence_fade_rate
 
         self.ground_grid = self.map_data.get_pyastar_grid() + last_ground_grid
         self.air_grid = self.map_data.get_clean_air_grid() + last_air_grid
-        #self.detection_grid = self.map_data.get_pyastar_grid() + last_detection_grid
+        self.detection_grid = self.map_data.get_pyastar_grid() + last_detection_grid
 
         for unit in self.bot.all_enemy_units:
             if unit.type_id in ALL_STRUCTURES:

bot/production_buffer.py

@@ -0,0 +1,68 @@
+"""DOCSTRING to shut up the Linter """
+from __future__ import annotations
+
+from typing import List
+
+from sc2.units import Units
+from sc2.bot_ai import BotAI
+from sc2.ids.unit_typeid import UnitTypeId
+
+
+class ProductionRequest:
+    """ Class Representing the a Production Request """
+
+    def __new__(cls, requested_unit:UnitTypeId, army_group_tag:int, build_structure_tag:int) -> ProductionRequest:
+        """ Creates new instance of Production Request
+
+        Args:
+            requested_unit (UnitTypeId): type of requested Unit
+            army_group_tag (int): identifier of the ArmyGroup requesting the Unit
+
+        Returns:
+            ProductionRequest: Request for a Unit (gets called in ArmyGroup)
+        """
+        instance = super().__new__(cls)
+        instance.requested_unit = requested_unit
+        instance.army_group_tag = army_group_tag
+        instance.build_structure_tag = build_structure_tag
+
+        return instance
+
+    def __init__(self, requested_unit:UnitTypeId, army_group_tag:int, build_structure_tag:int) -> None:
+        self.requested_unit:UnitTypeId = requested_unit
+        self.army_group_tag:int = army_group_tag
+        self.build_structure_tag = build_structure_tag
+
+    def __repr__(self) -> str:
+        return f"Unit: {self.requested_unit} requested by {self.army_group_tag}"
+
+    @property
+    def handled(self) -> bool:
+        return False
+
+    @handled.setter
+    def handled(self, new_status) -> None:
+        self.handled = new_status
+
+class ProductionBuffer:
+    """ Buffer for Production Requests before they are full filled"""
+
+    def __init__(self,bot:BotAI) -> None:
+        self.bot:BotAI = bot
+        self.requests:List[ProductionRequest] = []
+
+    @property
+    def gateways(self) -> Units:
+       return self.bot.units.filter(lambda struct: struct.type_id \
+           in [UnitTypeId.WARPGATE, UnitTypeId.GATEWAY])
+
+    @property
+    def stargates(self) -> Units:
+        return self.bot.units(UnitTypeId.STARGATE)
+
+    @property
+    def robofacilities(self) -> Units:
+        return self.bot.units(UnitTypeId.ROBOTICSFACILITY)
+
+    #def add_request(self) -> None:
+     #   self.