Type definitions for game objects

Sub-modules

server.games.coop

server.games.custom_game

server.games.game

server.games.game_results

server.games.ladder_game

server.games.typedefs

Classes

class CoopGame (*args, **kwargs)

Class for coop game

Expand source code

class CoopGame(Game):
    """Class for coop game"""
    init_mode = InitMode.NORMAL_LOBBY
    game_type = GameType.COOP

    def __init__(self, *args, **kwargs):
        kwargs["game_mode"] = "coop"
        super().__init__(*args, **kwargs)

        self.validity = ValidityState.COOP_NOT_RANKED
        self.game_options.update({
            "Victory": Victory.SANDBOX,
            "TeamSpawn": "fixed",
            "RevealedCivilians": "No",
            "Difficulty": 3,
            "Expansion": "true"
        })
        self.leaderboard_lock = asyncio.Lock()
        self.leaderboard_saved = False

    async def validate_game_mode_settings(self):
        """
        Checks which only apply to the coop mode
        """

        valid_options = {
            "Victory": (Victory.SANDBOX, ValidityState.WRONG_VICTORY_CONDITION),
            "TeamSpawn": ("fixed", ValidityState.SPAWN_NOT_FIXED),
            "RevealedCivilians": (FA.DISABLED, ValidityState.CIVILIANS_REVEALED),
            "Difficulty": (3, ValidityState.WRONG_DIFFICULTY),
            "Expansion": (FA.ENABLED, ValidityState.EXPANSION_DISABLED),
        }
        await self._validate_game_options(valid_options)

    async def process_game_results(self):
        """
        When a coop game ends, we don't expect there to be any game results.
        """
        pass

Ancestors

• Game

Class variables

var game_type

var init_mode

Methods

async def process_game_results(self)

When a coop game ends, we don't expect there to be any game results.

async def validate_game_mode_settings(self)

Checks which only apply to the coop mode

Inherited members

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • get_connected_players
  • get_player_option
  • get_team_sets
  • is_even
  • is_visible_to_player
  • launch
  • load_results
  • persist_results
  • players
  • remove_game_connection
  • set_ai_option
  • set_name_unchecked
  • set_player_option
  • teams
  • update_game_stats
  • validate_game_settings

class CustomGame (id, *args, **kwargs)

Object that lasts for the lifetime of a game on FAF.

Expand source code

@with_logger
class CustomGame(Game):
    init_mode = InitMode.NORMAL_LOBBY
    game_type = GameType.CUSTOM

    def __init__(self, id, *args, **kwargs):
        new_kwargs = {
            "rating_type": RatingType.GLOBAL,
            "setup_timeout": 30
        }
        new_kwargs.update(kwargs)
        super().__init__(id, *args, **new_kwargs)

    async def _run_pre_rate_validity_checks(self):
        limit = len(self.players) * 60
        if not self.enforce_rating and time.time() - self.launched_at < limit:
            await self.mark_invalid(ValidityState.TOO_SHORT)

Ancestors

• Game

Class variables

var game_type

var init_mode

Inherited members

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • get_connected_players
  • get_player_option
  • get_team_sets
  • is_even
  • is_visible_to_player
  • launch
  • load_results
  • persist_results
  • players
  • remove_game_connection
  • set_ai_option
  • set_name_unchecked
  • set_player_option
  • teams
  • update_game_stats
  • validate_game_mode_settings
  • validate_game_settings

class FeaturedMod (id: int, name: str, full_name: str, description: str, publish: bool, order: int)

FeaturedMod(id, name, full_name, description, publish, order)

Expand source code

class FeaturedMod(NamedTuple):
    id: int
    name: str
    full_name: str
    description: str
    publish: bool
    order: int

Ancestors

• builtins.tuple

Instance variables

var description : str

Alias for field number 3

var full_name : str

Alias for field number 2

var id : int

Alias for field number 0

var name : str

Alias for field number 1

var order : int

Alias for field number 5

var publish : bool

Alias for field number 4

class FeaturedModType

String constants for featured mod technical_name

Expand source code

class FeaturedModType():
    """
    String constants for featured mod technical_name
    """

    COOP = "coop"
    FAF = "faf"
    FAFBETA = "fafbeta"
    LADDER_1V1 = "ladder1v1"

Class variables

var COOP

var FAF

var FAFBETA

var LADDER_1V1

class Game (id: int, database: FAFDatabase, game_service: GameService, game_stats_service: GameStatsService, host: Optional[Player] = None, name: str = 'New Game', map: Map = Map(id=None, folder_name='scmp_007', ranked=False, weight=1), game_mode: str = 'faf', matchmaker_queue_id: Optional[int] = None, rating_type: Optional[str] = None, displayed_rating_range: Optional[InclusiveRange] = None, enforce_rating_range: bool = False, max_players: int = 12, setup_timeout: int = 60)

Object that lasts for the lifetime of a game on FAF.

Expand source code

