Skip to content

Typical Usage

from ares import AresBot
from ares.behaviors.macro.mining import Mining

class MyBot(AresBot):
    async def on_step(self, iteration: int) -> None:
        await super(MyBot, self).on_step(iteration)
        self.register_behavior(Mining())

MacroPlan dataclass

Bases: Behavior

Execute macro behaviors sequentially.

Idea here is to put macro behaviors in priority order.

Example: 

from ares.behaviors.macro import MacroPlan
from ares.behaviors.macro import (
    AutoSupply,
    Mining
    SpawnController
)

# initiate a new MacroPlan
macro_plan: MacroPlan = MacroPlan()

# then add behaviors in the order they should be executed
macro_plan.add(AutoSupply())
macro.plan.add(SpawnController(army_composition_dict=self.army_comp))


# register the macro plan
self.ai.register_behavior(macro_plan)

Attributes:

Name Type Description
macros list[Behavior]

A list of behaviors that should be executed. Defaults to an empty list.
Source code in src/ares/behaviors/macro/macro_plan.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass
class MacroPlan(Behavior):
    """Execute macro behaviors sequentially.

    Idea here is to put macro behaviors in priority order.

    Example:
    ```py
    from ares.behaviors.macro import MacroPlan
    from ares.behaviors.macro import (
        AutoSupply,
        Mining
        SpawnController
    )

    # initiate a new MacroPlan
    macro_plan: MacroPlan = MacroPlan()

    # then add behaviors in the order they should be executed
    macro_plan.add(AutoSupply())
    macro.plan.add(SpawnController(army_composition_dict=self.army_comp))


    # register the macro plan
    self.ai.register_behavior(macro_plan)
    ```

    Attributes:
        macros: A list of behaviors that should be executed. Defaults to an empty list.

    """

    macros: list[Behavior] = field(default_factory=list)

    def add(self, behavior: MacroBehavior) -> None:
        self.macros.append(behavior)

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        for macro in self.macros:
            if macro.execute(ai, config, mediator):
                # executed a macro behavior
                return True
        # none of the macro behaviors completed, no actions executed
        return False

AddonSwap dataclass

Bases: MacroBehavior

For Terran only, swap 3x3 structures. Pass in two structures and they will swap positions.

TODO: Extend this to support an exact swap, ie. swap techlab and reactor

Example: 

from ares.behaviors.macro import AddonSwap

# factory will find a reactor to fly to, any existing
# structure will fly to the factory's starting position
self.register_behavior(
    AddonSwap(factory, UnitID.REACTOR)
)

Attributes:

Name Type Description
structure_needing_addon Unit

The structure type we want the addon for.
addon_required UnitTypeId

Type of addon required.
Source code in src/ares/behaviors/macro/addon_swap.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@dataclass
class AddonSwap(MacroBehavior):
    """For Terran only, swap 3x3 structures.
    Pass in two structures and they will swap positions.

    TODO: Extend this to support an exact swap, ie. swap techlab and reactor


    Example:
    ```py
    from ares.behaviors.macro import AddonSwap

    # factory will find a reactor to fly to, any existing
    # structure will fly to the factory's starting position
    self.register_behavior(
        AddonSwap(factory, UnitID.REACTOR)
    )
    ```

    Attributes:
        structure_needing_addon: The structure type we want the addon for.
        addon_required: Type of addon required.
    """

    structure_needing_addon: Unit
    addon_required: UnitID

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        assert ai.race == Race.Terran, "Can only swap addons with Terran."
        assert (
            self.addon_required in ADDON_TYPES
        ), f"`self.addon_required` should be one of {ADDON_TYPES}"
        assert (
            self.structure_needing_addon.build_progress == 1
        ), "Structure requiring addon is not completed, and therefore can't fly"

        search_for_tags: set[int] = (
            ai.reactor_tags
            if self.addon_required == UnitID.REACTOR
            else ai.techlab_tags
        )

        # search for addon required
        add_ons: list[Unit] = [
            s
            for s in ai.structures
            if s.tag in search_for_tags and s.is_ready and s.is_idle
        ]
        if len(add_ons) == 0:
            return False

        closest_addon: Unit = cy_sorted_by_distance_to(
            add_ons, self.structure_needing_addon.position
        )[0]

        # is structure attached to this addon? then move it to `structure_needing_addon`
        if attached_structures := [
            s for s in ai.structures if s.add_on_tag == closest_addon.tag
        ]:
            mediator.move_structure(
                structure=attached_structures[0],
                target=self.structure_needing_addon.position,
            )

        mediator.move_structure(
            structure=self.structure_needing_addon,
            target=closest_addon.add_on_land_position,
        )

        return True

AutoSupply dataclass

Bases: MacroBehavior

Automatically build supply, works for all races.

Example: 

from ares.behaviors.macro import AutoSupply

self.register_behavior(AutoSupply(self.start_location))

Attributes:

Name Type Description
base_location Point2

The base location where supply should be built.
return_true_if_supply_required bool

If supply can't be afforded but is required, return true. Useful for creating a MacroPlan.
Source code in src/ares/behaviors/macro/auto_supply.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@dataclass
class AutoSupply(MacroBehavior):
    """Automatically build supply, works for all races.

    Example:
    ```py
    from ares.behaviors.macro import AutoSupply

    self.register_behavior(AutoSupply(self.start_location))
    ```

    Attributes:
        base_location: The base location where supply should be built.
        return_true_if_supply_required: If supply can't be afforded but is
            required, return true. Useful for creating a `MacroPlan`.

    """

    base_location: Point2
    return_true_if_supply_required: bool = True

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        if self._num_supply_required(ai, mediator) > 0:
            supply_type: UnitID = RACE_SUPPLY[ai.race]
            if ai.race == Race.Zerg:
                if ai.num_larva_left > 0 and ai.can_afford(supply_type):
                    ai.train(supply_type)
                    ai.num_larva_left -= 1
                    return True
            else:
                BuildStructure(self.base_location, supply_type).execute(
                    ai, config, mediator
                )
            return self.return_true_if_supply_required

        return False

    @staticmethod
    def _num_supply_required(ai: "AresBot", mediator: ManagerMediator) -> int:
        """
        TODO: Improve on this initial version
            Should calculate based on townhalls for Zerg only?
            Other races should be calculated based on production available
        """
        if ai.supply_cap >= 200:
            return 0

        num_ths: int = len(ai.ready_townhalls)
        supply_left: float = ai.supply_left
        supply_used: float = ai.supply_used
        pending_supply_units: int
        if ai.race == Race.Zerg:
            pending_supply_units = cy_unit_pending(ai, UnitID.OVERLORD)
        else:
            pending_supply_units = (
                int(ai.already_pending(RACE_SUPPLY[ai.race]))
                + mediator.get_building_counter[RACE_SUPPLY[ai.race]]
            )

        # zerg supply focus around townhalls
        if ai.race == Race.Zerg:
            # supply blocked
            if supply_left <= 0 and pending_supply_units < (num_ths + 1):
                max_supply: int = num_ths if supply_used < 72 else num_ths + 1
                return max_supply - pending_supply_units

            # low supply, restrict supply production
            if 60 > supply_used >= 13:
                supply_left_threshold: int = 5 if supply_used <= 36 else 6
                num_building = 1 if supply_left > 0 or supply_used < 29 else 2
                if (
                    supply_left <= supply_left_threshold
                    and pending_supply_units < num_building
                ):
                    return num_building - pending_supply_units
            else:
                if ai.race == Race.Zerg:
                    # scale up based on townhalls
                    if supply_left < 4 * num_ths and pending_supply_units < num_ths:
                        num: int = num_ths - pending_supply_units
                        return min(num, 6)
        # other races supply based on production facilities
        else:
            # scale up based on production structures
            num_production_structures: int = len(
                ai.structures.filter(
                    lambda s: s.type_id in ALL_PRODUCTION_STRUCTURES
                    and s.build_progress == 1.0
                )
            )
            if supply_left <= max(
                2 * num_production_structures, 5
            ) and pending_supply_units < math.ceil(num_production_structures / 2):
                num: int = (
                    math.ceil(num_production_structures / 2) - pending_supply_units
                )
                return min(num, 6)

            # we have no prod structures, just in case
            elif (
                num_production_structures == 0
                and supply_left <= 2
                and not pending_supply_units
            ):
                return 1 - pending_supply_units

        return 0

BuildStructure dataclass

Bases: MacroBehavior

Handy behavior for Terran and Protoss. Especially combined with Mining and ares built in placement solver. Finds an ideal mining worker, and an available precalculated placement. Then removes worker from mining records and provides a new role.

Example: 

from ares.behaviors.macro import BuildStructure

self.register_behavior(
    BuildStructure(self.start_location, UnitTypeId.BARRACKS)
)

Attributes:

Name Type Description
base_location Point2

The base location to build near.
structure_id UnitTypeId

The structure type we want to build.
max_on_route int

The max number of workers on route to build this. Defaults to 1.
first_pylon bool

Will look for the first pylon in placements dict. Defaults to False.
static_defence bool

Will look for static defense in placements dict. Defaults to False.
wall bool

Find wall placement if possible. Only the main base is currently supported. Defaults to False.
closest_to Optional[Point2]

Find placement at this base closest to the given point. Optional.
to_count int

Prevent going over this amount in total. Defaults to 0, turning this check off.
to_count_per_base int

