Fix pyrefly type checker issues

.github/workflows/ci.yml

@@ -51,7 +51,7 @@ jobs:
           uv run pre-commit install
 
       - name: Run pre-commit hooks
-        run: uv run pre-commit run --all-files --hook-stage push
+        run: uv run pre-commit run --all-files --hook-stage pre-push
 
   generate_dicts_from_data_json:
     name: Generate dicts from data.json
@@ -80,7 +80,7 @@ jobs:
         run: |
           mv sc2/dicts sc2/dicts_old
           uv run python generate_dicts_from_data_json.py
-          uv run pre-commit run --all-files --hook-stage push || true
+          uv run pre-commit run --all-files --hook-stage pre-push || true
           rm -rf sc2/dicts/__pycache__ sc2/dicts_old/__pycache__
 
       - name: Upload generated dicts folder as artifact

.pre-commit-config.yaml

@@ -38,22 +38,25 @@ repos:
   # Autoformat code
   - id: ruff-format-check
     name: Check if files are formatted
-    stages: [push]
+    stages: [pre-push]
     language: system
+    # Run the following command to fix:
+    # uv run ruff format . 
     entry: uv run ruff format . --check --diff
     pass_filenames: false
 
   - id: ruff-lint
     name: Lint files
-    stages: [push]
+    stages: [pre-push]
     language: system
+    # Run the following command to fix:
+    # uv run ruff check . --fix
     entry: uv run ruff check .
     pass_filenames: false
 
-  # TODO Fix issues
-  # - id: pyrefly
-  #   name: Static types checking with pyrefly
-  #   stages: [push]
-  #   language: system
-  #   entry: uv run pyrefly check
-  #   pass_filenames: false
+  - id: pyrefly
+    name: Static types checking with pyrefly
+    stages: [pre-push]
+    language: system
+    entry: uv run pyrefly check
+    pass_filenames: false

README.md