class Game:
    """
    Object that lasts for the lifetime of a game on FAF.
    """
    init_mode = InitMode.NORMAL_LOBBY
    game_type = GameType.CUSTOM

    def __init__(
        self,
        id: int,
        database: "FAFDatabase",
        game_service: "GameService",
        game_stats_service: "GameStatsService",
        host: Optional[Player] = None,
        name: str = "New Game",
        map: Map = MAP_DEFAULT,
        game_mode: str = FeaturedModType.FAF,
        matchmaker_queue_id: Optional[int] = None,
        rating_type: Optional[str] = None,
        displayed_rating_range: Optional[InclusiveRange] = None,
        enforce_rating_range: bool = False,
        max_players: int = 12,
        setup_timeout: int = 60,
    ):
        self.id = id
        self._db = database
        self._results = GameResultReports(id)
        self._army_stats_list = []
        self._players_with_unsent_army_stats = []
        self._game_stats_service = game_stats_service
        self.game_service = game_service
        self._player_options: dict[int, dict[str, Any]] = defaultdict(dict)
        self.hosted_at = None
        self.launched_at = None
        self.finished = False
        self._logger = logging.getLogger(
            f"{self.__class__.__qualname__}.{id}"
        )
        self.visibility = VisibilityState.PUBLIC
        self.host = host
        self.name = name
        self.map = map
        self.password = None
        self._players_at_launch: list[Player] = []
        self.AIs = {}
        self.desyncs = 0
        self.validity = ValidityState.VALID
        self.game_mode = game_mode
        self.rating_type = rating_type or RatingType.GLOBAL
        self.displayed_rating_range = displayed_rating_range or InclusiveRange()
        self.enforce_rating_range = enforce_rating_range
        self.matchmaker_queue_id = matchmaker_queue_id
        self.setup_timeout = setup_timeout
        self.state = GameState.INITIALIZING
        self._connections = {}
        self._configured_player_ids: set[int] = set()
        self.enforce_rating = False
        self.game_options = GameOptions(
            id,
            {
                "AIReplacement": "Off",
                "CheatsEnabled": "false",
                "FogOfWar": "explored",
                "GameSpeed": "normal",
                "NoRushOption": "Off",
                "PrebuiltUnits": "Off",
                "RestrictedCategories": 0,
                "ScenarioFile": (pathlib.PurePath(map.scenario_file)),
                "Slots": max_players,
                "TeamLock": "locked",
                "Unranked": "No",
                "Victory": Victory.DEMORALIZATION,
            }
        )
        self.game_options.add_async_callback(
            "ScenarioFile",
            self.on_scenario_file_changed,
        )
        self.game_options.add_callback("Title", self.on_title_changed)

        self.mods = {}
        self._hosted_future = asyncio.Future()
        self._finish_lock = asyncio.Lock()

        self._logger.debug("%s created", self)
        asyncio.get_event_loop().create_task(self.timeout_game(setup_timeout))

    async def timeout_game(self, timeout: int = 60):
        await asyncio.sleep(timeout)
        if self.state is GameState.INITIALIZING:
            self._logger.debug("Game setup timed out, cancelling game")
            await self.on_game_finish()

    async def on_scenario_file_changed(self, scenario_path: pathlib.PurePath):
        try:
            map_folder_name = scenario_path.parts[2].lower()
        except IndexError:
            return

        self.map = await self.game_service.get_map(map_folder_name)

    def on_title_changed(self, title: str):
        with contextlib.suppress(ValueError):
            self.name = title

    @property
    def name(self):
        return self._name

    @name.setter
    def name(self, value: str):
        """
        Verifies that names only contain ascii characters.
        """
        value = value.strip()

        if not value.isascii():
            raise ValueError("Game title must be ascii!")

        if not value:
            raise ValueError("Game title must not be empty!")

        self.set_name_unchecked(value)

    def set_name_unchecked(self, value: str):
        """
        Sets the game name without doing any validity checks.

        Truncates the game name to avoid crashing mysql INSERT statements.
        """
        max_len = game_stats.c.gameName.type.length
        self._name = value[:max_len]

    @property
    def max_players(self) -> int:
        return self.game_options["Slots"]

    @property
    def armies(self) -> frozenset[int]:
        return frozenset(
            self.get_player_option(player.id, "Army")
            for player in self.players
        )

    @property
    def players(self) -> list[Player]:
        """
        Players in the game

        Depending on the state, it is either:
          - (LOBBY) The currently connected players
          - (LIVE) Players who participated in the game
        """
        if self.state is GameState.LOBBY:
            return self.get_connected_players()
        else:
            return self._players_at_launch

    def get_connected_players(self) -> list[Player]:
        """
        Get a collection of all players currently connected to the game.
        """
        return [
            player for player in self._connections.keys()
            if player.id in self._configured_player_ids
        ]

    def _is_observer(self, player: Player) -> bool:
        army = self.get_player_option(player.id, "Army")
        return army is None or army < 0

    @property
    def connections(self) -> Iterable["GameConnection"]:
        return self._connections.values()

    @property
    def teams(self) -> frozenset[int]:
        """
        A set of all teams of this game's players.
        """
        return frozenset(
            self.get_player_option(player.id, "Team")
            for player in self.players
        )

    @property
    def is_ffa(self) -> bool:
        if len(self.players) < 3:
            return False

        return FFA_TEAM in self.teams

    @property
    def is_multi_team(self) -> bool:
        return len(self.teams) > 2

    @property
    def has_ai(self) -> bool:
        return len(self.AIs) > 0

    @property
    def is_even(self) -> bool:
        """
        If teams are balanced taking into account that players on the FFA team
        are on individual teams.

        # Returns
        `True` iff all teams have the same player count.

        Special cases:

        - `True` if there are zero teams.
        - `False` if there is a single team.
        """
        teams = self.get_team_sets()
        if len(teams) == 0:
            return True
        if len(teams) == 1:
            return False

        team_sizes = set(len(team) for team in teams)
        return len(team_sizes) == 1

    def get_team_sets(self) -> list[set[Player]]:
        """
        Returns a list of teams represented as sets of players.
        Note that FFA players will be separated into individual teams.
        """
        if None in self.teams:
            raise GameError(
                "Missing team for at least one player. (player, team): {}"
                .format([(player, self.get_player_option(player.id, "Team"))
                        for player in self.players])
            )

        teams = defaultdict(set)
        ffa_players = []
        for player in self.players:
            team_id = self.get_player_option(player.id, "Team")
            if team_id == FFA_TEAM:
                ffa_players.append({player})
            else:
                teams[team_id].add(player)

        return list(teams.values()) + ffa_players

    def set_hosted(self):
        self._hosted_future.set_result(None)
        self.hosted_at = datetime_now()

    async def add_result(
        self,
        reporter: int,
        army: int,
        result_type: str,
        score: int,
        result_metadata: frozenset[str] = frozenset(),
    ):
        """
        As computed by the game.

        # Params
        - `reporter`: player ID
        - `army`: the army number being reported for
        - `result_type`: a string representing the result
        - `score`: an arbitrary number assigned with the result
        - `result_metadata`: everything preceding the `result_type` in the
            result message from the game, one or more words, optional
        """
        if army not in self.armies:
            self._logger.debug(
                "Ignoring results for unknown army %s: %s %s reported by: %s",
                army, result_type, score, reporter
            )
            return

        try:
            outcome = ArmyReportedOutcome(result_type.upper())
        except ValueError:
            self._logger.debug(
                "Ignoring result reported by %s for army %s: %s %s",
                reporter, army, result_type, score
            )
            return

        result = GameResultReport(reporter, army, outcome, score, result_metadata)
        self._results.add(result)
        self._logger.info(
            "%s reported result for army %s: %s %s", reporter, army,
            result_type, score
        )

        self._process_pending_army_stats()

    def _process_pending_army_stats(self):
        for player in self._players_with_unsent_army_stats:
            army = self.get_player_option(player.id, "Army")
            if army not in self._results:
                continue

            for result in self._results[army]:
                if result.outcome is not GameOutcome.UNKNOWN:
                    self._process_army_stats_for_player(player)
                    break

    def _process_army_stats_for_player(self, player):
        try:
            if (
                len(self._army_stats_list) == 0
                or self.game_options["CheatsEnabled"] != "false"
            ):
                return

            self._players_with_unsent_army_stats.remove(player)
            asyncio.create_task(
                self._game_stats_service.process_game_stats(
                    player, self, self._army_stats_list
                )
            )
        except Exception:
            # Never let an error in processing army stats cascade
            self._logger.exception(
                "Army stats could not be processed from player %s in game %s",
                player, self
            )

    def add_game_connection(self, game_connection):
        """
        Add a game connection to this game.
        """
        if game_connection.state != GameConnectionState.CONNECTED_TO_HOST:
            raise GameError(
                f"Invalid GameConnectionState: {game_connection.state}"
            )
        if self.state is not GameState.LOBBY and self.state is not GameState.LIVE:
            raise GameError(f"Invalid GameState: {self.state}")

        self._logger.info("Added game connection %s", game_connection)
        self._connections[game_connection.player] = game_connection

    async def disconnect_player(self, player: Player):
        if player.game_connection not in self._connections.values():
            return

        self._configured_player_ids.discard(player.id)

        if self.state is GameState.LOBBY and player.id in self._player_options:
            del self._player_options[player.id]

        await self.remove_game_connection(player.game_connection)

    async def remove_game_connection(self, game_connection):
        """
        Remove a game connection from this game.

        Will trigger `on_game_finish` if there are no more active connections to the
        game.
        """
        if game_connection not in self._connections.values():
            return

        player = game_connection.player
        del self._connections[player]
        del player.game

        self._logger.info("Removed game connection %s", game_connection)

        await self.check_game_finish(player)

    async def check_game_finish(self, player):
        await self.check_sim_end()

        async with self._finish_lock:
            host_left_lobby = (
                player == self.host and self.state is not GameState.LIVE
            )

            if self.state is not GameState.ENDED and (
                self.finished or
                len(self._connections) == 0 or
                host_left_lobby
            ):
                await self.on_game_finish()
            else:
                self._process_pending_army_stats()

    async def check_sim_end(self):
        if self.finished:
            return
        if self.state is not GameState.LIVE:
            return
        if [conn for conn in self.connections if not conn.finished_sim]:
            return
        self.finished = True
        async with self._db.acquire() as conn:
            await conn.execute(
                game_stats.update().where(
                    game_stats.c.id == self.id
                ).values(
                    endTime=sql_now()
                )
            )

    async def on_game_finish(self):
        try:
            if self.state is GameState.LOBBY:
                self._logger.info("Game cancelled pre launch")
            elif self.state is GameState.INITIALIZING:
                self._logger.info("Game cancelled pre initialization")
            elif self.state is GameState.LIVE:
                self._logger.info("Game finished normally")

                if self.desyncs > 20:
                    await self.mark_invalid(ValidityState.TOO_MANY_DESYNCS)
                    return

                await self.process_game_results()

                self._process_pending_army_stats()
        except Exception:    # pragma: no cover
            self._logger.exception("Error during game end")
        finally:
            self.state = GameState.ENDED

            self.game_service.mark_dirty(self)

    async def _run_pre_rate_validity_checks(self):
        pass

    async def process_game_results(self):
        if not self._results:
            await self.mark_invalid(ValidityState.UNKNOWN_RESULT)
            return

        await self.persist_results()

        game_results = await self.resolve_game_results()
        await self.game_service.publish_game_results(game_results)

    async def resolve_game_results(self) -> EndedGameInfo:
        if self.state not in (GameState.LIVE, GameState.ENDED):
            raise GameError("Cannot rate game that has not been launched.")

        await self._run_pre_rate_validity_checks()

        basic_info = self.get_basic_info()

        team_army_results = [
            [self.get_army_results(player) for player in team]
            for team in basic_info.teams
        ]

        team_outcomes = [GameOutcome.UNKNOWN for _ in basic_info.teams]
        team_player_partial_outcomes = [
            {self.get_player_outcome(player) for player in team}
            for team in basic_info.teams
        ]

        try:
            # TODO: Remove override once game result messages are reliable
            team_outcomes = (
                self._outcome_override_hook()
                or resolve_game(team_player_partial_outcomes)
            )
        except GameResolutionError:
            if self.validity is ValidityState.VALID:
                await self.mark_invalid(ValidityState.UNKNOWN_RESULT)

        try:
            commander_kills = {
                army_stats["name"]: army_stats["units"]["cdr"]["kills"]
                for army_stats in self._army_stats_list
            }
        except KeyError:
            commander_kills = {}

        return EndedGameInfo.from_basic(
            basic_info,
            self.validity,
            team_outcomes,
            commander_kills,
            team_army_results,
        )

    def _outcome_override_hook(self) -> Optional[list[GameOutcome]]:
        return None

    async def load_results(self):
        """
        Load results from the database
        """
        self._results = await GameResultReports.from_db(self._db, self.id)

    async def persist_results(self):
        """
        Persist game results into the database

        Requires the game to have been launched and the appropriate rows to
        exist in the database.
        """

        self._logger.debug("Saving scores from game %s", self.id)
        scores = {}
        for player in self.players:
            army = self.get_player_option(player.id, "Army")
            outcome = self.get_player_outcome(player)
            score = self.get_army_score(army)
            scores[player] = (score, outcome)
            self._logger.info(
                "Result for army %s, player: %s: score %s, outcome %s",
                army, player, score, outcome
            )

        async with self._db.acquire() as conn:
            rows = []
            for player, (score, outcome) in scores.items():
                self._logger.info(
                    "Score for player %s: score %s, outcome %s",
                    player, score, outcome,
                )
                rows.append(
                    {
                        "score": score,
                        "result": outcome.name.upper(),
                        "game_id": self.id,
                        "player_id": player.id,
                    }
                )

            update_statement = game_player_stats.update().where(
                and_(
                    game_player_stats.c.gameId == bindparam("game_id"),
                    game_player_stats.c.playerId == bindparam("player_id"),
                )
            ).values(
                score=bindparam("score"),
                scoreTime=sql_now(),
                result=bindparam("result"),
            )
            await conn.deadlock_retry_execute(update_statement, rows)

    def get_basic_info(self) -> BasicGameInfo:
        return BasicGameInfo(
            self.id,
            self.rating_type,
            self.map.id,
            self.game_mode,
            list(self.mods.keys()),
            self.get_team_sets(),
        )

    def set_player_option(self, player_id: int, key: str, value: Any):
        """
        Set game-associative options for given player, by id
        """
        self._configured_player_ids.add(player_id)
        self._player_options[player_id][key] = value

    def get_player_option(self, player_id: int, key: str) -> Optional[Any]:
        """
        Retrieve game-associative options for given player, by their uid
        """
        return self._player_options[player_id].get(key)

    def set_ai_option(self, name, key, value):
        """
        Set game-associative options for given AI, by name
        """
        if name not in self.AIs:
            self.AIs[name] = {}
        self.AIs[name][key] = value

    def clear_slot(self, slot_index):
        """
        A somewhat awkward message while we're still half-slot-associated with
        a bunch of data.

        Just makes sure that any players associated with this slot aren't
        assigned an army or team, and deletes any AI's.
        """
        for player in self.players:
            if self.get_player_option(player.id, "StartSpot") == slot_index:
                self.set_player_option(player.id, "Team", -1)
                self.set_player_option(player.id, "Army", -1)
                self.set_player_option(player.id, "StartSpot", -1)

        to_remove = []
        for ai in self.AIs:
            if self.AIs[ai]["StartSpot"] == slot_index:
                to_remove.append(ai)
        for item in to_remove:
            del self.AIs[item]

    async def validate_game_settings(self):
        """
        Mark the game invalid if it has non-compliant options
        """

        # Only allow ranked mods
        for mod_id in self.mods.keys():
            if mod_id not in self.game_service.ranked_mods:
                await self.mark_invalid(ValidityState.BAD_MOD)
                return

        if self.has_ai:
            await self.mark_invalid(ValidityState.HAS_AI_PLAYERS)
            return
        if self.is_multi_team:
            await self.mark_invalid(ValidityState.MULTI_TEAM)
            return
        if self.is_ffa:
            await self.mark_invalid(ValidityState.FFA_NOT_RANKED)
            return
        valid_options = {
            "AIReplacement": (FA.DISABLED, ValidityState.HAS_AI_PLAYERS),
            "FogOfWar": ("explored", ValidityState.NO_FOG_OF_WAR),
            "CheatsEnabled": (FA.DISABLED, ValidityState.CHEATS_ENABLED),
            "PrebuiltUnits": (FA.DISABLED, ValidityState.PREBUILT_ENABLED),
            "NoRushOption": (FA.DISABLED, ValidityState.NORUSH_ENABLED),
            "RestrictedCategories": (0, ValidityState.BAD_UNIT_RESTRICTIONS),
            "TeamLock": ("locked", ValidityState.UNLOCKED_TEAMS),
            "Unranked": (FA.DISABLED, ValidityState.HOST_SET_UNRANKED)
        }
        if await self._validate_game_options(valid_options) is False:
            return

        await self.validate_game_mode_settings()

    async def validate_game_mode_settings(self):
        """
        A subset of checks that need to be overridden in coop games.
        """
        if len(self.players) < 2:
            await self.mark_invalid(ValidityState.SINGLE_PLAYER)
            return

        if None in self.teams or not self.is_even:
            await self.mark_invalid(ValidityState.UNEVEN_TEAMS_NOT_RANKED)
            return

        valid_options = {
            "Victory": (Victory.DEMORALIZATION, ValidityState.WRONG_VICTORY_CONDITION)
        }
        await self._validate_game_options(valid_options)

    async def _validate_game_options(
        self,
        valid_options: dict[str, tuple[Any, ValidityState]]
    ) -> bool:
        for key, value in self.game_options.items():
            if key in valid_options:
                valid_value, validity_state = valid_options[key]
                if value != valid_value:
                    await self.mark_invalid(validity_state)
                    return False
        return True

    async def launch(self):
        """
        Mark the game as live.

        Freezes the set of active players so they are remembered if they drop.
        """
        assert self.state is GameState.LOBBY
        self.launched_at = time.time()
        # Freeze currently connected players since we need them for rating when
        # the game ends.
        self._players_at_launch = [
            player for player in self.get_connected_players()
            if not self._is_observer(player)
        ]
        self._players_with_unsent_army_stats = list(self._players_at_launch)

        self.state = GameState.LIVE

        await self.on_game_launched()
        await self.validate_game_settings()

        self._logger.info("Game launched")

    async def on_game_launched(self):
        for player in self.players:
            player.state = PlayerState.PLAYING
        await self.update_game_stats()
        await self.update_game_player_stats()

    async def update_game_stats(self):
        """
        Runs at game-start to populate the game_stats table (games that start are ones we actually
        care about recording stats for, after all).
        """
        assert self.host is not None

        # Ensure map data is up to date
        self.map = await self.game_service.get_map(self.map.folder_name)

        if self.validity is ValidityState.VALID and not self.map.ranked:
            await self.mark_invalid(ValidityState.BAD_MAP)

        modId = self.game_service.featured_mods[self.game_mode].id

        # Write out the game_stats record.
        # In some cases, games can be invalidated while running: we check for
        # those cases when the game ends and update this record as appropriate.

        game_type = str(self.game_options.get("Victory").value)

        async with self._db.acquire() as conn:
            await conn.execute(
                game_stats.insert().values(
                    id=self.id,
                    gameType=game_type,
                    gameMod=modId,
                    host=self.host.id,
                    mapId=self.map.id,
                    gameName=self.name,
                    validity=self.validity.value,
                )
            )

            if self.matchmaker_queue_id is not None:
                await conn.execute(
                    matchmaker_queue_game.insert().values(
                        matchmaker_queue_id=self.matchmaker_queue_id,
                        game_stats_id=self.id,
                    )
                )

    async def update_game_player_stats(self):
        query_args = []
        for player in self.players:
            options = {
                key: self.get_player_option(player.id, key)
                for key in ["Team", "StartSpot", "Color", "Faction"]
            }

            is_observer = (
                options["Team"] is None
                or options["Team"] < 0
                or options["StartSpot"] is None
                or options["StartSpot"] < 0
            )
            if is_observer:
                continue

            # DEPRECATED: Rating changes are persisted by the rating service
            # in the `leaderboard_rating_journal` table.
            mean, deviation = player.ratings[self.rating_type]

            query_args.append(
                {
                    "gameId": self.id,
                    "playerId": player.id,
                    "faction": options["Faction"],
                    "color": options["Color"],
                    "team": options["Team"],
                    "place": options["StartSpot"],
                    "mean": mean,
                    "deviation": deviation,
                    "AI": 0,
                    "score": 0,
                }
            )
        if not query_args:
            self._logger.warning("No player options available!")
            return

        try:
            async with self._db.acquire() as conn:
                await conn.execute(game_player_stats.insert().values(query_args))
        except DBAPIError:
            self._logger.exception(
                "Failed to update game_player_stats. Query args %s:", query_args
            )
            raise

    async def mark_invalid(self, new_validity_state: ValidityState):
        self._logger.info(
            "Marked as invalid because: %s", repr(new_validity_state)
        )
        self.validity = new_validity_state

        # If we haven't started yet, the invalidity will be persisted to the database when we start.
        # Otherwise, we have to do a special update query to write this information out.
        if self.state is not GameState.LIVE:
            return

        # Currently, we can only end up here if a game desynced or was a custom game that terminated
        # too quickly.
        async with self._db.acquire() as conn:
            await conn.execute(
                game_stats.update().where(
                    game_stats.c.id == self.id
                ).values(
                    validity=new_validity_state.value
                )
            )

    def get_army_score(self, army):
        return self._results.score(army)

    def get_player_outcome(self, player: Player) -> ArmyOutcome:
        army = self.get_player_option(player.id, "Army")
        if army is None:
            return ArmyOutcome.UNKNOWN

        return self._results.outcome(army)

    def get_army_results(self, player: Player) -> ArmyResult:
        army = self.get_player_option(player.id, "Army")
        return ArmyResult(
            player.id,
            army,
            self.get_player_outcome(player).name,
            self._results.metadata(army),
        )

    def report_army_stats(self, stats_json):
        self._army_stats_list = json.loads(stats_json)["stats"]
        self._process_pending_army_stats()

    def is_visible_to_player(self, player: Player) -> bool:
        """
        Determine if a player should see this game in their games list.

        Note: This is a *hot* function, it can have significant impacts on
        performance.
        """
        if self.host is None:
            return False

        if player == self.host or player in self._connections:
            return True

        if (
            self.enforce_rating_range
            and player.ratings[self.rating_type].displayed()
            not in self.displayed_rating_range
        ):
            return False

        if self.visibility is VisibilityState.FRIENDS:
            return player.id in self.host.friends
        else:
            return player.id not in self.host.foes

    def to_dict(self):
        client_state = {
            GameState.LOBBY: "open",
            GameState.LIVE: "playing",
            GameState.ENDED: "closed",
            GameState.INITIALIZING: "closed",
        }.get(self.state, "closed")
        connected_players = self.get_connected_players()
        return {
            "command": "game_info",
            "visibility": self.visibility.value,
            "password_protected": self.password is not None,
            "uid": self.id,
            "title": self.name,
            "state": client_state,
            "game_type": self.game_type.value,
            "featured_mod": self.game_mode,
            "sim_mods": self.mods,
            "mapname": self.map.folder_name,
            # DEPRECATED: Use `mapname` instead
            "map_file_path": self.map.file_path,
            "host": self.host.login if self.host else "",
            "num_players": len(connected_players),
            "max_players": self.max_players,
            "hosted_at": self.hosted_at.isoformat() if self.hosted_at else None,
            "launched_at": self.launched_at,
            "rating_type": self.rating_type,
            "rating_min": self.displayed_rating_range.lo,
            "rating_max": self.displayed_rating_range.hi,
            "enforce_rating_range": self.enforce_rating_range,
            "teams_ids": [
                {
                    "team_id": team,
                    "player_ids": [
                        player.id for player in connected_players
                        if self.get_player_option(player.id, "Team") == team
                    ]
                }
                for team in self.teams if team is not None
            ],
            "teams": {
                team: [
                    player.login for player in connected_players
                    if self.get_player_option(player.id, "Team") == team
                ]
                for team in self.teams if team is not None
            }
        }

    def __eq__(self, other):
        if not isinstance(other, Game):
            return False
        else:
            return self.id == other.id

    def __hash__(self):
        return self.id.__hash__()

    def __str__(self) -> str:
        return (
            f"Game({self.id}, {self.host.login if self.host else ''}, "
            f"{self.map.file_path})"
        )