Prevent going over this amount at this base location. Defaults to 0, turning this check off.
tech_progress_check float

Check if tech is ready before trying to build. Defaults to 0.85; setting it to 0.0 turns this check off.
Source code in src/ares/behaviors/macro/build_structure.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@dataclass
class BuildStructure(MacroBehavior):
    """Handy behavior for Terran and Protoss.
    Especially combined with `Mining` and ares built in placement solver.
    Finds an ideal mining worker, and an available precalculated placement.
    Then removes worker from mining records and provides a new role.


    Example:
    ```py
    from ares.behaviors.macro import BuildStructure

    self.register_behavior(
        BuildStructure(self.start_location, UnitTypeId.BARRACKS)
    )
    ```

    Attributes:
        base_location: The base location to build near.
        structure_id: The structure type we want to build.
        max_on_route: The max number of workers on route to build this. Defaults to 1.
        first_pylon: Will look for the first pylon in placements dict.
            Defaults to False.
        static_defence: Will look for static defense in placements dict.
            Defaults to False.
        wall: Find wall placement if possible. Only the main base is currently
            supported. Defaults to False.
        closest_to: Find placement at this base closest to the given point. Optional.
        to_count: Prevent going over this amount in total.
            Defaults to 0, turning this check off.
        to_count_per_base: Prevent going over this amount at this base location.
            Defaults to 0, turning this check off.
        tech_progress_check: Check if tech is ready before trying to build.
            Defaults to 0.85; setting it to 0.0 turns this check off.

    """

    base_location: Point2
    structure_id: UnitID
    max_on_route: int = 1
    first_pylon: bool = False
    static_defence: bool = False
    wall: bool = False
    closest_to: Optional[Point2] = None
    to_count: int = 0
    to_count_per_base: int = 0
    tech_progress_check: float = 0.85

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        assert (
            ai.race != Race.Zerg
        ), "BuildStructure Behavior not currently supported for Zerg."

        # already enough workers on route to build this
        if (
            ai.not_started_but_in_building_tracker(self.structure_id)
            >= self.max_on_route
        ):
            return False
        # if `to_count` is set, see if there is already enough
        if self.to_count and self._enough_existing(ai, mediator):
            return False
        # if we have enough at this base
        if self.to_count_per_base and self._enough_existing_at_this_base(mediator):
            return False

        # tech progress
        if (
            self.tech_progress_check
            and ai.tech_requirement_progress(self.structure_id)
            < self.tech_progress_check
        ):
            return False

        within_psionic_matrix: bool = (
            ai.race == Race.Protoss and self.structure_id != UnitID.PYLON
        )

        if placement := mediator.request_building_placement(
            base_location=self.base_location,
            structure_type=self.structure_id,
            first_pylon=self.first_pylon,
            static_defence=self.static_defence,
            wall=self.wall,
            within_psionic_matrix=within_psionic_matrix,
            closest_to=self.closest_to,
        ):
            if worker := mediator.select_worker(
                target_position=placement,
                force_close=True,
            ):
                mediator.build_with_specific_worker(
                    worker=worker,
                    structure_type=self.structure_id,
                    pos=placement,
                )
                return True
        return False

    def _enough_existing(self, ai: "AresBot", mediator: ManagerMediator) -> bool:
        existing_structures = mediator.get_own_structures_dict[self.structure_id]
        num_existing: int = len(
            [s for s in existing_structures if s.is_ready]
        ) + ai.structure_pending(self.structure_id)
        return num_existing >= self.to_count

    def _enough_existing_at_this_base(self, mediator: ManagerMediator) -> bool:
        placement_dict: dict = mediator.get_placements_dict
        size: BuildingSize = STRUCTURE_TO_BUILDING_SIZE[self.structure_id]
        potential_placements: dict[Point2:dict] = placement_dict[self.base_location][
            size
        ]
        taken: list[Point2] = [
            placement
            for placement in potential_placements
            if not potential_placements[placement]["available"]
        ]
        num_structures: int = 0
        for t in taken:
            if [
                s
                for s in mediator.get_own_structures_dict[self.structure_id]
                if cy_distance_to_squared(s.position, t) < 9.0
            ]:
                num_structures += 1
        return num_structures >= self.to_count_per_base

BuildWorkers dataclass

Bases: MacroBehavior

Finds idle townhalls / larvae and build workers.

Example: 

from ares.behaviors.macro import BuildWorkers

self.register_behavior(
    BuildWorkers(to_count=80)
)

Attributes:

Name Type Description
to_count int

The target count of workers we want to hit.
Source code in src/ares/behaviors/macro/build_workers.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass
class BuildWorkers(MacroBehavior):
    """Finds idle townhalls / larvae and build workers.

    Example:
    ```py
    from ares.behaviors.macro import BuildWorkers

    self.register_behavior(
        BuildWorkers(to_count=80)
    )
    ```

    Attributes:
        to_count: The target count of workers we want to hit.

    """

    to_count: int

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        worker_type: UnitID = ai.worker_type
        if (
            ai.can_afford(worker_type)
            and ai.townhalls.idle
            and ai.supply_workers < self.to_count
        ):
            return SpawnController(
                {
                    worker_type: {"proportion": 1.0, "priority": 0},
                }
            ).execute(ai, config, mediator)

        return False

ExpansionController dataclass

Bases: MacroBehavior

Manage expanding.

Example: 

from ares.behaviors.macro import ExpansionController

self.register_behavior(
    ExpansionController(to_count=8, max_pending=2)
)

Attributes:

Name Type Description
to_count int

The target base count.
can_afford_check bool

Check if we can afford expansion. Setting this to False will allow the worker to move to a location ready to build the expansion. Defaults to True.
check_location_is_safe bool

Check if we don't knowingly expand at a dangerous location. Defaults to True.
max_pending int

Maximum pending townhalls at any time. Defaults to 1.
Source code in src/ares/behaviors/macro/expansion_controller.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass
class ExpansionController(MacroBehavior):
    """Manage expanding.

    Example:
    ```py
    from ares.behaviors.macro import ExpansionController

    self.register_behavior(
        ExpansionController(to_count=8, max_pending=2)
    )
    ```

    Attributes:
        to_count: The target base count.
        can_afford_check: Check if we can afford expansion. Setting this to False
            will allow the worker to move to a location ready to build the expansion.
            Defaults to True.
        check_location_is_safe: Check if we don't knowingly expand at a dangerous
            location. Defaults to True.
        max_pending: Maximum pending townhalls at any time. Defaults to 1.
    """

    to_count: int
    can_afford_check: bool = True
    check_location_is_safe: bool = True
    max_pending: int = 1

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        # already have enough / or enough pending
        if (
            len([th for th in ai.townhalls if th.is_ready])
            + ai.structure_pending(ai.base_townhall_type)
            >= self.to_count
            or ai.structure_pending(ai.base_townhall_type) >= self.max_pending
            or (self.can_afford_check and not ai.can_afford(ai.base_townhall_type))
        ):
            return False

        if location := self._get_next_expansion_location(mediator):
            if worker := mediator.select_worker(target_position=location):
                mediator.build_with_specific_worker(
                    worker=worker, structure_type=ai.base_townhall_type, pos=location
                )
                return True

        return False

    def _get_next_expansion_location(
        self, mediator: ManagerMediator
    ) -> Optional[Point2]:
        grid: np.ndarray = mediator.get_ground_grid
        for el in mediator.get_own_expansions:
            location: Point2 = el[0]
            if (
                self.check_location_is_safe
                and not mediator.is_position_safe(grid=grid, position=location)
                or self._location_is_blocked(mediator, location)
            ):
                continue

            return location

    @staticmethod
    def _location_is_blocked(mediator: ManagerMediator, position: Point2) -> bool:
        """
        Check if enemy or own townhalls are blocking `position`.

        Parameters
        ----------
        mediator : ManagerMediator
        position : Point2

        Returns
        -------
        bool : True if location is blocked by something.

        """
        # TODO: Not currently an issue, but maybe we should consider rocks
        close_enemy: Units = mediator.get_units_in_range(
            start_points=[position],
            distances=5.5,
            query_tree=UnitTreeQueryType.EnemyGround,
        )[0]

        close_enemy: Units = close_enemy.filter(
            lambda u: u.type_id != UnitID.AUTOTURRET
        )
        if close_enemy:
            return True

        if mediator.get_units_in_range(
            start_points=[position],
            distances=5.5,
            query_tree=UnitTreeQueryType.AllOwn,
        )[0].filter(lambda u: u.type_id in TOWNHALL_TYPES):
            return True

        return False

GasBuildingController dataclass

Bases: MacroBehavior

Maintain gas building count. Finds an ideal mining worker, and an available geyser. Then removes worker from mining records and provides a new role.

Example: 

from ares.behaviors.macro import GasBuildingController

self.register_behavior(
    GasBuildingController(to_count=len(self.townhalls)*2)
)

Attributes:

Name Type Description
to_count int

How many gas buildings would we like?
max_pending int

How many gas buildings can be pending at any time? Defaults to 1.
closest_to Optional[Point2]

