Classes

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