Subclasses

• CoopGame
• CustomGame
• LadderGame

Class variables

var game_type

var init_mode

Instance variables

prop armies : frozenset[int]

Expand source code

@property
def armies(self) -> frozenset[int]:
    return frozenset(
        self.get_player_option(player.id, "Army")
        for player in self.players
    )

prop connections : Iterable[GameConnection]

Expand source code

@property
def connections(self) -> Iterable["GameConnection"]:
    return self._connections.values()

prop has_ai : bool

Expand source code

@property
def has_ai(self) -> bool:
    return len(self.AIs) > 0

prop is_even : bool

If teams are balanced taking into account that players on the FFA team are on individual teams.

Returns

True iff all teams have the same player count.

Special cases:

• True if there are zero teams.
• False if there is a single team.

Expand source code

@property
def is_even(self) -> bool:
    """
    If teams are balanced taking into account that players on the FFA team
    are on individual teams.

    # Returns
    `True` iff all teams have the same player count.

    Special cases:

    - `True` if there are zero teams.
    - `False` if there is a single team.
    """
    teams = self.get_team_sets()
    if len(teams) == 0:
        return True
    if len(teams) == 1:
        return False

    team_sizes = set(len(team) for team in teams)
    return len(team_sizes) == 1

prop is_ffa : bool

