Module server.games
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
Class variables
var game_type
var init_mode
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
class CustomGame (id, *args, **kwargs)
-
Expand source code
@with_logger class CustomGame(Game): init_mode = InitMode.NORMAL_LOBBY game_type = GameType.CUSTOM def __init__(self, id, *args, **kwargs): new_kwargs = { "rating_type": RatingType.GLOBAL, "setup_timeout": 30 } new_kwargs.update(kwargs) super().__init__(id, *args, **new_kwargs) async def _run_pre_rate_validity_checks(self): limit = len(self.players) * 60 if not self.enforce_rating and time.time() - self.launched_at < limit: await self.mark_invalid(ValidityState.TOO_SHORT)
Object that lasts for the lifetime of a game on FAF.
Ancestors
Class variables
var game_type
var init_mode
Inherited members
Game
:add_game_connection
add_result
clear_slot
get_connected_players
get_player_option
get_team_sets
is_even
is_visible_to_player
launch
load_results
persist_results
players
remove_game_connection
set_ai_option
set_name_unchecked
set_player_option
teams
update_game_stats
validate_game_mode_settings
validate_game_settings
class FeaturedMod (id: int, name: str, full_name: str, description: str, publish: bool, order: int)
-
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
var FAF
var FAFBETA
var LADDER_1V1
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 = [] 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})" )
Object that lasts for the lifetime of a game on FAF.
Subclasses
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
-
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]
-
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 )
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 IDarmy
: the army number being reported forresult_type
: a string representing the resultscore
: an arbitrary number assigned with the resultresult_metadata
: everything preceding theresult_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") 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.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: 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. """ max_len = game_stats.c.gameName.type.length 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 = 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, ) )
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 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)
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 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()
Mark the game invalid if it has non-compliant options
class GameConnectionState (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class GameConnectionState(Enum): INITIALIZING = 0 INITIALIZED = 1 CONNECTED_TO_HOST = 2 ENDED = 3
An enumeration.
Ancestors
- enum.Enum
Class variables
var CONNECTED_TO_HOST
var ENDED
var INITIALIZED
var INITIALIZING
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 = 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, )
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 (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class GameState(Enum): INITIALIZING = 0 LOBBY = 1 LIVE = 2 ENDED = 3
An enumeration.
Ancestors
- enum.Enum
Class variables
var ENDED
var INITIALIZING
var LIVE
var LOBBY
class GameType (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class GameType(Enum): COOP = "coop" CUSTOM = "custom" MATCHMAKER = "matchmaker"
An enumeration.
Ancestors
- enum.Enum
Class variables
var COOP
var CUSTOM
var MATCHMAKER
class InitMode (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class InitMode(Enum): NORMAL_LOBBY = 0 AUTO_LOBBY = 1
An enumeration.
Ancestors
- enum.Enum
Class variables
var AUTO_LOBBY
var NORMAL_LOBBY
class LadderGame (id, *args, **kwargs)
-
Expand source code
class LadderGame(Game): """Class for 1v1 ladder games""" init_mode = InitMode.AUTO_LOBBY game_type = GameType.MATCHMAKER def __init__(self, id, *args, **kwargs): super().__init__(id, *args, **kwargs) self._launch_future = asyncio.Future() async def wait_hosted(self, timeout: float): return await asyncio.wait_for( self._hosted_future, timeout=timeout ) async def wait_launched(self, timeout: float): return await asyncio.wait_for( self._launch_future, timeout=timeout ) async def launch(self): await super().launch() self._launch_future.set_result(None) async def check_game_finish(self, player): if not self._hosted_future.done() and ( self.state in (GameState.INITIALIZING, GameState.LOBBY) ): assert self.host == player self._hosted_future.set_exception(GameClosedError(player)) if not self._launch_future.done() and ( self.state in (GameState.INITIALIZING, GameState.LOBBY) ): self._launch_future.set_exception(GameClosedError(player)) await super().check_game_finish(player) def is_winner(self, player: Player) -> bool: return self.get_player_outcome(player) is ArmyOutcome.VICTORY def get_army_score(self, army: int) -> int: """ We override this function so that ladder game scores are only reported as 1 for win and 0 for anything else. """ return self._results.victory_only_score(army) def _outcome_override_hook(self) -> Optional[list[GameOutcome]]: if not config.LADDER_1V1_OUTCOME_OVERRIDE or len(self.players) > 2: return None team_sets = self.get_team_sets() army_scores = [ self._results.score(self.get_player_option(team_set.pop().id, "Army")) for team_set in team_sets ] if army_scores[0] > army_scores[1]: return [GameOutcome.VICTORY, GameOutcome.DEFEAT] elif army_scores[0] < army_scores[1]: return [GameOutcome.DEFEAT, GameOutcome.VICTORY] else: return [GameOutcome.DRAW, GameOutcome.DRAW]
Class for 1v1 ladder games
Ancestors
Class variables
var game_type
var init_mode
Methods
async def check_game_finish(self, player)
-
Expand source code
async def check_game_finish(self, player): if not self._hosted_future.done() and ( self.state in (GameState.INITIALIZING, GameState.LOBBY) ): assert self.host == player self._hosted_future.set_exception(GameClosedError(player)) if not self._launch_future.done() and ( self.state in (GameState.INITIALIZING, GameState.LOBBY) ): self._launch_future.set_exception(GameClosedError(player)) await super().check_game_finish(player)
def get_army_score(self, army: int) ‑> int
-
Expand source code
def get_army_score(self, army: int) -> int: """ We override this function so that ladder game scores are only reported as 1 for win and 0 for anything else. """ return self._results.victory_only_score(army)
We override this function so that ladder game scores are only reported as 1 for win and 0 for anything else.
def is_winner(self,
player: Player) ‑> bool-
Expand source code
def is_winner(self, player: Player) -> bool: return self.get_player_outcome(player) is ArmyOutcome.VICTORY
async def wait_hosted(self, timeout: float)
-
Expand source code
async def wait_hosted(self, timeout: float): return await asyncio.wait_for( self._hosted_future, timeout=timeout )
async def wait_launched(self, timeout: float)
-
Expand source code
async def wait_launched(self, timeout: float): return await asyncio.wait_for( self._launch_future, timeout=timeout )
Inherited members
Game
:add_game_connection
add_result
clear_slot
get_connected_players
get_player_option
get_team_sets
is_even
is_visible_to_player
launch
load_results
persist_results
players
remove_game_connection
set_ai_option
set_name_unchecked
set_player_option
teams
update_game_stats
validate_game_mode_settings
validate_game_settings
class ValidityState (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
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
An enumeration.
Ancestors
- enum.Enum
Class variables
var BAD_MAP
var BAD_MOD
var BAD_UNIT_RESTRICTIONS
var CHEATS_ENABLED
var CIVILIANS_REVEALED
var COOP_NOT_RANKED
var EXPANSION_DISABLED
var FFA_NOT_RANKED
var HAS_AI_PLAYERS
var HOST_SET_UNRANKED
var MULTI_TEAM
var MUTUAL_DRAW
var NORUSH_ENABLED
var NO_FOG_OF_WAR
var OTHER_UNRANK
var PREBUILT_ENABLED
var SINGLE_PLAYER
var SPAWN_NOT_FIXED
var TOO_MANY_DESYNCS
var TOO_SHORT
var UNEVEN_TEAMS_NOT_RANKED
var UNKNOWN_RESULT
var UNLOCKED_TEAMS
var VALID
var WRONG_DIFFICULTY
var WRONG_VICTORY_CONDITION
class Victory (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class Victory(Enum): DEMORALIZATION = 0 DOMINATION = 1 ERADICATION = 2 SANDBOX = 3
An enumeration.
Ancestors
- enum.Enum
Class variables
var DEMORALIZATION
var DOMINATION
var ERADICATION
var SANDBOX
class VisibilityState (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
Expand source code
@unique class VisibilityState(Enum): PUBLIC = "public" FRIENDS = "friends"
An enumeration.
Ancestors
- enum.Enum
Class variables
var FRIENDS
var PUBLIC