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)

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

Class for coop game

Ancestors

• Game

Methods

async def process_game_results(self)

Expand source code

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

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

async def validate_game_mode_settings(self)

Expand source code

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)

Checks which only apply to the coop mode

Inherited members

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • game_type
  • get_connected_players
  • get_player_option
  • get_team_sets
  • init_mode
  • 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)

Expand source code

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):
        assert self.launched_at is not None

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

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

Ancestors

• Game

Inherited members

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • game_type
  • get_connected_players
  • get_player_option
  • get_team_sets
  • init_mode
  • 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)

Expand source code

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

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

Ancestors

• builtins.tuple

Instance variables

var description : str

Expand source code

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

Alias for field number 3

var full_name : str

Expand source code

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

Alias for field number 2

var id : int

Expand source code

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

Alias for field number 0

var name : str

Expand source code

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

Alias for field number 1

var order : int

Expand source code

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

Alias for field number 5

var publish : bool

Expand source code

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

Alias for field number 4

class FeaturedModType

Expand source code

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

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

String constants for featured mod technical_name

Class variables

var COOP

The type of the None singleton.

var FAF

The type of the None singleton.

var FAFBETA

The type of the None singleton.

var LADDER_1V1

The type of the None singleton.

class Game (id: int,
database: FAFDatabase,
game_service: GameService,
game_stats_service: GameStatsService,
host: Player | None = 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: int | None = None,
rating_type: str | None = None,
displayed_rating_range: InclusiveRange | None = None,
enforce_rating_range: bool = False,
max_players: int = 12,
setup_timeout: int = 60)

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: list[Any] = []
        self._players_with_unsent_army_stats: list[Player] = []
        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: Optional[datetime] = None
        self.launched_at: Optional[float] = 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: Optional[str] = None
        self._players_at_launch: list[Player] = []
        self.AIs: dict[str, dict[str, Any]] = {}
        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: dict[Player, "GameConnection"] = {}
        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: dict[str, str] = {}
        self._finish_lock = asyncio.Lock()

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

    @cached_property
    def _hosted_future(self) -> asyncio.Future:
        return asyncio.get_running_loop().create_future()

    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.
        """
        # mypy sees the column type as a generic base class so it doesn't
        # know about the `length` field
        max_len = game_stats.c.gameName.type.length  # type: ignore
        self._name = value[:max_len]

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

    @property
    def armies(self) -> frozenset[Optional[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[Optional[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) -> None:
        for player in self._players_with_unsent_army_stats:
            army = self.get_player_option(player.id, "Army")
            if army not in self._results:
                continue

            self._process_army_stats_for_player(player)

    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 as e:
            self._logger.warning("Game resolution failed: %s", e)
            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
        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 self.is_ffa:
            await self.mark_invalid(ValidityState.FFA_NOT_RANKED)
            return

        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

        if self.game_options.get("Victory") not in (
            Victory.DEMORALIZATION,
            Victory.DECAPITATION,
        ):
            await self.mark_invalid(ValidityState.WRONG_VICTORY_CONDITION)
            return

    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 = self.game_options.get("Victory")

        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")
        assert army is not None
        assert isinstance(army, int)

        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})"
        )

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

Subclasses

• CoopGame
• CustomGame
• LadderGame

Class variables

var game_type

The type of the None singleton.

var init_mode

The type of the None singleton.

Instance variables

prop armies : frozenset[int | None]

Expand source code

@property
def armies(self) -> frozenset[Optional[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

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

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.

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]

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

Players in the game

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

prop teams : frozenset[int | None]

Expand source code

@property
def teams(self) -> frozenset[Optional[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
    )

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

Methods

def add_game_connection(self, game_connection)

Expand source code

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

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())

Expand source code

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()

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)

Expand source code

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)

Expand source code

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()
            )
        )

def clear_slot(self, slot_index)

Expand source code

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]

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)

Expand source code

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)

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

Expand source code

def get_army_results(self, player: Player) -> ArmyResult:
    army = self.get_player_option(player.id, "Army")
    assert army is not None
    assert isinstance(army, int)

    return ArmyResult(
        player.id,
        army,
        self.get_player_outcome(player).name,
        self._results.metadata(army),
    )

def get_army_score(self, army)

Expand source code

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

def get_basic_info(self) ‑> BasicGameInfo

Expand source code

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 get_connected_players(self) ‑> list[Player]

Expand source code

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
    ]

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

def get_player_option(self, player_id: int, key: str) ‑> Any | None

Expand source code

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)

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

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

Expand source code

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_team_sets(self) ‑> list[set[Player]]

Expand source code

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

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

Expand source code

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

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)

Expand source code

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")

Mark the game as live.

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

async def load_results(self)

Expand source code

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

Load results from the database

async def mark_invalid(self,
new_validity_state: ValidityState)

Expand source code

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
            )
        )

async def on_game_finish(self)

Expand source code

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 on_game_launched(self)

Expand source code

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 on_scenario_file_changed(self, scenario_path: pathlib._local.PurePath)

Expand source code

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)

Expand source code

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

async def persist_results(self)

Expand source code

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)

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)

Expand source code

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 remove_game_connection(self, game_connection)

Expand source code

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)

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)

Expand source code

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

async def resolve_game_results(self) ‑> EndedGameInfo

Expand source code

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 as e:
        self._logger.warning("Game resolution failed: %s", e)
        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 set_ai_option(self, name, key, value)

Expand source code

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

Set game-associative options for given AI, by name

def set_hosted(self)

Expand source code

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

def set_name_unchecked(self, value: str)

Expand source code

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.
    """
    # mypy sees the column type as a generic base class so it doesn't
    # know about the `length` field
    max_len = game_stats.c.gameName.type.length  # type: ignore
    self._name = value[:max_len]

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)