Expand source code

@property
def is_ffa(self) -> bool:
    if len(self.players) < 3:
        return False

    return FFA_TEAM in self.teams

prop is_multi_team : bool

Expand source code

@property
def is_multi_team(self) -> bool:
    return len(self.teams) > 2

prop max_players : int

Expand source code

@property
def max_players(self) -> int:
    return self.game_options["Slots"]

prop name

Expand source code

@property
def name(self):
    return self._name

prop players : list[Player]

Players in the game

Depending on the state, it is either: - (LOBBY) The currently connected players - (LIVE) Players who participated in the game

Expand source code

@property
def players(self) -> list[Player]:
    """
    Players in the game

    Depending on the state, it is either:
      - (LOBBY) The currently connected players
      - (LIVE) Players who participated in the game
    """
    if self.state is GameState.LOBBY:
        return self.get_connected_players()
    else:
        return self._players_at_launch

prop teams : frozenset[int]

A set of all teams of this game's players.

Expand source code

@property
def teams(self) -> frozenset[int]:
    """
    A set of all teams of this game's players.
    """
    return frozenset(
        self.get_player_option(player.id, "Team")
        for player in self.players
    )

Methods

def add_game_connection(self, game_connection)

Add a game connection to this game.

async def add_result(self, reporter: int, army: int, result_type: str, score: int, result_metadata: frozenset[str] = frozenset())