Find available geyser closest to this location. Optional, defaults to None
Source code in src/ares/behaviors/macro/gas_building_controller.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@dataclass
class GasBuildingController(MacroBehavior):
    """Maintain gas building count.
    Finds an ideal mining worker, and an available geyser.
    Then removes worker from mining records and provides a new role.


    Example:
    ```py
    from ares.behaviors.macro import GasBuildingController

    self.register_behavior(
        GasBuildingController(to_count=len(self.townhalls)*2)
    )
    ```

    Attributes:
        to_count: How many gas buildings would we like?
        max_pending: How many gas buildings can be pending at any time?
            Defaults to 1.
        closest_to: Find available geyser closest to this location.
            Optional, defaults to `None`

    """

    to_count: int
    max_pending: int = 1
    closest_to: Optional[Point2] = None

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        num_gas: int
        if ai.race == Race.Terran:
            num_gas = ai.not_started_but_in_building_tracker(ai.gas_type) + len(
                ai.gas_buildings
            )
        else:
            num_gas = len(ai.gas_buildings) + mediator.get_building_counter[ai.gas_type]
        # we have enough gas / building gas then don't attempt behavior
        if (
            num_gas >= self.to_count
            or mediator.get_building_counter[ai.gas_type] >= self.max_pending
            or ai.minerals < 35
        ):
            return False

        existing_gas_buildings: Units = ai.all_gas_buildings
        if available_geysers := [
            u
            for u in ai.vespene_geyser
            if not [
                g
                for g in existing_gas_buildings
                if cy_distance_to_squared(u.position, g.position) < 25.0
            ]
            and [
                th
                for th in ai.townhalls
                if cy_distance_to_squared(u.position, th.position) < 144.0
                and th.build_progress > 0.9
            ]
        ]:
            if not self.closest_to:
                self.closest_to = ai.start_location

            available_geysers = cy_sorted_by_distance_to(
                available_geysers, self.closest_to
            )
            if worker := mediator.select_worker(
                target_position=available_geysers[0], force_close=True
            ):
                mediator.build_with_specific_worker(
                    worker=worker,
                    structure_type=ai.gas_type,
                    pos=available_geysers[0],
                )
                return True

        return False

Mining dataclass

Bases: MacroBehavior

Handle worker mining control.

Note: Could technically be CombatBehavior, but is treated here as a MacroBehavior since many tasks are carried out.

Example: 

from ares.behaviors.macro import Mining

self.register_behavior(Mining())

Attributes:

Name Type Description
flee_at_health_perc float

If worker is in danger, at what health percentage should it flee? Defaults to 0.5.
keep_safe bool

Should workers flee if they are in danger? Defaults to True.
long_distance_mine bool

Can the worker long distance mine if it has nothing to do? Defaults to True.
mineral_boost bool

Turn mineral boosting on or off. Defaults to True.
vespene_boost bool

Turn vespene boosting on or off (only active when workers_per_gas < 3). WARNING: VESPENE BOOSTING CURRENTLY NOT WORKING. Defaults to True.
workers_per_gas int

Control how many workers are assigned to each gas. Defaults to 3.
self_defence_active bool

If set to True, workers will have some basic defence. Certain workers will attack enemy in range. Defaults to True.
safe_long_distance_mineral_fields Optional[Units]

Used internally, set when a worker starts long distance mining. Defaults to None.
weight_safety_limit float

