Module server.games.game_results

Functions

def resolve_game(team_outcomes: list[set[ArmyOutcome]]) ‑> list[GameOutcome]
Expand source code
def resolve_game(team_outcomes: list[set[ArmyOutcome]]) -> list[GameOutcome]:
    """
    Takes a list of length two containing sets of ArmyOutcome
    for individual players on a team
    and converts a list of two GameOutcomes,
    either VICTORY and DEFEAT or DRAW and DRAW.

    # Params
    - `team_outcomes`: list of `GameOutcomes`

    # Errors
    Throws `GameResolutionError` if outcomes are inconsistent or ambiguous.

    # Returns
    A list of ranks as to be used with trueskill
    """
    if len(team_outcomes) != 2:
        raise GameResolutionError(
            "Will not resolve game with other than two parties."
        )

    victory0 = ArmyOutcome.VICTORY in team_outcomes[0]
    victory1 = ArmyOutcome.VICTORY in team_outcomes[1]
    both_claim_victory = victory0 and victory1
    someone_claims_victory = victory0 or victory1
    if both_claim_victory:
        raise GameResolutionError(
            "Cannot resolve game in which both teams claimed victory. "
            f" Team outcomes: {team_outcomes}"
        )
    elif someone_claims_victory:
        return [
            GameOutcome.VICTORY
            if ArmyOutcome.VICTORY in outcomes
            else GameOutcome.DEFEAT
            for outcomes in team_outcomes
        ]

    # Now know that no-one has GameOutcome.VICTORY
    draw0 = ArmyOutcome.DRAW in team_outcomes[0]
    draw1 = ArmyOutcome.DRAW in team_outcomes[1]
    both_claim_draw = draw0 and draw1
    someone_claims_draw = draw0 or draw1
    if both_claim_draw:
        return [GameOutcome.DRAW, GameOutcome.DRAW]
    elif someone_claims_draw:
        raise GameResolutionError(
            "Cannot resolve game with unilateral draw. "
            f"Team outcomes: {team_outcomes}"
        )

    # Now know that the only results are DEFEAT or UNKNOWN/CONFLICTING
    # Unrank if there are any players with unknown result
    all_outcomes = team_outcomes[0] | team_outcomes[1]
    if (
        ArmyOutcome.UNKNOWN in all_outcomes
        or ArmyOutcome.CONFLICTING in all_outcomes
    ):
        raise GameResolutionError(
            "Cannot resolve game with ambiguous outcome. "
            f" Team outcomes: {team_outcomes}"
        )

    # Otherwise everyone is DEFEAT, we return a draw
    return [GameOutcome.DRAW, GameOutcome.DRAW]

Takes a list of length two containing sets of ArmyOutcome for individual players on a team and converts a list of two GameOutcomes, either VICTORY and DEFEAT or DRAW and DRAW.

Params

  • team_outcomes: list of GameOutcomes

Errors

Throws GameResolutionError if outcomes are inconsistent or ambiguous.

Returns

A list of ranks as to be used with trueskill

Classes

class ArmyOutcome (value, names=None, *, module=None, qualname=None, type=None, start=1)
Expand source code
class ArmyOutcome(Enum):
    """
    The resolved outcome for an army. Each army has only one of these.
    """
    VICTORY = "VICTORY"
    DEFEAT = "DEFEAT"
    DRAW = "DRAW"
    UNKNOWN = "UNKNOWN"
    CONFLICTING = "CONFLICTING"

The resolved outcome for an army. Each army has only one of these.

Ancestors

  • enum.Enum

Class variables

var CONFLICTING
var DEFEAT
var DRAW
var UNKNOWN
var VICTORY
class ArmyReportedOutcome (value, names=None, *, module=None, qualname=None, type=None, start=1)
Expand source code
class ArmyReportedOutcome(Enum):
    """
    A reported outcome for an army. Each army may have several of these.
    """
    VICTORY = "VICTORY"
    DEFEAT = "DEFEAT"
    DRAW = "DRAW"
    # This doesn't seem to be reported by the game anymore
    MUTUAL_DRAW = "MUTUAL_DRAW"

    def to_resolved(self) -> ArmyOutcome:
        """
        Convert this result to the resolved version. For when all reported
        results agree.
        """
        value = self.value
        if value == "MUTUAL_DRAW":
            value = "DRAW"
        return ArmyOutcome(value)