As computed by the game.

Params

• reporter: player ID
• army: the army number being reported for
• result_type: a string representing the result
• score: an arbitrary number assigned with the result
• result_metadata: everything preceding the result_type in the result message from the game, one or more words, optional

async def check_game_finish(self, player)

async def check_sim_end(self)

def clear_slot(self, slot_index)

A somewhat awkward message while we're still half-slot-associated with a bunch of data.

Just makes sure that any players associated with this slot aren't assigned an army or team, and deletes any AI's.

async def disconnect_player(self, player: Player)

def get_army_results(self, player: Player) ‑> ArmyResult

def get_army_score(self, army)

def get_basic_info(self) ‑> BasicGameInfo

def get_connected_players(self) ‑> list[Player]

Get a collection of all players currently connected to the game.

def get_player_option(self, player_id: int, key: str) ‑> Optional[Any]

Retrieve game-associative options for given player, by their uid

def get_player_outcome(self, player: Player) ‑> ArmyOutcome

def get_team_sets(self) ‑> list[set[Player]]

Returns a list of teams represented as sets of players. Note that FFA players will be separated into individual teams.

def is_visible_to_player(self, player: Player) ‑> bool

Determine if a player should see this game in their games list.

Note: This is a hot function, it can have significant impacts on performance.

async def launch(self)