Workers will flee if enemy influence is above this number. Defaults to 12.0
Source code in src/ares/behaviors/macro/mining.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
@dataclass
class Mining(MacroBehavior):
    """Handle worker mining control.

    Note: Could technically be `CombatBehavior`, but is treated here as a
    MacroBehavior since many tasks are carried out.

    Example:
    ```py
    from ares.behaviors.macro import Mining

    self.register_behavior(Mining())
    ```

    Attributes:
        flee_at_health_perc: If worker is in danger, at what
            health percentage should it flee? Defaults to 0.5.
        keep_safe: Should workers flee if they are in danger?
            Defaults to True.
        long_distance_mine: Can the worker long distance mine if it has nothing to do?
            Defaults to True.
        mineral_boost: Turn mineral boosting on or off. Defaults to True.
        vespene_boost: Turn vespene boosting on or off
            (only active when workers_per_gas < 3).
            WARNING: VESPENE BOOSTING CURRENTLY NOT WORKING.
            Defaults to True.
        workers_per_gas: Control how many workers are assigned to each gas.
            Defaults to 3.
        self_defence_active: If set to True, workers will have some basic defence.
            Certain workers will attack enemy in range. Defaults to True.
        safe_long_distance_mineral_fields: Used internally, set when a worker starts
            long distance mining. Defaults to None.
        weight_safety_limit: Workers will flee if enemy influence is above this number.
            Defaults to 12.0

    """

    flee_at_health_perc: float = 0.5
    keep_safe: bool = True
    long_distance_mine: bool = True
    mineral_boost: bool = True
    vespene_boost: bool = False
    workers_per_gas: int = 3
    self_defence_active: bool = True
    safe_long_distance_mineral_fields: Optional[Units] = None
    locked_action_tags: dict[int, float] = field(default_factory=dict)
    weight_safety_limit: float = 12.0

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        workers: Units = mediator.get_units_from_role(
            role=UnitRole.GATHERING,
            unit_type=ai.worker_type,
        )
        if not workers:
            return False

        resources_dict: dict[int, Unit] = ai.unit_tag_dict
        health_perc: float = self.flee_at_health_perc
        avoidance_grid: np.ndarray = mediator.get_ground_avoidance_grid
        grid: np.ndarray = mediator.get_ground_grid
        mineral_patch_to_list_of_workers: dict[
            int, set[int]
        ] = mediator.get_mineral_patch_to_list_of_workers
        path_find: Callable = mediator.find_path_next_point
        pos_safe: Callable = mediator.is_position_safe
        th_dist_factor: float = config[MINING][TOWNHALL_DISTANCE_FACTOR]
        worker_to_geyser_dict: dict[int, int] = mediator.get_worker_to_vespene_dict
        worker_to_mineral_patch_dict: dict[
            int, int
        ] = mediator.get_worker_to_mineral_patch_dict
        worker_to_th: dict[int, int] = mediator.get_worker_tag_to_townhall_tag
        # for each mineral tag, get the position in front of the mineral
        min_target: dict[int, Point2] = mediator.get_mineral_target_dict
        main_enemy_ground_threats: Optional[Units] = None
        race: Race = ai.race
        if self.self_defence_active:
            main_enemy_ground_threats = mediator.get_main_ground_threats_near_townhall

        for worker in workers:
            worker_position: Point2 = worker.position
            worker_tag: int = worker.tag
            resource: Optional[Unit] = None
            resource_position: Optional[Point2] = None
            resource_tag: int = -1

            assigned_mineral_patch: bool = worker_tag in worker_to_mineral_patch_dict
            assigned_gas_building: bool = worker_tag in worker_to_geyser_dict
            dist_to_resource: float = 15.0
            if assigned_mineral_patch or assigned_gas_building:
                resource_tag: int = (
                    worker_to_mineral_patch_dict[worker_tag]
                    if assigned_mineral_patch
                    else worker_to_geyser_dict[worker_tag]
                )
                # using try except is faster than dict.get()
                try:
                    _resource = resources_dict[resource_tag]
                    resource_position = _resource.position
                    resource_tag = _resource.tag
                    resource = _resource
                    dist_to_resource = cy_distance_to(
                        worker_position, resource_position
                    )
                    if (
                        resource.type_id in GAS_BUILDINGS
                        and resource.vespene_contents == 0
                    ):
                        mediator.remove_gas_building(gas_building_tag=resource_tag)

                except KeyError:
                    # Mined out or no vision? Remove it
                    if assigned_mineral_patch:
                        mediator.remove_mineral_field(mineral_field_tag=resource_tag)
                    else:
                        mediator.remove_gas_building(gas_building_tag=resource_tag)
                    continue

            perc_health: float = (
                worker.health_percentage
                if race != Race.Protoss
                else worker.shield_health_percentage
            )
            # keeping worker safe is first priority
            if self.keep_safe and (
                # lib zone / nukes etc
                not pos_safe(grid=avoidance_grid, position=worker_position)
                # retreat based on self.flee_at_health_perc value
                or (
                    perc_health <= health_perc
                    and not pos_safe(grid=grid, position=worker_position)
                )
                or not pos_safe(
                    grid=grid,
                    position=worker_position,
                    weight_safety_limit=self.weight_safety_limit,
                )
            ):
                self._keep_worker_safe(mediator, grid, worker)

            elif main_enemy_ground_threats and self._worker_attacking_enemy(
                ai, dist_to_resource, worker
            ):
                pass

            # do we have record of this worker? If so mine from the relevant resource
            elif ai.townhalls and (assigned_mineral_patch or assigned_gas_building):
                # we are far away, path to min field to avoid enemies
                if dist_to_resource > 6.0 and not worker.is_carrying_resource:
                    worker.move(
                        path_find(
                            start=worker_position,
                            target=resource_position,
                            grid=grid,
                        )
                    )

                # fix realtime bug where worker is stuck with a move command
                # but already returned minerals
                elif (
                    len(worker.orders) == 1
                    and worker.orders[0].ability.id == AbilityId.MOVE
                    and ai.ready_townhalls
                    and worker.order_target
                    == cy_closest_to(worker_position, ai.ready_townhalls).tag
                    # shift worker to correct resource if it ends up on wrong one
                ) or (worker.is_gathering and worker.order_target != resource_tag):
                    # target being the mineral
                    worker(AbilityId.SMART, resource)

                elif (self.mineral_boost and assigned_mineral_patch) or (
                    self.vespene_boost and self.workers_per_gas < 3
                ):
                    self._do_mining_boost(
                        ai,
                        th_dist_factor,
                        min_target,
                        resource,
                        worker,
                        worker_to_th,
                        worker_position,
                        resource_position,
                    )
                else:
                    self._do_standard_mining(ai, worker, resource)

            # nowhere for this worker to go, long distance mining
            elif self.long_distance_mine and ai.minerals and ai.townhalls:
                self._long_distance_mining(
                    ai,
                    mediator,
                    grid,
                    worker,
                    mineral_patch_to_list_of_workers,
                    worker_position,
                )

            # this worker really has nothing to do, keep it safe at least
            # don't mine from anywhere since user requested no `long_distance_mine`
            elif not pos_safe(grid=grid, position=worker_position):
                self._keep_worker_safe(mediator, grid, worker)

        mediator.set_workers_per_gas(amount=self.workers_per_gas)
        return True

    @staticmethod
    def _keep_worker_safe(
        mediator: ManagerMediator, grid: np.ndarray, worker: Unit
    ) -> None:
        """Logic for keeping workers in danger safe.

        Parameters:
            mediator: ManagerMediator used for getting information from
                other managers.
            grid: Ground grid with enemy influence.
            worker: Worker to keep safe.
        """
        worker.move(
            mediator.find_closest_safe_spot(from_pos=worker.position, grid=grid)
        )

    def _do_standard_mining(self, ai: "AresBot", worker: Unit, resource: Unit) -> None:
        worker_tag: int = worker.tag
        # prevent spam clicking workers on patch to reduce APM
        if worker_tag in self.locked_action_tags:
            if ai.time > self.locked_action_tags[worker_tag] + 0.5:
                self.locked_action_tags.pop(worker_tag)
            return
        # moved worker from gas
        if worker.is_carrying_vespene and resource.is_mineral_field:
            worker.return_resource()
            self.locked_action_tags[worker_tag] = ai.time
        else:
            # work out when we need to issue command to mine resource
            if worker.is_idle or (
                cy_distance_to_squared(worker.position, resource.position) > 81.0
                and worker.order_target
                and worker.order_target != resource
            ):
                worker.gather(resource)
                self.locked_action_tags[worker_tag] = ai.time
                return

            # force worker to stay on correct resource
            # in game auto mining will sometimes shift worker
            if (
                not worker.is_carrying_minerals
                and not worker.is_carrying_vespene
                and worker.order_target != resource.tag
            ):
                worker.gather(resource)
                # to reduce apm we prevent spam clicking on same mineral
                self.locked_action_tags[worker_tag] = ai.time

    def _long_distance_mining(
        self,
        ai: "AresBot",
        mediator: ManagerMediator,
        grid: np.ndarray,
        worker: Unit,
        mineral_patch_to_list_of_workers: dict[int, set[int]],
        worker_position: Point2,
    ) -> None:
        """Logic for long distance mining.

        Parameters
        ----------
        ai : AresBot
            Bot object that will be running the game.
        mediator : ManagerMediator
            Used for getting information from other managers.
        grid : np.ndarray
            Ground grid with enemy influence.
        worker: Unit
            Worker we want to issue actions to.
        mineral_patch_to_list_of_workers: dict
            Record of assigned mineral patches, so we know which ones to avoid.
        worker_position :
            Pass this in for optimization purposes.
        Returns
        -------

        """
        if (
            worker.is_gathering
            and worker.order_target not in mediator.get_mineral_patch_to_list_of_workers
        ):
            return

        completed_bases: Units = ai.townhalls.ready
        # there is nowhere to return resources!
        if not completed_bases:
            return

        if not self.safe_long_distance_mineral_fields:
            self.safe_long_distance_mineral_fields = (
                self._safe_long_distance_mineral_fields(ai, mediator)
            )

        target_mineral: Optional[Unit] = None

        # on route to a far mineral patch
        if (
            (not worker.is_gathering and not worker.is_carrying_resource)
            # mining somewhere we manage ourselves
            or (
                worker.order_target
                and worker.order_target in mineral_patch_to_list_of_workers
            )
        ):
            # if there is a pending base, we should mine from there
            pending_bases: Units = ai.townhalls.filter(lambda th: not th.is_ready)
            if pending_bases and ai.mineral_field:
                target_base: Unit = cy_closest_to(worker_position, pending_bases)
                target_mineral = cy_closest_to(target_base.position, ai.mineral_field)
            # no pending base, find a mineral field
            elif self.safe_long_distance_mineral_fields:
                # early game, get closest to natural
                if ai.time < 260.0:
                    target_mineral = cy_closest_to(
                        mediator.get_own_nat, self.safe_long_distance_mineral_fields
                    )
                else:
                    target_mineral = cy_closest_to(
                        worker.position, self.safe_long_distance_mineral_fields
                    )

            if target_mineral:
                target_mineral_position: Point2 = target_mineral.position
                if (
                    not mediator.is_position_safe(
                        grid=grid,
                        position=worker_position,
                    )
                    and cy_distance_to_squared(worker_position, target_mineral_position)
                    > 25.0
                ):
                    move_to: Point2 = mediator.find_path_next_point(
                        start=worker_position,
                        target=target_mineral_position,
                        grid=grid,
                        sense_danger=False,
                    )
                    worker.move(move_to)
                elif ai.mineral_field:
                    if worker.order_target and worker.order_target == target_mineral:
                        return
                    worker.gather(target_mineral)
        # worker is travelling back to a ready townhall
        else:
            if worker.is_returning:
                return
            return_base: Unit = cy_closest_to(worker_position, completed_bases)
            return_base_position: Point2 = return_base.position
            if (
                not mediator.is_position_safe(grid=grid, position=worker_position)
                and cy_distance_to_squared(worker_position, return_base_position) > 64.0
            ):
                move_to: Point2 = mediator.find_path_next_point(
                    start=worker_position,
                    target=return_base_position,
                    grid=grid,
                    sense_danger=False,
                )
                worker.move(move_to)
            else:
                worker.return_resource()

    @staticmethod
    def _do_mining_boost(
        ai,
        distance_to_townhall_factor,
        mineral_target_dict,
        target,
        worker,
        worker_tag_to_townhall_tag,
        worker_position: Point2,
        target_position: Point2,
    ) -> bool:
        """Perform the trick so that worker does not decelerate.

        This avoids worker deceleration when mining by issuing a Move command near a
        mineral patch/townhall and then issuing a Gather or Return command once the
        worker is close enough to immediately perform the action instead of issuing a
        Gather command and letting the SC2 engine manage the worker.

        Parameters:
            ai: Main AresBot object
            distance_to_townhall_factor: Multiplier used for finding the target
                of the Move command when returning resources.
            target: Mineral field or Townhall that the worker should be
                moving toward/performing an action on.
            worker: The worker being boosted.
            worker_tag_to_townhall_tag: The townhall this worker belongs to,
                or where resources will be returned.
            worker_position: Pass in for optimization purposes.
            target_position: Pass in for optimization purposes.

        Returns:
            Whether this method carries out an action
        """

        if target.is_mineral_field or target.is_vespene_geyser:
            try:
                resource_target_pos: Point2 = mineral_target_dict[target_position]
            except KeyError:
                logger.warning(
                    f"{target_position} not found in resource_target_pos, "
                    f"no action will be provided for f{worker.tag}"
                )
                return False
        else:
            resource_target_pos: Point2 = Point2(
                cy_towards(target_position, worker_position, TOWNHALL_TARGET)
            )

        if not target:
            ai.mediator.remove_mineral_field(mineral_field_tag=target.tag)
            ai.mediator.remove_worker_from_mineral(worker_tag=worker.tag)
        elif not target.is_mineral_field and not target.vespene_contents:
            ai.mediator.remove_gas_building(gas_building_tag=target.tag)

        try:
            townhall: Unit = ai.unit_tag_dict[worker_tag_to_townhall_tag[worker.tag]]
        except KeyError:
            townhall: Unit = cy_closest_to(worker_position, ai.townhalls)

        return SpeedMining(
            worker,
            target,
            worker_position,
            resource_target_pos,
            distance_to_townhall_factor,
            townhall,
        ).execute(ai, ai.mediator, ai.config)

    @staticmethod
    def _safe_long_distance_mineral_fields(
        ai: "AresBot", mediator: ManagerMediator
    ) -> list[Unit]:
        """Find mineral fields for long distance miners.

        Parameters:
            ai: Main AresBot object
            mediator: Manager mediator to interact with the managers

        Returns:
            Units object of safe mineral patches if mineral patches still exist
        """
        if not ai.mineral_field:
            return

        assigned_patches: dict[int, set] = mediator.get_mineral_patch_to_list_of_workers
        grid: np.ndarray = mediator.get_ground_grid
        mfs: list[Unit] = cy_sorted_by_distance_to(ai.mineral_field, ai.start_location)
        weight_safety_limit: float = 6.0
        if ai.state.score.collection_rate_minerals < 300:
            weight_safety_limit = 100.0
        safe_mfs: list[Unit] = []
        for mf in mfs:
            if mf in assigned_patches or not mediator.is_position_safe(
                grid=grid, position=mf.position, weight_safety_limit=weight_safety_limit
            ):
                continue
            safe_mfs.append(mf)

        return safe_mfs

    @staticmethod
    def _worker_attacking_enemy(
        ai: "AresBot", dist_to_resource: float, worker: Unit
    ) -> bool:
        if not worker.is_collecting or dist_to_resource > 1.0:
            if enemies := cy_in_attack_range(worker, ai.enemy_units):
                target: Unit = cy_pick_enemy_target(enemies)

                if cy_attack_ready(ai, worker, target):
                    worker.attack(target)
                    return True
        return False

