test_comm_status.py

import asyncio
import random
from diplomacy.client.connection import connect
from diplomacy.utils import exceptions
from diplomacy_research.players import RuleBasedPlayer
from diplomacy_research.players.rulesets import dumbbot_ruleset
from random import *
from diplomacy.utils import strings

POWERS = ['AUSTRIA', 'ENGLAND', 'FRANCE', 'GERMANY', 'ITALY', 'RUSSIA', 'TURKEY']


async def create_game(game_id, hostname='localhost', port=8432):
    """ Creates a game on the server """
    connection = await connect(hostname, port)
    channel = await connection.authenticate('random_user', 'password')
    await channel.create_game(game_id=game_id, rules={'REAL_TIME', 'NO_DEADLINE', 'POWER_CHOICE'})

async def play(game_id, power_name, hostname='localhost', port=8432):
    """ Play as the specified power """
    connection = await connect(hostname, port)
    channel = await connection.authenticate("bot_"+power_name,'password')
    bot = RuleBasedPlayer(dumbbot_ruleset)

    # Waiting for the game, then joining it
    while not (await channel.list_games(game_id=game_id)):
        await asyncio.sleep(1.)
    game = await channel.join_game(game_id=game_id, power_name=power_name)




    while not game.is_game_done:
        current_phase = game.get_current_phase()

        if power_name == "AUSTRIA":
            await game.set_comm_status(comm_status=strings.BUSY)
            await asyncio.sleep(2)

        while current_phase == game.get_current_phase():
            await asyncio.sleep(0.1)

async def launch(game_id):
    """ Creates and plays a network game """
    game_id = "test21"
    print("GAME: " + game_id)
    #await create_game(game_id)
    await play(game_id, power_name="AUSTRIA")
    #await asyncio.gather(*[play(game_id, power_name) for power_name in POWERS])

if __name__ == '__main__':
    asyncio.run(launch(game_id=str(randint(1, 1000))))