Expand source code

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

Set game-associative options for given player, by id

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

Expand source code

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()

def to_dict(self)

Expand source code

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
        }
    }

async def update_game_player_stats(self)

Expand source code

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 update_game_stats(self)

Expand source code

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 = self.game_options.get("Victory")

    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,
                )
            )

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)

Expand source code

async def validate_game_mode_settings(self):
    """
    A subset of checks that need to be overridden in coop games.
    """
    if self.is_ffa:
        await self.mark_invalid(ValidityState.FFA_NOT_RANKED)
        return

    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

    if self.game_options.get("Victory") not in (
        Victory.DEMORALIZATION,
        Victory.DECAPITATION,
    ):
        await self.mark_invalid(ValidityState.WRONG_VICTORY_CONDITION)
        return

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

async def validate_game_settings(self)

Expand source code

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
    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()

Mark the game invalid if it has non-compliant options

class GameConnectionState (*args, **kwds)

Expand source code

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var CONNECTED_TO_HOST

The type of the None singleton.

var ENDED

The type of the None singleton.

var INITIALIZED

The type of the None singleton.

var INITIALIZING

The type of the None singleton.

class GameError (*args, **kwargs)

Expand source code

class GameError(Exception):
    pass

Common base class for all non-exit exceptions.

Ancestors

• builtins.Exception
• builtins.BaseException

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

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: dict[str, list[Callable[[Any], Any]]] = defaultdict(list)
        self.async_callbacks: dict[str, list[Callable[[Any], Awaitable[Any]]]] = 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,
            )

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)

Ancestors

• builtins.dict

Methods

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

Expand source code

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

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

Expand source code

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

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

Expand source code

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, ())
    ))

class GameState (*args, **kwds)

Expand source code

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var ENDED

The type of the None singleton.

var INITIALIZING

The type of the None singleton.

var LIVE

The type of the None singleton.

var LOBBY

The type of the None singleton.

class GameType (*args, **kwds)

Expand source code

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var COOP

The type of the None singleton.

var CUSTOM

The type of the None singleton.

var MATCHMAKER

The type of the None singleton.

class InitMode (*args, **kwds)

Expand source code

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var AUTO_LOBBY

The type of the None singleton.

var NORMAL_LOBBY

The type of the None singleton.

class LadderGame (id: int,
database: FAFDatabase,
game_service: GameService,
game_stats_service: GameStatsService,
host: Player | None = 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: int | None = None,
rating_type: str | None = None,
displayed_rating_range: InclusiveRange | None = None,
enforce_rating_range: bool = False,
max_players: int = 12,
setup_timeout: int = 60)

Expand source code

class LadderGame(Game):
    """Class for 1v1 ladder games"""

    init_mode = InitMode.AUTO_LOBBY
    game_type = GameType.MATCHMAKER

    @cached_property
    def _launch_future(self) -> asyncio.Future:
        return asyncio.get_running_loop().create_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

• Game

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

• Game:
  • add_game_connection
  • add_result
  • clear_slot
  • game_type
  • get_connected_players
  • get_player_option
  • get_team_sets
  • init_mode
  • 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 (*args, **kwds)

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var BAD_MAP

The type of the None singleton.

var BAD_MOD

The type of the None singleton.

var BAD_UNIT_RESTRICTIONS

The type of the None singleton.

var CHEATS_ENABLED

The type of the None singleton.

var CIVILIANS_REVEALED

The type of the None singleton.

var COOP_NOT_RANKED

The type of the None singleton.

var EXPANSION_DISABLED

The type of the None singleton.

var FFA_NOT_RANKED

The type of the None singleton.

var HAS_AI_PLAYERS

The type of the None singleton.

var HOST_SET_UNRANKED

The type of the None singleton.

var MULTI_TEAM

The type of the None singleton.

var MUTUAL_DRAW

The type of the None singleton.

var NORUSH_ENABLED

The type of the None singleton.

var NO_FOG_OF_WAR

The type of the None singleton.

var OTHER_UNRANK

The type of the None singleton.

var PREBUILT_ENABLED

The type of the None singleton.

var SINGLE_PLAYER

The type of the None singleton.

var SPAWN_NOT_FIXED

The type of the None singleton.

var TOO_MANY_DESYNCS

The type of the None singleton.

var TOO_SHORT

The type of the None singleton.

var UNEVEN_TEAMS_NOT_RANKED

The type of the None singleton.

var UNKNOWN_RESULT

The type of the None singleton.

var UNLOCKED_TEAMS

The type of the None singleton.

var VALID

The type of the None singleton.

var WRONG_DIFFICULTY

The type of the None singleton.

var WRONG_VICTORY_CONDITION

The type of the None singleton.

class Victory (*args, **kwds)

Expand source code

@unique
class Victory(Enum):
    DEMORALIZATION = "DEMORALIZATION"
    DOMINATION = "DOMINATION"
    ERADICATION = "ERADICATION"
    SANDBOX = "SANDBOX"
    DECAPITATION = "DECAPITATION"

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var DECAPITATION

The type of the None singleton.

var DEMORALIZATION

The type of the None singleton.

var DOMINATION

The type of the None singleton.

var ERADICATION

The type of the None singleton.

var SANDBOX

The type of the None singleton.

class VisibilityState (*args, **kwds)

Expand source code

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

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var FRIENDS

The type of the None singleton.

var PUBLIC

The type of the None singleton.