ProductionController dataclass

Bases: MacroBehavior

Handle creating extra production facilities based on an army composition dictionary. This dictionary should be structured the same as the one passed into SpawnController

Terran / Protoss only

Example bot code: 

from ares.behaviors.production_controller import ProductionController

# Note: This does not try to build production facilities and
# will ignore units that are impossible to currently spawn.
army_composition: dict[UnitID: {float, bool}] = {
    UnitID.MARINE: {"proportion": 0.6, "priority": 2},  # lowest priority
    UnitID.MEDIVAC: {"proportion": 0.25, "priority": 1},
    UnitID.SIEGETANK: {"proportion": 0.15, "priority": 0},  # highest priority
}
# where `self` is an `AresBot` object
self.register_behavior(ProductionController(
    army_composition, self.ai.start_location))

Attributes:

Name Type Description
army_composition_dict dict[UnitTypeId, dict[str, float, str, int]]

A dictionary detailing how an army composition should be made up. The proportional values should all add up to 1.0, with a priority integer for unit emphasis.
base_location Point2

The location where production should be built.
add_production_at_bank tuple[int, int]

When the bank reaches this size, calculate what extra production would be useful. Tuple where the first value is minerals and the second is vespene. Defaults to (300, 300).
alpha float

Controls how much production to add when the bank is higher than add_production_at_bank. Defaults to 0.9.
unit_pending_progress float

Check for production structures almost ready. For example, a marine might almost be ready, meaning we don't need to add extra production just yet. Defaults to 0.8.
ignore_below_proportion float

If we don't want many of a unit, there's no point adding production. Checks if it's possible to build a unit first. Defaults to 0.05.
should_repower_structures bool

Search for unpowered structures and build a new pylon if needed. Defaults to True.
Source code in src/ares/behaviors/macro/production_controller.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@dataclass
class ProductionController(MacroBehavior):
    """Handle creating extra production facilities based
    on an army composition dictionary.
    This dictionary should be structured the same as the one
    passed into SpawnController

    Terran / Protoss only

    Example bot code:
    ```py
    from ares.behaviors.production_controller import ProductionController

    # Note: This does not try to build production facilities and
    # will ignore units that are impossible to currently spawn.
    army_composition: dict[UnitID: {float, bool}] = {
        UnitID.MARINE: {"proportion": 0.6, "priority": 2},  # lowest priority
        UnitID.MEDIVAC: {"proportion": 0.25, "priority": 1},
        UnitID.SIEGETANK: {"proportion": 0.15, "priority": 0},  # highest priority
    }
    # where `self` is an `AresBot` object
    self.register_behavior(ProductionController(
        army_composition, self.ai.start_location))

    ```

    Attributes:
        army_composition_dict: A dictionary detailing how an army composition
            should be made up. The proportional values should all add up to 1.0,
            with a priority integer for unit emphasis.
        base_location: The location where production should be built.
        add_production_at_bank: When the bank reaches this size, calculate what
            extra production would be useful. Tuple where the first value is
            minerals and the second is vespene. Defaults to `(300, 300)`.
        alpha: Controls how much production to add when the bank is higher than
            `add_production_at_bank`. Defaults to `0.9`.
        unit_pending_progress: Check for production structures almost ready.
            For example, a marine might almost be ready, meaning we don't need to
            add extra production just yet. Defaults to `0.8`.
        ignore_below_proportion: If we don't want many of a unit, there's no point
            adding production. Checks if it's possible to build a unit first.
            Defaults to `0.05`.
        should_repower_structures: Search for unpowered structures and build a
            new pylon if needed. Defaults to `True`.

    """

    army_composition_dict: dict[UnitID, dict[str, float, str, int]]
    base_location: Point2
    add_production_at_bank: tuple[int, int] = (300, 300)
    alpha: float = 0.9
    unit_pending_progress: float = 0.75
    ignore_below_proportion: float = 0.05
    should_repower_structures: bool = True

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        assert (
            ai.race != Race.Zerg
        ), "ProductionController behavior is for Protoss and Terran only"
        if ai.race == Race.Protoss and self.should_repower_structures:
            if RestorePower().execute(ai, config, mediator):
                return True

        army_comp_dict: dict = self.army_composition_dict
        assert isinstance(
            army_comp_dict, dict
        ), f"self.army_composition_dict should be dict type, got {type(army_comp_dict)}"

        # get the current standing army based on the army comp dict
        # note we don't consider units outside the army comp dict
        unit_types: list[UnitID] = [*army_comp_dict]

        num_total_units: int = sum(
            [
                mediator.get_own_unit_count(unit_type_id=unit_type)
                for unit_type in unit_types
            ]
        )
        proportion_sum: float = 0.0
        structure_dict: dict[UnitID, Units] = mediator.get_own_structures_dict
        flying_structures: dict[int, dict] = mediator.get_flying_structure_tracker
        # +1 to avoid division by zero
        collection_rate_minerals: int = ai.state.score.collection_rate_minerals + 1
        collection_rate_vespene: int = ai.state.score.collection_rate_vespene + 1

        # iterate through desired army comp starting with the highest priority unit
        for unit_type_id, army_comp_info in sorted(
            army_comp_dict.items(), key=lambda x: x[1].get("priority", int(0))
        ):
            assert isinstance(unit_type_id, UnitID), (
                f"army_composition_dict expects UnitTypeId type as keys, "
                f"got {type(unit_type_id)}"
            )

            num_this_unit: int = mediator.get_own_unit_count(unit_type_id=unit_type_id)
            current_proportion: float = num_this_unit / (num_total_units + 1e-16)
            target_proportion: float = army_comp_info["proportion"]
            proportion_sum += target_proportion
            train_from: set[UnitID] = UNIT_TRAINED_FROM[unit_type_id]
            trained_from: UnitID = next(iter(UNIT_TRAINED_FROM[unit_type_id]))
            if unit_type_id in GATEWAY_UNITS:
                trained_from = UnitID.GATEWAY

            existing_structures: list[Unit] = []
            for structure_type in train_from:
                existing_structures.extend(structure_dict[structure_type])

            # we need to tech up, no further action is required
            if TechUp(
                unit_type_id,
                base_location=self.base_location,
                ignore_existing_techlabs=current_proportion < target_proportion,
            ).execute(ai, config, mediator):
                return True

            if ai.tech_requirement_progress(trained_from) < 0.95:
                continue

            # we have a worker on route to build this production
            # leave alone for now
            if ai.not_started_but_in_building_tracker(trained_from):
                continue

            # we can afford prod, work out how much prod to support
            # based on income
            if (
                ai.minerals > self.add_production_at_bank[0]
                and ai.vespene > self.add_production_at_bank[1]
            ):
                if self._building_production_due_to_bank(
                    ai,
                    unit_type_id,
                    collection_rate_minerals,
                    collection_rate_vespene,
                    existing_structures,
                    trained_from,
                    target_proportion,
                ):
                    return True

            # target proportion is low and something is pending, don't add extra yet
            if target_proportion <= 0.15 and (
                any([ai.structure_pending(type_id) for type_id in train_from])
            ):
                continue

            # existing production is enough for our income?
            cost: Cost = ai.calculate_cost(unit_type_id)
            total_cost = cost.minerals + cost.vespene
            divide_by: float = total_cost * 4.5
            if len(existing_structures) >= int(
                (collection_rate_minerals + collection_rate_vespene) / divide_by
            ):
                continue

            # if Terran has a production building floating, wait
            if self.is_flying_production(ai, flying_structures, train_from):
                continue

            # already have enough of this unit type, don't need production
            if current_proportion * 1.05 >= target_proportion:
                continue

            # already could build this unit if we wanted to?
            if self._can_already_produce(train_from, structure_dict):
                continue

            # add max depending on income
            max_pending = int(
                (collection_rate_minerals + collection_rate_vespene) / 1000
            )

            if ai.structure_pending(trained_from) >= max_pending:
                continue

            built = BuildStructure(self.base_location, trained_from).execute(
                ai, ai.config, ai.mediator
            )
            if built:
                logger.info(
                    f"{ai.time_formatted} Adding {trained_from} so that we can build "
                    f"more {unit_type_id}. Current proportion: {current_proportion}"
                    f" Target proportion: {target_proportion}"
                )
                return built

        # we checked everything and no action is required
        return False

    def _building_production_due_to_bank(
        self,
        ai: "AresBot",
        unit_type_id: UnitID,
        collection_rate_minerals: int,
        collection_rate_vespene: int,
        existing_structures: list[Unit],
        trained_from: UnitID,
        target_proportion: float,
    ) -> bool:
        # work out how many units we could afford at once
        cost_of_unit: Cost = ai.calculate_cost(unit_type_id)
        simul_afford_min: int = int(
            (collection_rate_minerals / (cost_of_unit.minerals + 1))
            * self.alpha
            * target_proportion
        )
        simul_afford_ves: int = int(
            (collection_rate_vespene / (cost_of_unit.vespene + 1))
            * self.alpha
            * target_proportion
        )
        num_existing: int = len([s for s in existing_structures if s.is_ready])
        num_production: int = num_existing + ai.structure_pending(trained_from)

        if num_production < simul_afford_min and num_production < simul_afford_ves:
            if BuildStructure(self.base_location, trained_from).execute(
                ai, ai.config, ai.mediator
            ):
                logger.info(f"Adding {trained_from} as income level will support this.")
                return True
        return False

    def _can_already_produce(self, train_from, structure_dict) -> bool:
        for structure_type in train_from:
            if structure_type == UnitID.WARPGATE and [
                s for s in structure_dict[structure_type] if not s.is_ready
            ]:
                return True

            for s in structure_dict[structure_type]:
                if s.is_ready and s.is_idle:
                    return True
                if s.orders:
                    if s.orders[0].progress >= self.unit_pending_progress:
                        return True
                # structure about to come online
                if 1.0 > s.build_progress >= 0.9:
                    return True

        return False

    def is_flying_production(
        self, ai: "AresBot", flying_structures: dict, train_from: set[UnitID]
    ) -> bool:
        if ai.race == Race.Terran:
            prod_flying: bool = False
            # might have this structure flying
            for tag in flying_structures:
                if unit := ai.unit_tag_dict.get(tag, None):
                    # make sure flying structure is nearby
                    if (
                        unit.type_id in UNIT_UNIT_ALIAS
                        and cy_distance_to_squared(unit.position, self.base_location)
                        < 360.0
                    ):
                        for s_id in train_from:
                            if UNIT_UNIT_ALIAS[unit.type_id] == s_id:
                                prod_flying = True
                                break
            if prod_flying:
                return True
        return False

    def _add_techlab_to_existing(
        self, ai: "AresBot", unit_type_id: UnitID, researched_from_id
    ) -> bool:
        structures_dict: dict = ai.mediator.get_own_structures_dict
        build_techlab_from: UnitID = BUILD_TECHLAB_FROM[researched_from_id]
        _build_techlab_from_structures: list[Unit] = structures_dict[
            build_techlab_from
        ].copy()
        if without_techlabs := [
            s
            for s in _build_techlab_from_structures
            if s.is_ready and s.is_idle and not s.has_add_on
        ]:
            without_techlabs[0].build(researched_from_id)
            logger.info(
                f"{ai.time_formatted} Adding {researched_from_id} so that we can "
                f"build more {unit_type_id}"
            )
            return True
        return False

