-
Notifications
You must be signed in to change notification settings - Fork 0
/
__init__.py
90 lines (72 loc) · 3.2 KB
/
__init__.py
1
2
3
4
5
6
7
8
9
10
11
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
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
import argparse
import sys
import asyncio
import logging
import aiohttp
import sc2
from sc2.bot_ai import BotAI
from sc2.data import Race
from sc2.player import Bot, Computer
from sc2.sc2process import SC2Process
from sc2.client import Client
from sc2.protocol import ConnectionAlreadyClosed
# Run ladder game
# This lets python-sc2 connect to a LadderManager game: https://github.com/Cryptyc/Sc2LadderServer
# Based on: https://github.com/Dentosal/python-sc2/blob/master/examples/run_external.py
def run_ladder_game(bot):
# Load command line arguments
parser = argparse.ArgumentParser()
parser.add_argument("--GamePort", type=int, nargs="?", help="Game port")
parser.add_argument("--StartPort", type=int, nargs="?", help="Start port")
parser.add_argument("--LadderServer", type=str, nargs="?", help="Ladder server")
parser.add_argument("--ComputerOpponent", type=str, nargs="?", help="Computer opponent")
parser.add_argument("--ComputerRace", type=str, nargs="?", help="Computer race")
parser.add_argument("--ComputerDifficulty", type=str, nargs="?", help="Computer difficulty")
parser.add_argument("--OpponentId", type=str, nargs="?", help="Opponent ID")
args, unknown = parser.parse_known_args()
if args.LadderServer:
host = args.LadderServer
else:
host = "127.0.0.1"
host_port = args.GamePort
lan_port = args.StartPort
# Add opponent_id to the bot class (accessed through self.opponent_id)
bot.ai.opponent_id = args.OpponentId
# Versus Computer doesn't work yet
computer_opponent = False
if args.ComputerOpponent:
computer_opponent = True
computer_race = args.ComputerRace
computer_difficulty = args.ComputerDifficulty
# Port config
ports = [lan_port + p for p in range(1, 6)]
portconfig = sc2.portconfig.Portconfig()
portconfig.shared = ports[0] # Not used
portconfig.server = [ports[1], ports[2]]
portconfig.players = [[ports[3], ports[4]]]
# Join ladder game
g = join_ladder_game(host=host, port=host_port, players=[bot], realtime=False, portconfig=portconfig)
# Run it
result = asyncio.get_event_loop().run_until_complete(g)
return result, args.OpponentId
# 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, step_time_limit=None, game_time_limit=None
):
ws_url = f"ws://{host}:{port}/sc2api"
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, step_time_limit, game_time_limit)
if save_replay_as is not None:
await client.save_replay(save_replay_as)
except ConnectionAlreadyClosed:
logging.error(f"Connection was closed before the game ended")
return None
finally:
# await ws_connection.close() ?
# __init__.py:81: RuntimeWarning: coroutine 'ClientWebSocketResponse.close' was never awaited
# ws_connection.close()
ws_connection.close()
return result