Module server.games.ladder_game

Classes

class GameClosedError (player: Player)
Expand source code 
class GameClosedError(Exception):
    """
    The game has been closed during the setup phase
    """

    def __init__(self, player: Player):
        self.player = player

The game has been closed during the setup phase

Ancestors

  • builtins.Exception
  • builtins.BaseException
class LadderGame (id, *args, **kwargs)
Expand source code 
class LadderGame(Game):
    """Class for 1v1 ladder games"""

    init_mode = InitMode.AUTO_LOBBY
    game_type = GameType.MATCHMAKER

    def __init__(self, id, *args, **kwargs):
        super().__init__(id, *args, **kwargs)
        self._launch_future = asyncio.Future()

    async def wait_hosted(self, timeout: float):
        return await asyncio.wait_for(
            self._hosted_future,
            timeout=timeout
        )

    async def wait_launched(self, timeout: float):
        return await asyncio.wait_for(
            self._launch_future,
            timeout=timeout
        )

    async def launch(self):
        await super().launch()
        self._launch_future.set_result(None)

    async def check_game_finish(self, player):
        if not self._hosted_future.done() and (
            self.state in (GameState.INITIALIZING, GameState.LOBBY)
        ):
            assert self.host == player
            self._hosted_future.set_exception(GameClosedError(player))

        if not self._launch_future.done() and (
            self.state in (GameState.INITIALIZING, GameState.LOBBY)
        ):
            self._launch_future.set_exception(GameClosedError(player))

        await super().check_game_finish(player)

    def is_winner(self, player: Player) -> bool:
        return self.get_player_outcome(player) is ArmyOutcome.VICTORY

    def get_army_score(self, army: int) -> int:
        """
        We override this function so that ladder game scores are only reported
        as 1 for win and 0 for anything else.
        """
        return self._results.victory_only_score(army)

    def _outcome_override_hook(self) -> Optional[list[GameOutcome]]:
        if not config.LADDER_1V1_OUTCOME_OVERRIDE or len(self.players) > 2:
            return None
        team_sets = self.get_team_sets()
        army_scores = [
            self._results.score(self.get_player_option(team_set.pop().id, "Army"))
            for team_set in team_sets
        ]
        if army_scores[0] > army_scores[1]:
            return [GameOutcome.VICTORY, GameOutcome.DEFEAT]
        elif army_scores[0] < army_scores[1]:
            return [GameOutcome.DEFEAT, GameOutcome.VICTORY]
        else:
            return [GameOutcome.DRAW, GameOutcome.DRAW]

Class for 1v1 ladder games

Ancestors

Class variables

var game_type
var init_mode

Methods

async def check_game_finish(self, player)
Expand source code 
async def check_game_finish(self, player):
    if not self._hosted_future.done() and (
        self.state in (GameState.INITIALIZING, GameState.LOBBY)
    ):
        assert self.host == player
        self._hosted_future.set_exception(GameClosedError(player))

    if not self._launch_future.done() and (
        self.state in (GameState.INITIALIZING, GameState.LOBBY)
    ):
        self._launch_future.set_exception(GameClosedError(player))

    await super().check_game_finish(player)
def get_army_score(self, army: int) ‑> int
Expand source code 
def get_army_score(self, army: int) -> int:
    """
    We override this function so that ladder game scores are only reported
    as 1 for win and 0 for anything else.
    """
    return self._results.victory_only_score(army)

We override this function so that ladder game scores are only reported as 1 for win and 0 for anything else.

def is_winner(self,
player: Player) ‑> bool
Expand source code 
def is_winner(self, player: Player) -> bool:
    return self.get_player_outcome(player) is ArmyOutcome.VICTORY
async def wait_hosted(self, timeout: float)
Expand source code 
async def wait_hosted(self, timeout: float):
    return await asyncio.wait_for(
        self._hosted_future,
        timeout=timeout
    )
async def wait_launched(self, timeout: float)
Expand source code 
async def wait_launched(self, timeout: float):
    return await asyncio.wait_for(
        self._launch_future,
        timeout=timeout
    )

Inherited members