RestorePower dataclass

Bases: MacroBehavior

Restore power for protoss structures.

Note: ProductionController is set to call this automatically configured via should_repower_structures parameter. Though this behavior may also be used separately.

Example: 

from ares.behaviors.restore_power import RestorePower

self.register_behavior(RestorePower())

Source code in src/ares/behaviors/macro/restore_power.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@dataclass
class RestorePower(MacroBehavior):
    """Restore power for protoss structures.

    Note: `ProductionController` is set to call this automatically
    configured via `should_repower_structures` parameter.
    Though this behavior may also be used separately.

    Example:
    ```py
    from ares.behaviors.restore_power import RestorePower

    self.register_behavior(RestorePower())
    ```
    """

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        if structures_no_power := [
            s
            for s in ai.structures
            if s.type_id in REQUIRE_POWER_STRUCTURE_TYPES
            and not cy_pylon_matrix_covers(
                s.position,
                mediator.get_own_structures_dict[UnitID.PYLON],
                ai.game_info.terrain_height.data_numpy,
                pylon_build_progress=1e-16,
            )
        ]:
            for structure in structures_no_power:
                if self._already_restoring(structure, mediator):
                    continue

                if self._restoring_power(structure, ai, config, mediator):
                    return True

        return False

    @staticmethod
    def _already_restoring(structure: Unit, mediator: ManagerMediator) -> bool:
        """
        Check if unpowered `structure` is currently being restored.
        Potentially probe already on the way?

        Parameters
        ----------
        structure
        mediator

        Returns
        -------

        """
        building_tracker: dict = mediator.get_building_tracker_dict
        for tag, building_info in building_tracker.items():
            type_id: UnitID = building_info[ID]
            if type_id == UnitID.PYLON:
                pos: Point2 = building_info[TARGET]
                if (
                    cy_distance_to_squared(structure.position, pos)
                    < PYLON_POWERED_DISTANCE_SQUARED
                ):
                    return True

        return False

    @staticmethod
    def _restoring_power(
        structure: Unit, ai: "AresBot", config: dict, mediator: ManagerMediator
    ) -> bool:
        """Given an unpowered structure, find a pylon position.

        Parameters
        ----------
        structure
        ai
        mediator

        Returns
        -------

        """
        placements_dict: dict = mediator.get_placements_dict
        position: Point2 = structure.position
        size: BuildingSize = BuildingSize.TWO_BY_TWO
        offset: float = 1.0

        for base_loc, placements_info in placements_dict.items():
            two_by_twos = placements_info[size]
            if available := [
                placement
                for placement in two_by_twos
                if two_by_twos[placement]["available"]
                and cy_distance_to_squared(placement, position)
                < PYLON_POWERED_DISTANCE_SQUARED
                and not two_by_twos[placement]["worker_on_route"]
                and cy_can_place_structure(
                    (placement[0] - offset, placement[1] - offset),
                    (2, 2),
                    ai.state.creep.data_numpy,
                    ai.game_info.placement_grid.data_numpy,
                    mediator.get_ground_grid.astype(np.uint8).T,
                    avoid_creep=True,
                    include_addon=False,
                )
            ]:
                return BuildStructure(
                    base_loc, UnitID.PYLON, closest_to=available[0], wall=True
                ).execute(ai, config, mediator)

        return False

SpawnController dataclass

Bases: MacroBehavior

Handle spawning army compositions.

Example bot code: 

from ares.behaviors.spawn_controller import SpawnController

# Note: This does not try to build production facilities and
# will ignore units that are impossible to currently spawn.
army_composition: dict[UnitID: {float, bool}] = {
    UnitID.MARINE: {"proportion": 0.6, "priority": 2},  # lowest priority
    UnitID.MEDIVAC: {"proportion": 0.25, "priority": 1},
    UnitID.SIEGETANK: {"proportion": 0.15, "priority": 0},  # highest priority
}
# where `self` is an `AresBot` object
self.register_behavior(SpawnController(army_composition))

Attributes:

Name Type Description
army_composition_dict dict[UnitTypeId, dict[str, float, str, int]]

A dictionary detailing how an army composition should be made up. The proportional values should all add up to 1.0, with a priority integer for unit emphasis.
freeflow_mode bool

If set to True, army composition proportions are ignored, and resources will be spent freely. Defaults to False.
ignore_proportions_below_unit_count int

In early game, units affect the army proportions significantly. This allows some units to be freely built before proportions are respected. Defaults to 0.
over_produce_on_low_tech bool

If only one tech is available for a unit, this allows that unit to be constantly produced. Defaults to True.
ignored_build_from_tags set[int]

A set of tags to prevent the spawn controller from morphing from these tags. Example: Prevent selecting barracks that need to build an addon.
maximum int

The maximum number of a unit type that can be produced in a single step. Defaults to 20.
spawn_target Optional[Point2]

