Module server.matchmaker.algorithm.stable_marriage

def avg_mean(search: Search) -> float:
    """
    Get the average of all trueskill means for a search counting means with
    high deviation as 0.
    """
    return stats.mean(mean if dev < 250 else 0 for mean, dev in search.ratings)

class StableMarriage(MatchmakingPolicy1v1):
    def find(self, ranks: WeightedGraph) -> dict[Search, Search]:
        """Perform the stable matching algorithm until a maximal stable matching
        is found.
        """
        self.matches.clear()

        max_degree = max((len(edges) for edges in ranks.values()), default=0)
        for i in range(max_degree):
            self._logger.debug(
                "Round %i of stable marriage, currently %i matches",
                i,
                len(self.matches) // 2,
            )
            # Do one round of proposals
            if len(self.matches) == len(ranks):
                # Everyone found a match so we are done
                break

            for search in ranks:
                if search in self.matches:
                    continue

                if not ranks[search]:
                    # Preference list exhausted
                    continue

                preferred, quality = ranks[search].pop()

                self._logger.debug(
                    "Quality between %s and %s: %f thresholds: [%f, %f]",
                    search,
                    preferred,
                    quality,
                    search.match_threshold,
                    preferred.match_threshold,
                )

                self._propose(search, preferred, quality)

        return self.matches

    def _propose(self, search: Search, preferred: Search, new_quality: float):
        """An unmatched search proposes to it's preferred opponent.

        If the opponent is not matched, they become matched. If the opponent is
        matched, but prefers this new search to its current one, then the opponent
        unmatches from its previous adversary and matches with the new search instead.
        """
        if preferred not in self.matches:
            self._match(search, preferred)
            return

        current_match = self.matches[preferred]
        current_quality = preferred.quality_with(current_match)

        if new_quality > current_quality:
            # Found a better match
            self._unmatch(preferred)
            self._match(search, preferred)

@with_logger
class StableMarriageMatchmaker(Matchmaker):
    """
    Runs stable marriage to produce a list of matches
    and afterwards adds random matchups for previously unmatched new players.
    """

    def find(
        self, searches: Iterable[Search], team_size: int, rating_peak: float
    ) -> tuple[list[Match], list[Search]]:
        if team_size != 1:
            self._logger.error(
                "Invalid team size %i for stable marriage matchmaker will be ignored",
                team_size,
            )

        searches = list(searches)
        matches: dict[Search, Search] = {}

        self._logger.debug("Matching with stable marriage...")
        if len(searches) < 30:
            ranks = _MatchingGraph.build_full(searches)
        else:
            ranks = _MatchingGraph.build_fast(searches)
        _MatchingGraph.remove_isolated(ranks)
        matches.update(StableMarriage().find(ranks))

        unmatched_searches = [
            search for search in searches if search not in matches
        ]

        return self._remove_duplicates(matches), unmatched_searches

    @staticmethod
    def _remove_duplicates(matches: dict[Search, Search]) -> list[Match]:
        matches_set: set[Match] = set()
        for s1, s2 in matches.items():
            if (s1, s2) in matches_set or (s2, s1) in matches_set:
                continue
            matches_set.add((s1, s2))
        return list(matches_set)