Mark the game as live.

Freezes the set of active players so they are remembered if they drop.

async def load_results(self)

Load results from the database

async def mark_invalid(self, new_validity_state: ValidityState)

async def on_game_finish(self)

async def on_game_launched(self)

async def on_scenario_file_changed(self, scenario_path: pathlib.PurePath)

def on_title_changed(self, title: str)

async def persist_results(self)

Persist game results into the database

Requires the game to have been launched and the appropriate rows to exist in the database.

async def process_game_results(self)

async def remove_game_connection(self, game_connection)

Remove a game connection from this game.

Will trigger on_game_finish if there are no more active connections to the game.

def report_army_stats(self, stats_json)

async def resolve_game_results(self) ‑> EndedGameInfo

def set_ai_option(self, name, key, value)

Set game-associative options for given AI, by name

def set_hosted(self)

def set_name_unchecked(self, value: str)

Sets the game name without doing any validity checks.

Truncates the game name to avoid crashing mysql INSERT statements.

def set_player_option(self, player_id: int, key: str, value: Any)

Set game-associative options for given player, by id

async def timeout_game(self, timeout: int = 60)

def to_dict(self)

async def update_game_player_stats(self)

async def update_game_stats(self)

Runs at game-start to populate the game_stats table (games that start are ones we actually care about recording stats for, after all).