A location to prioritize spawning units near.
Source code in src/ares/behaviors/macro/spawn_controller.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
@dataclass
class SpawnController(MacroBehavior):
    """Handle spawning army compositions.

    Example bot code:
    ```py
    from ares.behaviors.spawn_controller import SpawnController

    # Note: This does not try to build production facilities and
    # will ignore units that are impossible to currently spawn.
    army_composition: dict[UnitID: {float, bool}] = {
        UnitID.MARINE: {"proportion": 0.6, "priority": 2},  # lowest priority
        UnitID.MEDIVAC: {"proportion": 0.25, "priority": 1},
        UnitID.SIEGETANK: {"proportion": 0.15, "priority": 0},  # highest priority
    }
    # where `self` is an `AresBot` object
    self.register_behavior(SpawnController(army_composition))

    ```

    Attributes:
        army_composition_dict: A dictionary detailing how an army
            composition should be made up. The proportional values should
            all add up to 1.0, with a priority integer for unit emphasis.
        freeflow_mode: If set to True, army composition proportions are ignored,
            and resources will be spent freely.
            Defaults to `False`.
        ignore_proportions_below_unit_count: In early game, units affect the
            army proportions significantly. This allows some units to be freely
            built before proportions are respected. Defaults to `0`.
        over_produce_on_low_tech: If only one tech is available for a unit,
            this allows that unit to be constantly produced.
            Defaults to `True`.
        ignored_build_from_tags: A set of tags to prevent the spawn controller
            from morphing from these tags.
            Example: Prevent selecting barracks that need to build an addon.
        maximum: The maximum number of a unit type that can be produced
            in a single step. Defaults to `20`.
        spawn_target: A location to prioritize spawning units near.
        Defaults to `None`.


    """

    army_composition_dict: dict[UnitID, dict[str, float, str, int]]
    freeflow_mode: bool = False
    ignore_proportions_below_unit_count: int = 0
    over_produce_on_low_tech: bool = True
    ignored_build_from_tags: set[int] = field(default_factory=set)
    maximum: int = 20
    spawn_target: Optional[Point2] = None

    # key: Unit that should get a build order, value: what UnitID to build
    __build_dict: dict[Unit, UnitID] = field(default_factory=dict)
    # already used tags
    __excluded_structure_tags: set[int] = field(default_factory=set)
    __supply_available: float = 0.0

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        # allow gateways to morph before issuing commands
        if UpgradeId.WARPGATERESEARCH in ai.state.upgrades and [
            g
            for g in mediator.get_own_structures_dict[UnitID.GATEWAY]
            if g.is_ready and g.is_idle
        ]:
            return False

        self.__supply_available = ai.supply_left

        army_comp_dict: dict = self.army_composition_dict
        assert isinstance(
            army_comp_dict, dict
        ), f"self.army_composition_dict should be dict type, got {type(army_comp_dict)}"

        # get the current standing army based on the army comp dict
        # note we don't consider units outside the army comp dict
        unit_types: list[UnitID] = [*army_comp_dict]
        num_total_units: int = 0
        for unit_type in unit_types:
            num_total_units += mediator.get_own_unit_count(unit_type_id=unit_type)

        check_proportion: bool = True
        proportion_sum: float = 0.0
        # remember units that meet tech requirement
        units_ready_to_build: list[UnitID] = []
        # keep track of what units we have tech for
        tech_ready_for: list[UnitID] = []
        # iterate through desired army comp starting with the highest priority unit
        for unit_type_id, army_comp_info in sorted(
            army_comp_dict.items(), key=lambda x: x[1].get("priority", int(0))
        ):
            assert isinstance(unit_type_id, UnitID), (
                f"army_composition_dict expects UnitTypeId type as keys, "
                f"got {type(unit_type_id)}"
            )
            priority: int = army_comp_info["priority"]
            assert 0 <= priority < 11, (
                f"Priority for {unit_type_id} is set to {priority},"
                f"it should be an integer between 0 - 10."
                f"Where 0 has highest priority."
            )

            target_proportion: float = army_comp_info["proportion"]
            proportion_sum += target_proportion

            # work out if we are able to produce this unit
            if not ai.tech_ready_for_unit(unit_type_id):
                continue

            tech_ready_for.append(unit_type_id)

            trained_from: set[UnitID]
            if unit_type_id == UnitID.ARCHON:
                trained_from = {UnitID.HIGHTEMPLAR, UnitID.DARKTEMPLAR}
            else:
                trained_from = UNIT_TRAINED_FROM[unit_type_id]

            # get all idle build structures/units we can create this unit from
            build_structures: list[Unit] = ai.get_build_structures(
                trained_from,
                unit_type_id,
                self.__build_dict,
                self.ignored_build_from_tags,
            )
            # there is no possible way to build this unit, skip even if higher priority
            if len(build_structures) == 0:
                continue

            # archon is a special case that can't be handled generically
            if unit_type_id == UnitID.ARCHON:
                self._handle_archon_morph(ai, build_structures, mediator)
                continue

            # prioritize spawning close to spawn target
            if self.spawn_target:
                build_structures = cy_sorted_by_distance_to(
                    build_structures, self.spawn_target
                )

            # can't afford unit?
            # then we might want to break out loop till we can afford
            if not self._can_afford(ai, unit_type_id):
                if (
                    self.freeflow_mode
                    or num_total_units < self.ignore_proportions_below_unit_count
                ):
                    continue
                # break out the loop, don't spend resources on lower priority units
                else:
                    check_proportion = False
                    break

            # keep track of which unit types the build_structures/ tech is ready for
            units_ready_to_build.append(unit_type_id)

            num_this_unit: int = mediator.get_own_unit_count(unit_type_id=unit_type_id)
            current_proportion: float = num_this_unit / (num_total_units + 1e-16)
            # already have enough of this unit type,
            # but we could add it if:
            # freeflow mode or we don't have much army yet
            if (
                current_proportion >= target_proportion
                and not self.freeflow_mode
                and num_total_units >= self.ignore_proportions_below_unit_count
            ):
                continue

            amount, supply, cost = self._calculate_build_amount(
                ai,
                unit_type_id,
                build_structures,
                self.__supply_available,
                self.maximum,
            )
            self._add_to_build_dict(
                ai, unit_type_id, build_structures, amount, supply, cost
            )
        # if we can only build one type of unit, keep adding them
        if (
            len(tech_ready_for) == 1
            and self.over_produce_on_low_tech
            and len(units_ready_to_build) > 0
            and self.maximum > 1
        ):
            build_structures = ai.get_build_structures(
                UNIT_TRAINED_FROM[units_ready_to_build[0]],
                units_ready_to_build[0],
                self.__build_dict,
                self.ignored_build_from_tags,
            )
            amount, supply, cost = self._calculate_build_amount(
                ai, units_ready_to_build[0], build_structures, self.__supply_available
            )
            # prioritize spawning close to spawn target
            if self.spawn_target:
                build_structures = cy_sorted_by_distance_to(
                    build_structures, self.spawn_target
                )
            self._add_to_build_dict(
                ai, units_ready_to_build[0], build_structures, amount, supply, cost
            )

        if check_proportion and not self.freeflow_mode:
            assert isclose(
                proportion_sum, 1.0
            ), f"The army comp proportions should equal 1.0, got {proportion_sum}"

        return self._morph_units(ai, mediator)

    def _add_to_build_dict(
        self,
        ai: "AresBot",
        type_id: UnitID,
        base_unit: list[Unit],
        amount: int,
        supply_cost: float,
        cost_per_unit: Cost,
    ) -> None:
        """Execute the spawn controller task (Called from `behavior_executioner.py`).

        Handle unit production as per the .........

        Parameters
        ----------
        ai :
            Bot object that will be running the game
        type_id :
            Type of unit we want to spawn.
        base_unit :
            Unit objects we can spawn this unit from.
        amount :
            How many type_id we intend to spawn.
        supply_cost :
            Supply cost of spawning type_id amount.
        cost_per_unit :
            Minerals and vespene cost.
        """
        # min check to make sure we don't pop from empty lists
        for _ in range(min(len(base_unit), amount)):
            self.__build_dict[base_unit.pop(0)] = type_id
            ai.minerals -= cost_per_unit.minerals
            ai.vespene -= cost_per_unit.vespene
            self.__supply_available -= supply_cost

    @staticmethod
    def _calculate_build_amount(
        ai: "AresBot",
        unit_type: UnitID,
        base_units: list[Unit],
        supply_left: float,
        maximum: int = 20,
    ) -> tuple[int, float, Cost]:
        """Execute the spawn controller task (Called from `behavior_executioner.py`).

        Handle unit production as per the .........

        Parameters
        ----------
        ai :
            Bot object that will be running the game
        unit_type :
            Type of unit we want to spawn.
        base_units :
            Unit objects we can spawn this unit from.
        supply_left :
            How much total supply we have available.
        maximum :
            A limit on how many units can be spawned in one go.
        """
        cost = ai.cost_dict[unit_type]
        supply_cost = ai.calculate_supply_cost(unit_type)
        amount = min(
            int(ai.minerals / cost.minerals) if cost.minerals else 9999999,
            int(ai.vespene / cost.vespene) if cost.vespene else 9999999,
            int(supply_left / supply_cost) if supply_cost else 9999999,
            len(base_units),
            maximum,
        )
        return amount, supply_cost, cost

    @staticmethod
    def _can_afford(ai: "AresBot", unit_type_id: UnitID) -> bool:
        if unit_type_id == UnitID.ARCHON:
            return True
        return ai.can_afford(unit_type_id)

    @staticmethod
    def _handle_archon_morph(
        ai: "AresBot", build_structures: list[Unit], mediator: ManagerMediator
    ) -> None:
        unit_role_dict: dict[UnitRole, set] = mediator.get_unit_role_dict
        build_structures = [
            b
            for b in build_structures
            if b.tag not in unit_role_dict[UnitRole.MORPHING] and b.is_ready
        ]
        if len(build_structures) < 2:
            return

        templar: list[Unit] = build_structures[:2]
        ai.request_archon_morph(templar)

    def _morph_units(self, ai: "AresBot", mediator: ManagerMediator) -> bool:
        did_action: bool = False
        for unit, value in self.__build_dict.items():
            did_action = True
            mediator.clear_role(tag=unit.tag)
            if value == UnitID.BANELING:
                unit(AbilityId.MORPHTOBANELING_BANELING)
            elif value == UnitID.RAVAGER:
                unit(AbilityId.MORPHTORAVAGER_RAVAGER)
            # prod building is warp gate, but we really
            # want to spawn from psionic field
            elif unit.type_id == UnitID.WARPGATE:
                mediator.request_warp_in(
                    build_from=unit, unit_type=value, target=self.spawn_target
                )
            else:
                unit.train(value)
                ai.num_larva_left -= 1

        return did_action

TechUp dataclass

Bases: MacroBehavior

Automatically tech up so desired upgrade/unit can be built.

Example: 

from ares.behaviors.macro import TechUp
from sc2.ids.upgrade_id import UpgradeId

self.register_behavior(
    TechUp(UpgradeId.BANSHEECLOAK, base_location=self.start_location)
)

Attributes:

Name Type Description
desired_tech Union[UpgradeId, UnitTypeId]