A reported outcome for an army. Each army may have several of these.

Ancestors

  • enum.Enum

Class variables

var DEFEAT
var DRAW
var MUTUAL_DRAW
var VICTORY

Methods

def to_resolved(self) ‑> ArmyOutcome
Expand source code
def to_resolved(self) -> ArmyOutcome:
    """
    Convert this result to the resolved version. For when all reported
    results agree.
    """
    value = self.value
    if value == "MUTUAL_DRAW":
        value = "DRAW"
    return ArmyOutcome(value)

Convert this result to the resolved version. For when all reported results agree.

class ArmyResult (player_id: int, army: int | None, army_outcome: str, metadata: list[str])
Expand source code
class ArmyResult(NamedTuple):
    """
    Broadcast in the end of game rabbitmq message.
    """
    player_id: int
    army: Optional[int]
    army_outcome: str
    metadata: list[str]

Broadcast in the end of game rabbitmq message.

Ancestors

  • builtins.tuple

Instance variables

var army : int | None
Expand source code
class ArmyResult(NamedTuple):
    """
    Broadcast in the end of game rabbitmq message.
    """
    player_id: int
    army: Optional[int]
    army_outcome: str
    metadata: list[str]

Alias for field number 1

var army_outcome : str
Expand source code
class ArmyResult(NamedTuple):
    """
    Broadcast in the end of game rabbitmq message.
    """
    player_id: int
    army: Optional[int]
    army_outcome: str
    metadata: list[str]

Alias for field number 2

var metadata : list[str]
Expand source code
class ArmyResult(NamedTuple):
    """
    Broadcast in the end of game rabbitmq message.
    """
    player_id: int
    army: Optional[int]
    army_outcome: str
    metadata: list[str]

Alias for field number 3

var player_id : int
Expand source code
class ArmyResult(NamedTuple):
    """
    Broadcast in the end of game rabbitmq message.
    """
    player_id: int
    army: Optional[int]
    army_outcome: str
    metadata: list[str]

Alias for field number 0

class GameOutcome (value, names=None, *, module=None, qualname=None, type=None, start=1)
Expand source code
@unique
class GameOutcome(Enum):
    VICTORY = "VICTORY"
    DEFEAT = "DEFEAT"
    DRAW = "DRAW"
    UNKNOWN = "UNKNOWN"

An enumeration.

Ancestors

  • enum.Enum

Class variables

var DEFEAT
var DRAW
var UNKNOWN
var VICTORY
class GameResolutionError (*args, **kwargs)
Expand source code
class GameResolutionError(Exception):
    pass

Common base class for all non-exit exceptions.

Ancestors

  • builtins.Exception
  • builtins.BaseException
class GameResultReport (reporter: int,
army: int,
outcome: ArmyReportedOutcome,
score: int,
metadata: frozenset[str] = frozenset())
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

These are sent from each player's FA when they quit the game. 'Score' depends on the number of ACUs killed, whether the player died, maybe other factors.

Ancestors

  • builtins.tuple

Instance variables

var army : int
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

Alias for field number 1

var metadata : frozenset[str]
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

Alias for field number 4

var outcomeArmyReportedOutcome
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

Alias for field number 2

var reporter : int
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

Alias for field number 0

var score : int
Expand source code
class GameResultReport(NamedTuple):
    """
    These are sent from each player's FA when they quit the game. 'Score'
    depends on the number of ACUs killed, whether the player died, maybe other
    factors.
    """

    reporter: int
    army: int
    outcome: ArmyReportedOutcome
    score: int
    metadata: frozenset[str] = frozenset()

Alias for field number 3