async def validate_game_mode_settings(self)

A subset of checks that need to be overridden in coop games.

async def validate_game_settings(self)

Mark the game invalid if it has non-compliant options

class GameConnectionState (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class GameConnectionState(Enum):
    INITIALIZING = 0
    INITIALIZED = 1
    CONNECTED_TO_HOST = 2
    ENDED = 3

Ancestors

• enum.Enum

Class variables

var CONNECTED_TO_HOST

var ENDED

var INITIALIZED

var INITIALIZING

class GameError (*args, **kwargs)

Common base class for all non-exit exceptions.

Expand source code

class GameError(Exception):
    pass

Ancestors

• builtins.Exception
• builtins.BaseException

class GameOptions (id: int, *args, **kwargs)

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Expand source code

class GameOptions(dict):
    def __init__(self, id: int, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._logger = logging.getLogger(
            f"{self.__class__.__qualname__}.{id}"
        )
        self.callbacks = defaultdict(list)
        self.async_callbacks = defaultdict(list)

    def add_callback(self, key: str, callback: Callable[[Any], Any]):
        self.callbacks[key].append(callback)

    def add_async_callback(
        self,
        key: str,
        callback: Callable[[Any], Awaitable[Any]],
    ):
        self.async_callbacks[key].append(callback)

    async def set_option(self, k: str, v: Any) -> None:
        v = self._set_option(k, v)
        self._run_sync_callbacks(k, v)

        await asyncio.gather(*(
            self._log_async_exception(
                async_callback(v),
                k,
                v,
            )
            for async_callback in self.async_callbacks.get(k, ())
        ))

    def __setitem__(self, k: str, v: Any) -> None:
        v = self._set_option(k, v)
        self._run_sync_callbacks(k, v)

        for async_callback in self.async_callbacks.get(k, ()):
            asyncio.create_task(
                self._log_async_exception(
                    async_callback(v),
                    k,
                    v,
                )
            )

    def _set_option(self, k: str, v: Any) -> Any:
        """
        Set the new value potentially transforming it first. Returns the value
        that was set.
        """
        if k == "Victory" and not isinstance(v, Victory):
            victory = Victory.__members__.get(v.upper())
            if victory is None:
                victory = self.get("Victory")
                self._logger.warning(
                    "Invalid victory type '%s'! Using '%s' instead.",
                    v,
                    victory.name if victory else None,
                )
                return
            v = victory
        elif k == "Slots":
            v = int(v)
        elif k == "ScenarioFile":
            # Convert to a posix path. Since posix paths are also interpreted
            # the same way as windows paths (but not the other way around!) we
            # can do this by parsing as a PureWindowsPath first
            v = pathlib.PurePath(pathlib.PureWindowsPath(v).as_posix())

        super().__setitem__(k, v)

        return v

    def _run_sync_callbacks(self, k: str, v: Any):
        for callback in self.callbacks.get(k, ()):
            try:
                callback(v)
            except Exception:
                self._logger.exception(
                    "Error running callback for '%s' (value %r)",
                    k,
                    v,
                )

    async def _log_async_exception(self, coro: Awaitable[Any], k: str, v: Any):
        try:
            return await coro
        except Exception:
            self._logger.exception(
                "Error running async callback for '%s' (value %r)",
                k,
                v,
            )

Ancestors

• builtins.dict

Methods

def add_async_callback(self, key: str, callback: Callable[[Any], Awaitable[Any]])

def add_callback(self, key: str, callback: Callable[[Any], Any])

async def set_option(self, k: str, v: Any) ‑> None

class GameState (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class GameState(Enum):
    INITIALIZING = 0
    LOBBY = 1
    LIVE = 2
    ENDED = 3

Ancestors

• enum.Enum

Class variables

var ENDED

var INITIALIZING

var LIVE

var LOBBY

class GameType (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class GameType(Enum):
    COOP = "coop"
    CUSTOM = "custom"
    MATCHMAKER = "matchmaker"

Ancestors

• enum.Enum

Class variables

var COOP

var CUSTOM

var MATCHMAKER

class InitMode (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class InitMode(Enum):
    NORMAL_LOBBY = 0
    AUTO_LOBBY = 1

Ancestors

• enum.Enum

Class variables

var AUTO_LOBBY

var NORMAL_LOBBY

class LadderGame (id, *args, **kwargs)

Class for 1v1 ladder games

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]

Ancestors

• Game

Class variables

var game_type

var init_mode

Methods

async def check_game_finish(self, player)

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.

def is_winner(self, player: Player) ‑> bool

async def wait_hosted(self, timeout: float)

async def wait_launched(self, timeout: float)

Inherited members

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • get_connected_players
  • get_player_option
  • get_team_sets
  • is_even
  • is_visible_to_player
  • launch
  • load_results
  • persist_results
  • players
  • remove_game_connection
  • set_ai_option
  • set_name_unchecked
  • set_player_option
  • teams
  • update_game_stats
  • validate_game_mode_settings
  • validate_game_settings

class ValidityState (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class ValidityState(Enum):
    VALID = 0
    TOO_MANY_DESYNCS = 1
    WRONG_VICTORY_CONDITION = 2
    NO_FOG_OF_WAR = 3
    CHEATS_ENABLED = 4
    PREBUILT_ENABLED = 5
    NORUSH_ENABLED = 6
    BAD_UNIT_RESTRICTIONS = 7
    BAD_MAP = 8
    TOO_SHORT = 9
    BAD_MOD = 10
    COOP_NOT_RANKED = 11
    MUTUAL_DRAW = 12
    SINGLE_PLAYER = 13
    FFA_NOT_RANKED = 14
    UNEVEN_TEAMS_NOT_RANKED = 15
    UNKNOWN_RESULT = 16
    UNLOCKED_TEAMS = 17
    MULTI_TEAM = 18
    HAS_AI_PLAYERS = 19
    CIVILIANS_REVEALED = 20
    WRONG_DIFFICULTY = 21
    EXPANSION_DISABLED = 22
    SPAWN_NOT_FIXED = 23
    OTHER_UNRANK = 24
    HOST_SET_UNRANKED = 25

Ancestors

• enum.Enum

Class variables

var BAD_MAP

var BAD_MOD

var BAD_UNIT_RESTRICTIONS

var CHEATS_ENABLED

var CIVILIANS_REVEALED

var COOP_NOT_RANKED

var EXPANSION_DISABLED

var FFA_NOT_RANKED

var HAS_AI_PLAYERS

var HOST_SET_UNRANKED

var MULTI_TEAM

var MUTUAL_DRAW

var NORUSH_ENABLED

var NO_FOG_OF_WAR

var OTHER_UNRANK

var PREBUILT_ENABLED

var SINGLE_PLAYER

var SPAWN_NOT_FIXED

var TOO_MANY_DESYNCS

var TOO_SHORT

var UNEVEN_TEAMS_NOT_RANKED

var UNKNOWN_RESULT

var UNLOCKED_TEAMS

var VALID

var WRONG_DIFFICULTY

var WRONG_VICTORY_CONDITION

class Victory (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class Victory(Enum):
    DEMORALIZATION = 0
    DOMINATION = 1
    ERADICATION = 2
    SANDBOX = 3

Ancestors

• enum.Enum

Class variables

var DEMORALIZATION

var DOMINATION

var ERADICATION

var SANDBOX

class VisibilityState (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

@unique
class VisibilityState(Enum):
    PUBLIC = "public"
    FRIENDS = "friends"

Ancestors

• enum.Enum

Class variables

var FRIENDS

var PUBLIC