@@ -186,5 +186,5 @@ Git commit messages use [imperative-style messages](https://stackoverflow.com/a/
 To run pre-commit hooks (which run autoformatting and autosort imports) you can run
 ```sh
 uv run pre-commit install
-uv run pre-commit run --all-files --hook-stage push
+uv run pre-commit run --all-files --hook-stage pre-push
 ```

dockerfiles/test_docker_image.sh

@@ -41,7 +41,7 @@ docker cp test test_container:/root/python-sc2/test
 # Run various test bots
 docker exec -i test_container bash -c "cd python-sc2 && uv run python test/travis_test_script.py test/autotest_bot.py"
 docker exec -i test_container bash -c "cd python-sc2 && uv run python test/travis_test_script.py test/queries_test_bot.py"
-#docker exec -i test_container bash -c "cd python-sc2 && uv run python test/travis_test_script.py test/damagetest_bot.py"
+docker exec -i test_container bash -c "cd python-sc2 && uv run python test/travis_test_script.py test/damagetest_bot.py"
 
 docker cp examples test_container:/root/python-sc2/examples
 docker exec -i test_container bash -c "cd python-sc2 && uv run python test/run_example_bots_vs_computer.py"

examples/arcade_bot.py

@@ -101,7 +101,6 @@ def position_around_unit(
         step_size: int = 1,
         exclude_out_of_bounds: bool = True,
     ):
-        # pyre-ignore[16]
         pos = pos.position.rounded
         positions = {
             pos.offset(Point2((x, y)))
@@ -114,7 +113,6 @@ def position_around_unit(
             positions = {
                 p
                 for p in positions
-                # pyre-ignore[16]
                 if 0 <= p[0] < self.game_info.pathing_grid.width and 0 <= p[1] < self.game_info.pathing_grid.height
             }
         return positions

examples/bot_vs_bot.py

@@ -15,7 +15,7 @@
 
 
 def main_old():
-    result: list[Result] = run_game(
+    result: Result | list[Result | None] = run_game(
         maps.get("AcropolisLE"),
         [
             Bot(Race.Protoss, WarpGateBot()),

examples/competitive/__init__.py

@@ -4,8 +4,9 @@
 import aiohttp
 from loguru import logger
 
-import sc2
 from sc2.client import Client
+from sc2.main import _play_game
+from sc2.portconfig import Portconfig
 from sc2.protocol import ConnectionAlreadyClosedError
 
 
@@ -41,7 +42,7 @@ def run_ladder_game(bot):
     else:
         ports = [lan_port + p for p in range(1, 6)]
 
-        portconfig = sc2.portconfig.Portconfig()
+        portconfig = Portconfig()
         portconfig.server = [ports[1], ports[2]]
         portconfig.players = [[ports[3], ports[4]]]
 
@@ -56,10 +57,11 @@ def run_ladder_game(bot):
 # Modified version of sc2.main._join_game to allow custom host and port, and to not spawn an additional sc2process (thanks to alkurbatov for fix)
 async def join_ladder_game(host, port, players, realtime, portconfig, save_replay_as=None, game_time_limit=None):
     ws_url = f"ws://{host}:{port}/sc2api"
+    # pyrefly: ignore
     ws_connection = await aiohttp.ClientSession().ws_connect(ws_url, timeout=120)
     client = Client(ws_connection)
     try:
-        result = await sc2.main._play_game(players[0], client, realtime, portconfig, game_time_limit)
+        result = await _play_game(players[0], client, realtime, portconfig, game_time_limit)
         if save_replay_as is not None:
             await client.save_replay(save_replay_as)
         # await client.leave()
@@ -68,6 +70,6 @@ async def join_ladder_game(host, port, players, realtime, portconfig, save_repla
         logger.error("Connection was closed before the game ended")
         return None
     finally:
-        ws_connection.close()
+        await ws_connection.close()
 
     return result

examples/competitive/bot.py

@@ -11,7 +11,6 @@ async def on_step(self, iteration: int):
         # Populate this function with whatever your bot should do!
         pass
 
-    # pyre-ignore[11]
     async def on_end(self, game_result: Result):
         print("Game ended.")
         # Do things here after the game ends

examples/competitive/run.py

@@ -1,10 +1,9 @@
-# pyre-ignore-all-errors[16, 21]
 import sys
 
-from __init__ import run_ladder_game
+from examples.competitive.__init__ import run_ladder_game
 
 # Load bot
-from bot import CompetitiveBot
+from examples.competitive.bot import CompetitiveBot
 
 from sc2 import maps
 from sc2.data import Difficulty, Race

examples/distributed_workers.py

@@ -1,4 +1,3 @@
-# pyre-ignore-all-errors[16]
 from sc2 import maps
 from sc2.bot_ai import BotAI
 from sc2.data import Difficulty, Race

examples/fastreload.py

@@ -20,7 +20,8 @@ def main():
         input("Press enter to reload ")
 
         reload(zerg_rush)
-        player_config[0].ai = zerg_rush.ZergRushBot()
+        if isinstance(player_config[0], Bot):
+            player_config[0].ai = zerg_rush.ZergRushBot()
         gen.send(player_config)
 
 

examples/host_external_norestart.py

@@ -10,6 +10,7 @@ def main():
     portconfig: Portconfig = Portconfig()
     print(portconfig.as_json)
 
+    # pyrefly: ignore
     player_config = [Bot(Race.Zerg, ZergRushBot()), Bot(Race.Zerg, None)]
 
     for g in _host_game_iter(maps.get("Abyssal Reef LE"), player_config, realtime=False, portconfig=portconfig):

examples/protoss/find_adept_shades.py

@@ -18,15 +18,14 @@ def __init__(self):
     async def on_start(self):
         self.client.game_step = 2
         await self.client.debug_create_unit(
-            [[UnitTypeId.ADEPT, 10, self.townhalls[0].position.towards(self.game_info.map_center, 5), 1]]
+            [(UnitTypeId.ADEPT, 10, self.townhalls[0].position.towards(self.game_info.map_center, 5), 1)]
         )
 
     async def on_step(self, iteration: int):
         adepts = self.units(UnitTypeId.ADEPT)
         if adepts and not self.shaded:
             # Wait for adepts to spawn and then cast ability
             for adept in adepts:
-                # pyre-ignore[16]
                 adept(AbilityId.ADEPTPHASESHIFT_ADEPTPHASESHIFT, self.game_info.map_center)
             self.shaded = True
         elif self.shades_mapping:
@@ -38,7 +37,6 @@ async def on_step(self, iteration: int):
                     # logger.info(f"Remaining shade time: {shade.buff_duration_remain} / {shade.buff_duration_max}")
                     pass
                 if adept and shade:
-                    # pyre-ignore[16]
                     self.client.debug_line_out(adept, shade, (0, 255, 0))
             # logger.info(self.shades_mapping)
         elif self.shaded:
@@ -53,7 +51,7 @@ async def on_step(self, iteration: int):
                 previous_shade_location = shade.position.towards(
                     forward_position, -(self.client.game_step / 16) * shade.movement_speed
                 )  # See docstring of movement_speed attribute
-                # pyre-ignore[6]
+
                 closest_adept = remaining_adepts.closest_to(previous_shade_location)
                 self.shades_mapping[closest_adept.tag] = shade.tag
 

examples/protoss/threebase_voidray.py

@@ -94,16 +94,12 @@ async def on_step(self, iteration: int):
             for nexus in self.townhalls.ready:
                 vgs = self.vespene_geyser.closer_than(15, nexus)
                 for vg in vgs:
-                    if not self.can_afford(UnitTypeId.ASSIMILATOR):
-                        break
-
-                    worker = self.select_build_worker(vg.position)
-                    if worker is None:
-                        break
-
-                    if not self.gas_buildings or not self.gas_buildings.closer_than(1, vg):
-                        worker.build_gas(vg)
-                        worker.stop(queue=True)
+                    if self.can_afford(UnitTypeId.ASSIMILATOR):
+                        worker = self.select_build_worker(vg.position)
+                        if worker is not None:
+                            if not self.gas_buildings or not self.gas_buildings.closer_than(1, vg):
+                                worker.build_gas(vg)
+                                worker.stop(queue=True)
 
         # If we have less than 3  but at least 3 nexuses, build stargate
         if self.structures(UnitTypeId.PYLON).ready and self.structures(UnitTypeId.CYBERNETICSCORE).ready:

examples/protoss/warpgate_push.py

@@ -79,14 +79,12 @@ async def on_step(self, iteration: int):
         for nexus in self.townhalls.ready:
             vgs = self.vespene_geyser.closer_than(15, nexus)
             for vg in vgs:
-                if not self.can_afford(UnitTypeId.ASSIMILATOR):
-                    break
-                worker = self.select_build_worker(vg.position)
-                if worker is None:
-                    break
-                if not self.gas_buildings or not self.gas_buildings.closer_than(1, vg):
-                    worker.build_gas(vg)
-                    worker.stop(queue=True)
+                if self.can_afford(UnitTypeId.ASSIMILATOR):
+                    worker = self.select_build_worker(vg.position)
+                    if worker is not None:
+                        if not self.gas_buildings or not self.gas_buildings.closer_than(1, vg):
+                            worker.build_gas(vg)
+                            worker.stop(queue=True)
 
         # Research warp gate if cybercore is completed
         if (

examples/simulate_fight_scenario.py

@@ -42,19 +42,21 @@ async def on_step(self, iteration: int):
             return
 
     async def reset_arena(self):
+        if self.enemy_location is None:
+            return
         await self.client.debug_kill_unit(self.all_units)
 
         await self.client.debug_create_unit(
             [
-                [UnitTypeId.SUPPLYDEPOT, 1, self.enemy_location, OPPONENT_PLAYER_ID],
-                [UnitTypeId.MARINE, 4, self.enemy_location.towards(self.start_location, 8), OPPONENT_PLAYER_ID],
+                (UnitTypeId.SUPPLYDEPOT, 1, self.enemy_location, OPPONENT_PLAYER_ID),
+                (UnitTypeId.MARINE, 4, self.enemy_location.towards(self.start_location, 8), OPPONENT_PLAYER_ID),
             ]
         )
 
         await self.client.debug_create_unit(
             [
-                [UnitTypeId.SUPPLYDEPOT, 1, self.start_location, MY_PLAYER_ID],
-                [UnitTypeId.MARINE, 4, self.start_location.towards(self.enemy_location, 8), MY_PLAYER_ID],
+                (UnitTypeId.SUPPLYDEPOT, 1, self.start_location, MY_PLAYER_ID),
+                (UnitTypeId.MARINE, 4, self.start_location.towards(self.enemy_location, 8), MY_PLAYER_ID),
             ]
         )
 
@@ -63,6 +65,8 @@ async def manage_enemy_units(self):
             unit.attack(self.start_location)
 
     async def manage_own_units(self):
+        if self.enemy_location is None:
+            return
         for unit in self.units(UnitTypeId.MARINE):
             unit.attack(self.enemy_location)
             # TODO: implement your fight logic here

examples/terran/cyclone_push.py

@@ -78,31 +78,29 @@ async def on_step(self, iteration: int):
                     # Near same command as above with the depot
                     await self.build(UnitTypeId.BARRACKS, near=cc.position.towards(self.game_info.map_center, 8))
 
-            # If we have a barracks (complete or under construction) and less than 2 gas structures (here: refineries)
-            elif self.structures(UnitTypeId.BARRACKS) and self.gas_buildings.amount < 2:
-                if self.can_afford(UnitTypeId.REFINERY):
-                    # All the vespene geysirs nearby, including ones with a refinery on top of it
-                    vgs = self.vespene_geyser.closer_than(10, cc)
-                    for vg in vgs:
-                        if self.gas_buildings.filter(lambda unit: unit.distance_to(vg) < 1):
-                            continue
-                        # Select a worker closest to the vespene geysir
-                        worker: Unit | None = self.select_build_worker(vg)
-                        # Worker can be none in cases where all workers are dead
-                        # or 'select_build_worker' function only selects from workers which carry no minerals
-                        if worker is None:
-                            continue
+        # If we have a barracks (complete or under construction) and less than 2 gas structures (here: refineries)
+        if self.structures(UnitTypeId.BARRACKS) and self.gas_buildings.amount < 2:
+            if self.can_afford(UnitTypeId.REFINERY):
+                # All the vespene geysirs nearby, including ones with a refinery on top of it
+                vgs = self.vespene_geyser.closer_than(10, cc)
+                for vg in vgs:
+                    has_refinery = self.gas_buildings.filter(lambda unit: unit.distance_to(vg) < 1)
+                    if has_refinery:
+                        continue
+                    # Select a worker closest to the vespene geysir
+                    worker: Unit | None = self.select_build_worker(vg)
+                    # Worker can be none in cases where all workers are dead
+                    # or 'select_build_worker' function only selects from workers which carry no minerals
+                    if worker is not None:
                         # Issue the build command to the worker, important: vg has to be a Unit, not a position
                         worker.build_gas(vg)
-                        # Only issue one build geysir command per frame
-                        break
-
-            # If we have at least one barracks that is compelted, build factory
-            if self.structures(UnitTypeId.BARRACKS).ready:
-                if self.structures(UnitTypeId.FACTORY).amount < 3 and not self.already_pending(UnitTypeId.FACTORY):
-                    if self.can_afford(UnitTypeId.FACTORY):
-                        position: Point2 = cc.position.towards_with_random_angle(self.game_info.map_center, 16)
-                        await self.build(UnitTypeId.FACTORY, near=position)
+
+        # If we have at least one barracks that is completed, build factory
+        if self.structures(UnitTypeId.BARRACKS).ready:
+            if self.structures(UnitTypeId.FACTORY).amount < 3 and not self.already_pending(UnitTypeId.FACTORY):
+                if self.can_afford(UnitTypeId.FACTORY):
+                    position: Point2 = cc.position.towards_with_random_angle(self.game_info.map_center, 16)
+                    await self.build(UnitTypeId.FACTORY, near=position)
 
         for factory in self.structures(UnitTypeId.FACTORY).ready.idle:
             # Reactor allows us to build two at a time