class GameResultReports (game_id: int)
Expand source code
@with_logger
class GameResultReports(Mapping):
    """
    Collects all results from a single game. Allows to determine results for an
    army and game as a whole. Supports a dict-like access to lists of results
    for each army, but don't modify these.
    """

    def __init__(self, game_id: int):
        Mapping.__init__(self)
        self._game_id = game_id  # Just for logging
        self._back: dict[int, list[GameResultReport]] = {}
        # Outcome caching
        self._outcomes: dict[int, ArmyOutcome] = {}
        self._dirty_armies: set[int] = set()

    def __getitem__(self, key: int) -> list[GameResultReport]:
        return self._back[key]

    def __iter__(self) -> Iterator[int]:
        return iter(self._back)

    def __len__(self) -> int:
        return len(self._back)

    def add(self, result: GameResultReport) -> None:
        army_results = self._back.setdefault(result.army, [])
        army_results.append(result)
        self._dirty_armies.add(result.army)

    def outcome(self, army: int) -> ArmyOutcome:
        """
        Determines what the outcome was for a given army.
        Returns the unique reported outcome if all players agree,
        or the majority outcome if only a few reports disagree.
        Otherwise returns CONFLICTING if there is too much disagreement
        or UNKNOWN if no reports were filed.
        """
        if army not in self._outcomes or army in self._dirty_armies:
            self._outcomes[army] = self._compute_outcome(army)
            self._dirty_armies.discard(army)

        return self._outcomes[army]

    def _compute_outcome(self, army: int) -> ArmyOutcome:
        if army not in self:
            return ArmyOutcome.UNKNOWN

        voters = defaultdict(set)
        for report in self[army]:
            voters[report.outcome].add(report.reporter)

        if len(voters) == 0:
            return ArmyOutcome.UNKNOWN

        if len(voters) == 1:
            unique_outcome = voters.popitem()[0]
            return unique_outcome.to_resolved()

        sorted_outcomes = sorted(
            voters.keys(),
            reverse=True,
            key=lambda outcome: (len(voters[outcome]), outcome.value),
        )

        top_votes = len(voters[sorted_outcomes[0]])
        runner_up_votes = len(voters[sorted_outcomes[1]])
        if top_votes > 1 >= runner_up_votes or top_votes >= runner_up_votes + 3:
            decision = sorted_outcomes[0].to_resolved()
        else:
            decision = ArmyOutcome.CONFLICTING

        self._logger.info(
            "Multiple outcomes for game %s army %s resolved to %s. Reports are: %s",
            self._game_id, army, decision, voters,
        )
        return decision

    def metadata(self, army: int) -> list[str]:
        """
        If any users have sent metadata tags in their messages about this army
        this function will compare those tags across all messages trying to find
        common ones to return.
        """
        if army not in self:
            return []

        all_metadata = [report.metadata for report in self[army]]
        metadata_count = Counter(all_metadata).most_common()

        if len(metadata_count) == 1:
            # Everyone agrees!
            return sorted(list(metadata_count[0][0]))

        most_common, next_most_common, *_ = metadata_count
        if most_common[1] > next_most_common[1]:
            resolved_to = sorted(list(most_common[0]))
            self._logger.info(
                "Conflicting metadata for game %s army %s resolved to %s. Reports are: %s",
                self._game_id, army, resolved_to, all_metadata,
            )
            return resolved_to

        # We have a tie
        self._logger.info(
            "Conflicting metadata for game %s army %s, unable to resolve. Reports are: %s",
            self._game_id, army, all_metadata,
        )
        return []

    def score(self, army: int) -> int:
        """
        Pick and return most frequently reported score for an army. If multiple
        scores are most frequent, pick the largest one. Returns 0 if there are
        no results for a given army.
        """
        if army not in self:
            return 0

        scores = Counter(r.score for r in self[army])
        if len(scores) == 1:
            return scores.popitem()[0]

        self._logger.info(
            "Conflicting scores (%s) reported for game %s", scores, self._game_id
        )
        score, _ = max(scores.items(), key=lambda kv: kv[::-1])
        return score

    def victory_only_score(self, army: int) -> int:
        """
        Calculate our own score depending *only* on victory.
        """
        if army not in self:
            return 0

        if self.outcome(army) is ArmyOutcome.VICTORY:
            return 1
        else:
            return 0

    @classmethod
    async def from_db(cls, database, game_id):
        results = cls(game_id)
        async with database.acquire() as conn:
            result = await conn.execute(
                select(
                    game_player_stats.c.place,
                    game_player_stats.c.score,
                    game_player_stats.c.result
                ).where(game_player_stats.c.gameId == game_id)
            )

            for row in result:
                # FIXME: Assertion about startspot == army
                with contextlib.suppress(ValueError):
                    outcome = ArmyReportedOutcome(row.result.value)
                    report = GameResultReport(0, row.place, outcome, row.score)
                    results.add(report)
        return results