The desired upgrade or unit type.
base_location Point2

The main building location to make tech.
ignore_existing_techlabs bool

If set to True, will keep building techlabs even if others exist. Defaults to False.
Source code in src/ares/behaviors/macro/tech_up.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
@dataclass
class TechUp(MacroBehavior):
    """Automatically tech up so desired upgrade/unit can be built.

    Example:
    ```py
    from ares.behaviors.macro import TechUp
    from sc2.ids.upgrade_id import UpgradeId

    self.register_behavior(
        TechUp(UpgradeId.BANSHEECLOAK, base_location=self.start_location)
    )
    ```

    Attributes:
        desired_tech: The desired upgrade or unit type.
        base_location: The main building location to make tech.
        ignore_existing_techlabs: If set to `True`, will
            keep building techlabs even if others exist.
            Defaults to `False`.

    """

    desired_tech: Union[UpgradeId, UnitID]
    base_location: Point2
    ignore_existing_techlabs: bool = False

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        assert isinstance(
            self.desired_tech, (UpgradeId, UnitID)
        ), f"Wrong type provided for `desired_tech`, got {type(self.desired_tech)}"

        # figure out where we research this unit / upgrade from
        researched_from_id: UnitID
        tech_required: list[UnitID]
        if isinstance(self.desired_tech, UpgradeId):
            researched_from_id = UPGRADE_RESEARCHED_FROM[self.desired_tech]
            tech_required = UNIT_TECH_REQUIREMENT[researched_from_id]
        else:
            if self.desired_tech in ALL_STRUCTURES:
                researched_from_id = self.desired_tech
                tech_required = UNIT_TECH_REQUIREMENT[researched_from_id]
            else:
                researched_from_id = next(iter(UNIT_TRAINED_FROM[self.desired_tech]))
                if self.desired_tech in GATEWAY_UNITS:
                    researched_from_id = UnitID.GATEWAY
                tech_required = UNIT_TECH_REQUIREMENT[self.desired_tech]

        # special handling of teching to techlabs
        if researched_from_id in TECHLAB_TYPES:
            if ai.can_afford(researched_from_id) and self._adding_techlab(
                ai,
                self.base_location,
                researched_from_id,
                tech_required,
                self.desired_tech,
                self.ignore_existing_techlabs,
            ):
                return True
            return False

        # can we build this tech building right away?
        # 1.0 = Yes, < 1.0 = No
        tech_progress: float = ai.tech_requirement_progress(researched_from_id)

        # we have the tech ready to build this upgrade building right away :)
        if tech_progress == 1.0 and not ai.structure_present_or_pending(
            researched_from_id
        ):
            # need a gateway, but we have a warpgate already
            if (
                researched_from_id == UnitID.GATEWAY
                and mediator.get_own_structures_dict[UnitID.WARPGATE]
            ):
                return False
            logger.info(
                f"{ai.time_formatted} Building {researched_from_id} "
                f"for {self.desired_tech}"
            )
            return BuildStructure(ai.start_location, researched_from_id).execute(
                ai, config, mediator
            )

        # we can't even build the upgrade building :(
        # figure out what to build to get there
        else:
            for structure_type in tech_required:
                checks: list[UnitID] = [structure_type]
                if structure_type == UnitID.GATEWAY:
                    checks.append(UnitID.WARPGATE)

                if structure_type in TECHLAB_TYPES:
                    if self._adding_techlab(
                        ai,
                        self.base_location,
                        structure_type,
                        tech_required,
                        self.desired_tech,
                        self.ignore_existing_techlabs,
                    ):
                        return True
                    continue

                if any(ai.structure_present_or_pending(check) for check in checks):
                    continue

                # found something to build?
                if ai.tech_requirement_progress(structure_type) == 1.0:
                    building: bool = BuildStructure(
                        self.base_location, structure_type
                    ).execute(ai, ai.config, ai.mediator)
                    if building:
                        logger.info(
                            f"{ai.time_formatted} Adding {structure_type} to"
                            f" tech towards {self.desired_tech}"
                        )
                    return building

        return False

    def _adding_techlab(
        self,
        ai: "AresBot",
        base_location: Point2,
        researched_from_id: UnitID,
        tech_required: list[UnitID],
        desired_tech: Union[UnitID, UpgradeId],
        ignore_existing_techlabs: bool = False,
    ) -> bool:
        """
        ai :
        base_location :
        researched_from_id :
            The building we require
        tech_required :
            The list of tech buildings we would need to build this
        desired_tech :
            The thing we are actually trying to research / build
        """
        structures_dict: dict = ai.mediator.get_own_structures_dict
        build_techlab_from: UnitID = BUILD_TECHLAB_FROM[researched_from_id]
        _build_techlab_from_structures: list[Unit] = structures_dict[
            build_techlab_from
        ].copy()
        # looks like we already have what we are looking for?
        if (
            not ignore_existing_techlabs
            and len(
                [
                    s
                    for s in _build_techlab_from_structures
                    if s.has_techlab and s.is_idle
                ]
            )
            > 0
        ):
            return False

        # no possible way of building this techlab, tech towards it
        if not ai.structure_present_or_pending(build_techlab_from):
            for s in tech_required:
                if ai.structure_present_or_pending(s):
                    continue
                if TechUp(s, base_location).execute(ai, ai.config, ai.mediator):
                    logger.info(
                        f"{ai.time_formatted} Adding {s} so that we can "
                        f"tech towards {desired_tech}"
                    )
                    return True
            # no point continuing
            return False

        without_techlabs: list[Unit] = [
            s
            for s in _build_techlab_from_structures
            if s.is_ready and s.is_idle and not s.has_add_on
        ]
        if without_techlabs:
            without_techlabs[0].build(researched_from_id)
            logger.info(
                f"{ai.time_formatted} Adding {researched_from_id} so that we can "
                f"tech towards {desired_tech}"
            )
            return True
        return False

UpgradeController dataclass

Bases: MacroBehavior

Research upgrades, if the upgrade is not currently researchable this behavior will automatically make the tech buildings required.

Example: 

from ares.behaviors.macro import UpgradeController
from sc2.ids.upgrade_id import UpgradeId

desired_upgrades: list[UpgradeId] = [
    UpgradeId.TERRANINFANTRYWEAPONSLEVEL1,
    UpgradeId.TERRANINFANTRYWEAPONSLEVEL2,
    UpgradeId.BANSHEECLOAK,
    UpgradeId.PERSONALCLOAKING
]

self.register_behavior(
    UpgradeController(desired_upgrades, base_location=self.start_location)
)

Attributes:

Name Type Description
upgrade_list list[UpgradeId]

List of desired upgrades.
base_location Point2

Location to build upgrade buildings.
Source code in src/ares/behaviors/macro/upgrade_controller.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass
class UpgradeController(MacroBehavior):
    """Research upgrades, if the upgrade is not
    currently researchable this behavior will automatically
    make the tech buildings required.


    Example:
    ```py
    from ares.behaviors.macro import UpgradeController
    from sc2.ids.upgrade_id import UpgradeId

    desired_upgrades: list[UpgradeId] = [
        UpgradeId.TERRANINFANTRYWEAPONSLEVEL1,
        UpgradeId.TERRANINFANTRYWEAPONSLEVEL2,
        UpgradeId.BANSHEECLOAK,
        UpgradeId.PERSONALCLOAKING
    ]

    self.register_behavior(
        UpgradeController(desired_upgrades, base_location=self.start_location)
    )
    ```

    Attributes:
        upgrade_list: List of desired upgrades.
        base_location: Location to build upgrade buildings.

    """

    upgrade_list: list[UpgradeId]
    base_location: Point2

    def execute(self, ai: "AresBot", config: dict, mediator: ManagerMediator) -> bool:
        for upgrade in self.upgrade_list:
            if ai.pending_or_complete_upgrade(upgrade):
                continue

            researched_from_id: UnitID = UPGRADE_RESEARCHED_FROM[upgrade]
            researched_from: list[Unit] = [
                s for s in mediator.get_own_structures_dict[researched_from_id]
            ]

            # there is nowhere to research this from, tech up to it
            if not researched_from:
                teching: bool = TechUp(
                    desired_tech=upgrade, base_location=self.base_location
                ).execute(ai, config, mediator)
                # we've carried out an action, return True
                if teching:
                    return True

            # we have somewhere to research from, if it's possible
            # carry out the action
            else:
                idle: list[Unit] = [
                    s
                    for s in researched_from
                    if s.is_ready
                    and s.is_idle
                    and (not ai.race == Race.Protoss or s.is_powered)
                ]
                if idle and ai.can_afford(upgrade):
                    building: Unit = idle[0]
                    research_info: dict = RESEARCH_INFO[researched_from_id][upgrade]
                    ability: AbilityId = research_info["ability"]
                    if ability in building.abilities:
                        building.research(upgrade)
                        logger.info(f"{ai.time_formatted}: Researching {upgrade}")
                        return True
                    # there is a structure to upgrade from, but:
                    # we can't do the upgrade, might need something like:
                    # hive for 3/3? twilight council for 2/2?
                    elif required_building := research_info.get(
                        "required_building", None
                    ):
                        return TechUp(
                            desired_tech=required_building,
                            base_location=self.base_location,
                        ).execute(ai, config, mediator)

        # found nothing to do
        return False