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

macros : list[Behavior] (optional, default: []) A list of behaviors that should be executed

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
56
@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 : list[Behavior] (optional, default: [])
        A list of behaviors that should be executed
    """

    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

structure_needing_addon : Unit The structure type we want the addon for addon_required : UnitID 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
89
90
91
@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 : Unit
        The structure type we want the addon for
    addon_required : UnitID
        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

base_location : Point2 The base location where supply should be built. return_true_if_supply_required : bool Can't afford supply, but it's required, return true? Useful if creating a MacroPlan

Returns

bool : True if this Behavior carried out an action.

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
127
128
129
130
131
132
133
@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 : Point2
        The base location where supply should be built.
    return_true_if_supply_required : bool
        Can't afford supply, but it's required, return true?
        Useful if creating a `MacroPlan`

    Returns
    ----------
    bool :
        True if this Behavior carried out an action.
    """

    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

base_location : Point2 The base location to build near. structure_id : UnitTypeId The structure type we want to build. wall : bool Find wall placement if possible. (Only main base currently supported) closest_to : Point2 (optional) Find placement at this base closest to

Source code in src/ares/behaviors/macro/build_structure.py 
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@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 : Point2
        The base location to build near.
    structure_id : UnitTypeId
        The structure type we want to build.
    wall : bool
        Find wall placement if possible.
        (Only main base currently supported)
    closest_to : Point2 (optional)
        Find placement at this base closest to
    """

    base_location: Point2
    structure_id: UnitID
    wall: bool = False
    closest_to: Optional[Point2] = None

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

        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,
            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

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

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
49
@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 : int
        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

to_count : int The target base count. can_afford_check : bool, optional Check we can afford expansion. Setting this to False will allow worker to move to location ready to build expansion. (default is True) check_location_is_safe : bool, optional Check we don't knowingly expand at a dangerous location. (default is True) max_pending : int, optional Maximum pending townhalls at any time. (default is 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
117
118
119
120
121
122
123
@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 : int
        The target base count.
    can_afford_check : bool, optional
        Check we can afford expansion.
        Setting this to False will allow worker to move
        to location ready to build expansion.
        (default is `True`)
    check_location_is_safe : bool, optional
        Check we don't knowingly expand at a dangerous location.
        (default is `True`)
    max_pending : int, optional
        Maximum pending townhalls at any time.
        (default is `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

to_count : int How many gas buildings would we like? max_pending : int (optional) How many gas pending at any time? closest_to : Point2 (optional) Find available geyser closest to this location.

Source code in src/ares/behaviors/macro/gas_building_controller.py 
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
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 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 : int
        How many gas buildings would we like?
    max_pending : int (optional)
        How many gas pending at any time?
    closest_to : Point2 (optional)
        Find available geyser closest to this location.
    """

    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 = (
            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

flee_at_health_perc : float, optional If worker is in danger, at what health perc should it flee (default is 0.5). keep_safe : bool, optional Workers flee if they are in danger? (default is True). long_distance_mine : bool, optional If worker has nothing to do, can it long distance mine (default is True). mineral_boost : bool, optional Turn mineral boosting off / on (default is True). vespene_boost : bool, optional Turn vespene boosting off / on (only active when workers_per_gas < 3) WARNING: VESPENE BOOSTING CURRENTLY NOT WORKING (default is True). workers_per_gas : bool, optional (default: 3) Control how many workers are assigned to each gas. self_defence_active : bool, optional (default: True) If set to True, workers will have some basic defence. Certain workers will attack enemy in range. safe_long_distance_mineral_fields : Optional[Units], optional (default is None) Used internally, value is set if a worker starts long distance mining.

Source code in src/ares/behaviors/macro/mining.py 
 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
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
@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 : float, optional
        If worker is in danger, at what health perc should it flee (default is 0.5).
    keep_safe : bool, optional
        Workers flee if they are in danger? (default is True).
    long_distance_mine : bool, optional
        If worker has nothing to do, can it long distance mine (default is True).
    mineral_boost : bool, optional
        Turn mineral boosting off / on (default is True).
    vespene_boost : bool, optional
        Turn vespene boosting off / on (only active when workers_per_gas < 3)
        WARNING: VESPENE BOOSTING CURRENTLY NOT WORKING
        (default is True).
    workers_per_gas : bool, optional (default: 3)
        Control how many workers are assigned to each gas.
    self_defence_active : bool, optional (default: True)
        If set to True, workers will have some basic defence.
        Certain workers will attack enemy in range.
    safe_long_distance_mineral_fields : Optional[Units], optional (default is None)
        Used internally, value is set if a worker starts long distance mining.
    """

    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)

    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)
                )
            ):
                self._keep_worker_safe(
                    mediator,
                    grid,
                    worker,
                    resource,
                    resource_position,
                    dist_to_resource,
                )

            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,
                    resource,
                    resource_position,
                    dist_to_resource,
                )

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

    @staticmethod
    def _keep_worker_safe(
        mediator: ManagerMediator,
        grid: np.ndarray,
        worker: Unit,
        resource: Unit,
        resource_position: Point2,
        dist_to_resource: float,
    ) -> 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.
        resource :
            Resource worker is assigned to.
        resource_position :
            Resource this worker is assigned to.
        dist_to_resource :
            How far away are we from intended resource?

        Returns
        -------
        """
        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:
            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,
    ) -> None:
        """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.
        """

        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}"
                )
        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)

        closest_th: Unit = cy_closest_to(worker_position, ai.townhalls)

        # fix realtime bug where worker is stuck with a move command but already
        # returned minerals
        if (
            len(worker.orders) == 1
            and worker.orders[0].ability.id == AbilityId.MOVE
            and worker.order_target == closest_th.tag
        ):
            worker(AbilityId.SMART, target)

        elif (worker.is_returning or worker.is_carrying_resource) and len(
            worker.orders
        ) < 2:
            try:
                townhall: Unit = ai.unit_tag_dict[
                    worker_tag_to_townhall_tag[worker.tag]
                ]
            except KeyError:
                townhall: Unit = closest_th

            target_pos: Point2 = townhall.position

            target_pos: Point2 = Point2(
                cy_towards(
                    target_pos,
                    worker_position,
                    TOWNHALL_RADIUS * distance_to_townhall_factor,
                )
            )

            if 0.5625 < cy_distance_to_squared(worker_position, target_pos) < 4.0:
                worker.move(target_pos)
                worker(AbilityId.SMART, townhall, True)
            # not at right distance to get boost command, but doesn't have return
            # resource command for some reason
            elif not worker.is_returning:
                worker(AbilityId.SMART, townhall)

        elif not worker.is_returning and len(worker.orders) < 2:
            min_distance: float = 0.5625 if target.is_mineral_field else 0.01
            max_distance: float = 4.0 if target.is_mineral_field else 0.25
            if (
                min_distance
                < cy_distance_to_squared(worker_position, resource_target_pos)
                < max_distance
                or worker.is_idle
            ):
                worker.move(resource_target_pos)
                worker(AbilityId.SMART, target, True)

        # on rare occasion above conditions don't hit and worker goes idle
        elif worker.is_idle or not worker.is_moving:
            if worker.is_carrying_resource:
                worker.return_resource(closest_th)
            else:
                worker.gather(target)

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

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

        Returns
        -------
            Optional[Units] :
                Units object of safe mineral patches if mineral patches still exist,
                None otherwise.
        """
        if not ai.mineral_field:
            return

        th_tags: set[int] = ai.townhalls.tags
        mf_to_enemy: dict[int, Units] = mediator.get_units_in_range(
            start_points=ai.mineral_field,
            distances=30,
            query_tree=UnitTreeQueryType.AllEnemy,
            return_as_dict=True,
        )

        mf_to_own: dict[int, Units] = mediator.get_units_in_range(
            start_points=ai.mineral_field,
            distances=8,
            query_tree=UnitTreeQueryType.AllOwn,
            return_as_dict=True,
        )

        enemy_ground_dangerous_tags: set[int] = ai.enemy_units.filter(
            lambda e: e.can_attack_ground and e.type_id not in WORKER_TYPES
        ).tags

        safe_fields = []
        for mf in ai.mineral_field:
            if th_tags & mf_to_own[mf.tag].tags:  # intersection
                # there's a shared tag in our units close to the mineral field and our
                # townhalls
                continue
            else:
                found = False
                for tag in mf_to_enemy[mf.tag].tags:
                    if tag in enemy_ground_dangerous_tags:
                        # there's an enemy nearby
                        found = True
                        break
                if found:
                    continue
                safe_fields.append(mf)
        return Units(safe_fields, ai)

    @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

army_composition_dict : dict[UnitID: float, bool] A dictionary that details how an army composition should be made up. The proportional values should all add up to 1.0. With a priority integer to give units emphasis. base_location : Point2 Where abouts do we build production? add_production_at_bank : Tuple[int, int], optional When we reach this bank size, work out what extra production would be useful. Tuple where first value is minerals and second is vespene. (default = (300, 300)) alpha : float, optional Controls how much production to add when bank is higher than add_production_at_bank. (default = 0.9) unit_pending_progress : float, optional 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. (default = 0.8) ignore_below_proportion: float, optional If we don't want many of this unit, no point adding production. Will check if possible to build unit first. Default is 0.05 should_repower_structures: bool, optional Search for unpowered structures, and build a new pylon as needed. Default is 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
309
310
311
312
313
314
315
316
317
318
319
@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 : dict[UnitID: float, bool]
        A dictionary that details how an army composition should be made up.
        The proportional values should all add up to 1.0.
        With a priority integer to give units emphasis.
    base_location : Point2
        Where abouts do we build production?
    add_production_at_bank : Tuple[int, int], optional
        When we reach this bank size, work out what extra production
        would be useful.
        Tuple where first value is minerals and second is vespene.
        (default = `(300, 300)`)
    alpha : float, optional
        Controls how much production to add when bank is
        higher than `add_production_at_bank`.
        (default = `0.9`)
    unit_pending_progress : float, optional
        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.
         (default = 0.8)
    ignore_below_proportion: float, optional
        If we don't want many of this unit, no point adding production.
        Will check if possible to build unit first.
        Default is `0.05`
    should_repower_structures: bool, optional
        Search for unpowered structures, and build a new
        pylon as needed.
        Default is `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

army_composition_dict : dict[UnitID: float, bool] A dictionary that details how an army composition should be made up. The proportional values should all add up to 1.0. With a priority integer to give units emphasis. freeflow_mode : bool (default: False) If set to True, army comp proportions are ignored, and resources will be spent freely. ignore_proportions_below_unit_count : int (default 0) In early game units effect the army proportions significantly. This allows some units to be freely built before proportions are respected. over_produce_on_low_tech : bool (default True) If there is only tech available for one unit, this will allow this unit to be constantly produced. ignored_build_from_tags : set[int] Prevent spawn controller from morphing from these tags Example: Prevent selecting barracks that needs to build an addon maximum : int (default 20) The maximum unit type we can produce in a single step. spawn_target : Union[Point2, None] (default None) Prioritize spawning units near this location.

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
344
@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 : dict[UnitID: float, bool]
        A dictionary that details how an army composition should be made up.
        The proportional values should all add up to 1.0.
        With a priority integer to give units emphasis.
    freeflow_mode : bool (default: False)
        If set to True, army comp proportions are ignored, and resources
        will be spent freely.
    ignore_proportions_below_unit_count : int (default 0)
        In early game units effect the army proportions significantly.
        This allows some units to be freely built before proportions are respected.
    over_produce_on_low_tech : bool (default True)
        If there is only tech available for one unit, this will allow this
        unit to be constantly produced.
    ignored_build_from_tags : set[int]
        Prevent spawn controller from morphing from these tags
        Example: Prevent selecting barracks that needs to build an addon
    maximum : int (default 20)
        The maximum unit type we can produce in a single step.
    spawn_target : Union[Point2, None] (default None)
        Prioritize spawning units near this location.

    """

    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

desired_tech : Union[UpgradeId, UnitTypeId] What is the desired thing we want? base_location : bool The main building location to make tech. ignore_existing_techlabs : bool Will keep building techlabs even if others exist (default = False)

Returns

bool : True if this Behavior carried out an action.

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
212
213
214
215
216
217
218
219
@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 : Union[UpgradeId, UnitTypeId]
        What is the desired thing we want?
    base_location : bool
        The main building location to make tech.
    ignore_existing_techlabs : bool
        Will keep building techlabs even if others exist
        (default = False)

    Returns
    ----------
    bool :
        True if this Behavior carried out an action.

    """

    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

upgrade_list : list[UpgradeId] List of desired upgrades. base_location : Point2 Where to build upgrade buildings.

Returns

bool : True if this Behavior carried out an action.

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
106
107
108
109
110
111
112
@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[UpgradeId]
        List of desired upgrades.
    base_location : Point2
        Where to build upgrade buildings.

    Returns
    ----------
    bool :
        True if this Behavior carried out an action.
    """

    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