Collects all results from a single game. Allows to determine results for an army and game as a whole. Supports a dict-like access to lists of results for each army, but don't modify these.

Ancestors

  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Static methods

async def from_db(database, game_id)

Methods

def add(self,
result: GameResultReport) ‑> None
Expand source code
def add(self, result: GameResultReport) -> None:
    army_results = self._back.setdefault(result.army, [])
    army_results.append(result)
    self._dirty_armies.add(result.army)
def metadata(self, army: int) ‑> list[str]
Expand source code
def metadata(self, army: int) -> list[str]:
    """
    If any users have sent metadata tags in their messages about this army
    this function will compare those tags across all messages trying to find
    common ones to return.
    """
    if army not in self:
        return []

    all_metadata = [report.metadata for report in self[army]]
    metadata_count = Counter(all_metadata).most_common()

    if len(metadata_count) == 1:
        # Everyone agrees!
        return sorted(list(metadata_count[0][0]))

    most_common, next_most_common, *_ = metadata_count
    if most_common[1] > next_most_common[1]:
        resolved_to = sorted(list(most_common[0]))
        self._logger.info(
            "Conflicting metadata for game %s army %s resolved to %s. Reports are: %s",
            self._game_id, army, resolved_to, all_metadata,
        )
        return resolved_to

    # We have a tie
    self._logger.info(
        "Conflicting metadata for game %s army %s, unable to resolve. Reports are: %s",
        self._game_id, army, all_metadata,
    )
    return []

If any users have sent metadata tags in their messages about this army this function will compare those tags across all messages trying to find common ones to return.

def outcome(self, army: int) ‑> ArmyOutcome
Expand source code
def outcome(self, army: int) -> ArmyOutcome:
    """
    Determines what the outcome was for a given army.
    Returns the unique reported outcome if all players agree,
    or the majority outcome if only a few reports disagree.
    Otherwise returns CONFLICTING if there is too much disagreement
    or UNKNOWN if no reports were filed.
    """
    if army not in self._outcomes or army in self._dirty_armies:
        self._outcomes[army] = self._compute_outcome(army)
        self._dirty_armies.discard(army)

    return self._outcomes[army]

Determines what the outcome was for a given army. Returns the unique reported outcome if all players agree, or the majority outcome if only a few reports disagree. Otherwise returns CONFLICTING if there is too much disagreement or UNKNOWN if no reports were filed.

def score(self, army: int) ‑> int
Expand source code
def score(self, army: int) -> int:
    """
    Pick and return most frequently reported score for an army. If multiple
    scores are most frequent, pick the largest one. Returns 0 if there are
    no results for a given army.
    """
    if army not in self:
        return 0

    scores = Counter(r.score for r in self[army])
    if len(scores) == 1:
        return scores.popitem()[0]

    self._logger.info(
        "Conflicting scores (%s) reported for game %s", scores, self._game_id
    )
    score, _ = max(scores.items(), key=lambda kv: kv[::-1])
    return score

Pick and return most frequently reported score for an army. If multiple scores are most frequent, pick the largest one. Returns 0 if there are no results for a given army.

def victory_only_score(self, army: int) ‑> int
Expand source code
def victory_only_score(self, army: int) -> int:
    """
    Calculate our own score depending *only* on victory.
    """
    if army not in self:
        return 0

    if self.outcome(army) is ArmyOutcome.VICTORY:
        return 1
    else:
        return 0

Calculate our own score depending only on victory.