From 1a0573cba9a1f5d311fce86b38a79316530b4398 Mon Sep 17 00:00:00 2001 From: David Rodenkirchen Date: Tue, 27 Jan 2026 18:24:30 +0100 Subject: [PATCH] add tournament data model --- requirements.txt | Bin 228 -> 218 bytes src/ezgg_lan_manager/types/Game.py | 39 +++ src/ezgg_lan_manager/types/Match.py | 133 ++++++++ src/ezgg_lan_manager/types/Participant.py | 20 ++ src/ezgg_lan_manager/types/Tournament.py | 319 +++++++++++++++++++ src/ezgg_lan_manager/types/TournamentBase.py | 51 +++ testing/unittests/TournamentDomainTests.py | 66 ++++ 7 files changed, 628 insertions(+) create mode 100644 src/ezgg_lan_manager/types/Game.py create mode 100644 src/ezgg_lan_manager/types/Match.py create mode 100644 src/ezgg_lan_manager/types/Participant.py create mode 100644 src/ezgg_lan_manager/types/Tournament.py create mode 100644 src/ezgg_lan_manager/types/TournamentBase.py create mode 100644 testing/unittests/TournamentDomainTests.py diff --git a/requirements.txt b/requirements.txt index 2a749b4f5ef783a7645599d1596d7db63e3b2a89..d2dea2e3f934aebec47d8fe918599e0abf17e341 100644 GIT binary patch delta 11 ScmaFDc#CmD9HY_1%xVA|galUr delta 21 ccmcb`_=Isn9Je8Z9)l4>5koSA*~Iv207H%i!2kdN diff --git a/src/ezgg_lan_manager/types/Game.py b/src/ezgg_lan_manager/types/Game.py new file mode 100644 index 0000000..1f70f91 --- /dev/null +++ b/src/ezgg_lan_manager/types/Game.py @@ -0,0 +1,39 @@ +from typing import Optional + +from src.ezgg_lan_manager.types.TournamentBase import TournamentError + + +class Game: + def __init__(self, id_: tuple[int, int], match_id: int, game_number: int, winner_id: Optional[int], score: Optional[tuple[int, int]], game_done: bool) -> None: + self._id = id_ + self._match_id = match_id + self._game_number = game_number + self._winner_id = winner_id + self._score = score + self._done = game_done + + @property + def id(self) -> tuple[int, int]: + return self._id + + @property + def is_done(self) -> bool: + return self._done + + @property + def winner(self) -> Optional[int]: + return self._winner_id + + @property + def number(self) -> int: + return self._game_number + + + def finish(self, winner_id: int, score: tuple[int, int], force: bool = False) -> None: + """ NEVER call this outside Match or a Testsuite """ + if self._done and not force: + raise TournamentError("Game is already finished") + + self._winner_id = winner_id + self._score = score + self._done = True diff --git a/src/ezgg_lan_manager/types/Match.py b/src/ezgg_lan_manager/types/Match.py new file mode 100644 index 0000000..409d00e --- /dev/null +++ b/src/ezgg_lan_manager/types/Match.py @@ -0,0 +1,133 @@ +from collections import Counter + +from math import ceil +from typing import Literal, Optional, Callable + +from src.ezgg_lan_manager.types.Game import Game +from src.ezgg_lan_manager.types.TournamentBase import MatchStatus, TournamentError, Bracket + + +class MatchParticipant: + def __init__(self, participant_id: int, slot_number: Literal[1, 2]) -> None: + self._participant_id = participant_id + if slot_number not in (1, 2): + raise TournamentError("Invalid slot number") + self.slot_number = slot_number + + @property + def participant_id(self) -> int: + return self._participant_id + + +class Match: + def __init__(self, + match_id: int, + tournament_id: int, + round_number: int, + bracket: Bracket, + best_of: int, + status: MatchStatus, + next_match_win_lose_ids: tuple[Optional[int], Optional[int]], + match_has_ended_callback: Callable) -> None: + self._match_id = match_id + self._tournament_id = tournament_id + self._round_number = round_number + self._bracket = bracket + self._best_of = best_of + self._status = status + self._next_match_win_id = next_match_win_lose_ids[0] + self._next_match_lose_id = next_match_win_lose_ids[1] + self._match_has_ended_callback = match_has_ended_callback + + self._participants: list[MatchParticipant] = [] + self._games: tuple[Game] = self._prepare_games() + + def _prepare_games(self) -> tuple[Game]: + games = [] + for game_number in range(1, self._best_of + 1): + game_id = (self._match_id, game_number) + games.append(Game(game_id, self._match_id, game_number, None, None, False)) + return tuple(games) + + @property + def status(self) -> MatchStatus: + if self._status == MatchStatus.COMPLETED: + return self._status + return self._status if self.is_fully_seeded else MatchStatus.WAITING + + @status.setter + def status(self, new_status: MatchStatus) -> None: + if new_status in (MatchStatus.COMPLETED, MatchStatus.PENDING) and not self.is_fully_seeded: + raise TournamentError("Can't complete/pend match that is not fully seeded") + if self._status == MatchStatus.COMPLETED and new_status != MatchStatus.CANCELED: + raise TournamentError("Can't change COMPLETED match back to another active status") + self._status = new_status + + @property + def games(self) -> tuple[Game]: + return self._games + + @property + def winner(self) -> Optional[int]: + wins_needed = ceil(self._best_of / 2) + counts = Counter(game.winner for game in self._games if game.is_done) + + for participant_id, wins in counts.items(): + if wins >= wins_needed: + return participant_id + + return None + + @property + def is_fully_seeded(self) -> bool: + slots = {p.slot_number for p in self._participants} + return slots == {1, 2} + + @property + def match_id(self) -> int: + return self._match_id + + @property + def participants(self) -> list[MatchParticipant]: + return self._participants + + @property + def next_match_win_id(self) -> Optional[int]: + return self._next_match_win_id + + @property + def next_match_lose_id(self) -> Optional[int]: + return self._next_match_lose_id + + def assign_participant(self, participant_id: int, slot: Literal[1, 2]) -> None: + new_participant = MatchParticipant(participant_id, slot) + if len(self._participants) < 2 and not any(p.participant_id == participant_id for p in self._participants): + if len(self._participants) == 1 and self._participants[-1].slot_number == new_participant.slot_number: + raise TournamentError(f"Match with ID {self._match_id} encountered slot collision") + self._participants.append(new_participant) + return + raise TournamentError(f"Match with ID {self._match_id} already has the maximum number of participants") + + def check_completion(self) -> None: + winner = self.winner + if winner is not None: + self._match_has_ended_callback(self) + self._status = MatchStatus.COMPLETED + + def report_game_result(self, game_number: int, winner_id: int, score: tuple[int, int]) -> None: + if winner_id not in {p.participant_id for p in self._participants}: + raise TournamentError("Winner is not a participant of this match") + + self._games[game_number - 1].finish(winner_id, score) + + self.check_completion() + + def cancel(self) -> None: + self._status = MatchStatus.CANCELED + + def __repr__(self) -> str: + participants = ", ".join( + f"{p.participant_id} (slot {p.slot_number})" for p in self._participants + ) + return (f"") diff --git a/src/ezgg_lan_manager/types/Participant.py b/src/ezgg_lan_manager/types/Participant.py new file mode 100644 index 0000000..4ac1bc1 --- /dev/null +++ b/src/ezgg_lan_manager/types/Participant.py @@ -0,0 +1,20 @@ +from src.ezgg_lan_manager.types.TournamentBase import ParticipantType + + +class Participant: + def __init__(self, id_: int, display_name: str, participant_type: ParticipantType) -> None: + self._id = id_ + self._participant_type = participant_type + self._display_name = display_name + + @property + def id(self) -> int: + return self._id + + @property + def participant_type(self) -> ParticipantType: + return self._participant_type + + @property + def display_name(self) -> str: + return self._display_name diff --git a/src/ezgg_lan_manager/types/Tournament.py b/src/ezgg_lan_manager/types/Tournament.py new file mode 100644 index 0000000..f83ebe3 --- /dev/null +++ b/src/ezgg_lan_manager/types/Tournament.py @@ -0,0 +1,319 @@ +import uuid +from datetime import datetime +from typing import Optional +from math import ceil, log2 + +from src.ezgg_lan_manager.types.Match import Match +from src.ezgg_lan_manager.types.Participant import Participant +from src.ezgg_lan_manager.types.TournamentBase import GameTitle, TournamentFormat, TournamentStatus, TournamentError, Bracket, MatchStatus + + +class Tournament: + def __init__(self, + id_: int, + name: str, + game_title: GameTitle, + format_: TournamentFormat, + start_time: datetime, + status: TournamentStatus, + participants: list[Participant], + matches: Optional[tuple[Match]], + rounds: list[list[Match]]) -> None: + self._id = id_ + self._name = name + self._game_title = game_title + self._format = format_ + self._start_time = start_time + self._status = status + self._participants = participants + self._matches = matches + self._rounds = rounds + + @property + def id(self) -> int: + return self._id + + @property + def name(self) -> str: + return self._name + + @property + def game_title(self) -> GameTitle: + return self._game_title + + @property + def format(self) -> TournamentFormat: + return self._format + + @property + def start_time(self) -> datetime: + return self._start_time + + @property + def status(self) -> TournamentStatus: + return self._status + + @status.setter + def status(self, new_status: TournamentStatus) -> None: + if new_status == TournamentStatus.OPEN and self._status == TournamentStatus.CLOSED and self._matches is not None: + # Deletes all tournament preparation ! + self._matches = None + self._status = new_status + + + @property + def participants(self) -> list[Participant]: + return self._participants + + @property + def matches(self) -> list[Match]: + return self._matches if self._matches else [] + + def add_participant(self, participant: Participant) -> None: + if participant.id in (p.id for p in self._participants): + raise TournamentError(f"Participant with ID {participant.id} already registered for tournament") + self._participants.append(participant) + + + def match_has_ended_callback(self, match: Match) -> None: + if self._matches is None: + return + + winner = match.winner + next_match = next((m for m in self._matches if m.match_id == match.next_match_win_id), None) + if next_match is not None: + try: + next_match.assign_participant(winner, 1) + except TournamentError: + next_match.assign_participant(winner, 2) + else: # No next match = final round + pass + + if match.next_match_lose_id is not None: + loser = next(p for p in match.participants if p.participant_id != winner) + next_match = next((m for m in self._matches if m.match_id == match.next_match_lose_id), None) + if next_match is not None: + try: + next_match.assign_participant(loser.participant_id, 1) + except TournamentError: + next_match.assign_participant(loser.participant_id, 2) + else: # No next match = final round + pass + + def start(self) -> None: + """ This builds the tournament tree and sets it to ONGOING """ + + def parse_format(fmt: TournamentFormat) -> tuple[str, int]: + if fmt.name.startswith("SINGLE_ELIMINATION"): + bracket = "SINGLE" + elif fmt.name.startswith("DOUBLE_ELIMINATION"): + bracket = "DOUBLE" + else: + raise TournamentError(f"Unsupported tournament format: {fmt}") + + if fmt.name.endswith("_BO_1"): + bo = 1 + elif fmt.name.endswith("_BO_3"): + bo = 3 + elif fmt.name.endswith("_BO_5"): + bo = 5 + else: + raise TournamentError(f"Unsupported best-of in format: {fmt}") + + return bracket, bo + + if len(self._participants) < 2: + raise TournamentError("Cannot start tournament: not enough participants") + + bracket_type, best_of = parse_format(self._format) + num_participants = len(self.participants) + match_id_counter = 1 + + if bracket_type == "SINGLE": + # --- single-elimination as before --- + num_rounds = ceil(log2(num_participants)) + rounds: list[list[Match]] = [] + + for round_number in range(1, num_rounds + 1): + num_matches = 2 ** (num_rounds - round_number) + round_matches = [] + for _ in range(num_matches): + match = Match( + match_id=match_id_counter, + tournament_id=self._id, + round_number=round_number, + bracket=Bracket.UPPER if round_number != num_rounds else Bracket.FINAL, + best_of=best_of, + status=MatchStatus.WAITING, + next_match_win_lose_ids=(None, None), + match_has_ended_callback=self.match_has_ended_callback + ) + round_matches.append(match) + match_id_counter += 1 + rounds.append(round_matches) + + # Link winner IDs + for i in range(len(rounds) - 1): + current_round = rounds[i] + next_round = rounds[i + 1] + for idx, match in enumerate(current_round): + next_match = next_round[idx // 2] + match._next_match_win_id = next_match.match_id + + # Assign participants to first round + participant_iter = iter(self.participants) + first_round = rounds[0] + for match in first_round: + try: + p1 = next(participant_iter) + match.assign_participant(p1.id, 1) + except StopIteration: + continue + try: + p2 = next(participant_iter) + match.assign_participant(p2.id, 2) + except StopIteration: + if not match.is_fully_seeded: # Auto-Bye + for game in match.games: + match.report_game_result(game.number, p1.id, (1, 0)) + continue + + if match.is_fully_seeded: + match.status = MatchStatus.PENDING + + + # Flatten all rounds + self._matches = [m for round_matches in rounds for m in round_matches] + + elif bracket_type == "DOUBLE": + # --- double-elimination bracket generation --- + # ToDo: Rounds are not correctly persisted into self._rounds here. What data structure to use? + # ToDo: Bye-Handling not done + # Implementation Notice: Do not implement yet! + num_rounds_upper = ceil(log2(num_participants)) + upper_rounds: list[list[Match]] = [] + for round_number in range(1, num_rounds_upper + 1): + num_matches = 2 ** (num_rounds_upper - round_number) + round_matches = [] + for _ in range(num_matches): + match = Match( + match_id=match_id_counter, + tournament_id=self._id, + round_number=round_number, + bracket=Bracket.UPPER, + best_of=best_of, + status=MatchStatus.WAITING, + next_match_win_lose_ids=(None, None), # will fill later + match_has_ended_callback=self.match_has_ended_callback + ) + round_matches.append(match) + match_id_counter += 1 + upper_rounds.append(round_matches) + + # Lower bracket (Losers) + # Double-elim lower bracket has roughly (2*num_rounds_upper - 2) rounds + num_rounds_lower = 2 * (num_rounds_upper - 1) + lower_rounds: list[list[Match]] = [] + for round_number in range(1, num_rounds_lower + 1): + num_matches = 2 ** (num_rounds_lower - round_number - 1) if round_number != 1 else 2 ** (num_rounds_upper - 1) + round_matches = [] + for _ in range(num_matches): + match = Match( + match_id=match_id_counter, + tournament_id=self._id, + round_number=round_number, + bracket=Bracket.LOWER, + best_of=best_of, + status=MatchStatus.WAITING, + next_match_win_lose_ids=(None, None), + match_has_ended_callback=self.match_has_ended_callback + ) + round_matches.append(match) + match_id_counter += 1 + lower_rounds.append(round_matches) + + # Link upper bracket winners to next upper-round matches + for i in range(len(upper_rounds) - 1): + for idx, match in enumerate(upper_rounds[i]): + next_match = upper_rounds[i + 1][idx // 2] + match._next_match_win_id = next_match.match_id + + # Link upper bracket losers to lower bracket first rounds + lower_round1 = lower_rounds[0] if lower_rounds else [] + for idx, match in enumerate(upper_rounds[0]): + if idx < len(lower_round1): + match._next_match_lose_id = lower_round1[idx].match_id + + # Link lower bracket winners to next lower-round matches + for i in range(len(lower_rounds) - 1): + for idx, match in enumerate(lower_rounds[i]): + next_match = lower_rounds[i + 1][idx // 2] + match._next_match_win_id = next_match.match_id + + # Final match + final_match = Match( + match_id=match_id_counter, + tournament_id=self._id, + round_number=max(num_rounds_upper, num_rounds_lower) + 1, + bracket=Bracket.FINAL, + best_of=best_of, + status=MatchStatus.WAITING, + next_match_win_lose_ids=(None, None), + match_has_ended_callback=self.match_has_ended_callback + ) + match_id_counter += 1 + + # Last upper winner and last lower winner feed into final + if upper_rounds: + upper_last = upper_rounds[-1][0] + upper_last._next_match_win_id = final_match.match_id + if lower_rounds: + lower_last = lower_rounds[-1][0] + lower_last._next_match_win_id = final_match.match_id + + # Flatten all matches + self._matches = [m for round_matches in upper_rounds + lower_rounds for m in round_matches] + [final_match] + + # Assign participants to first upper round + participant_iter = iter(self._participants) + first_upper = upper_rounds[0] + for match in first_upper: + try: + p1 = next(participant_iter) + match.assign_participant(p1.id, 1) + except StopIteration: + continue + try: + p2 = next(participant_iter) + match.assign_participant(p2.id, 2) + except StopIteration: + if not match.is_fully_seeded: # Auto-Bye + for game in match.games: + match.report_game_result(game.number, p1.id, (1, 0)) + match.check_completion() + continue + + if match.is_fully_seeded: + match.status = MatchStatus.PENDING + + else: + raise TournamentError(f"Unknown bracket type: {bracket_type}") + + self._status = TournamentStatus.ONGOING + for match in self._matches: + match.check_completion() + + +def generate_new_tournament(name: str, game_title: GameTitle, format_: TournamentFormat, start_time: datetime, initial_status: TournamentStatus = TournamentStatus.CLOSED) -> Tournament: + id_ = uuid.uuid4().int + return Tournament( + id_, + name, + game_title, + format_, + start_time, + initial_status, + list(), + None, + list() + ) diff --git a/src/ezgg_lan_manager/types/TournamentBase.py b/src/ezgg_lan_manager/types/TournamentBase.py new file mode 100644 index 0000000..5b79e23 --- /dev/null +++ b/src/ezgg_lan_manager/types/TournamentBase.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass +from enum import Enum + + +@dataclass +class GameTitle: + name: str + description: str + web_link: str + + +class TournamentFormat(Enum): + SINGLE_ELIMINATION_BO_1 = 1 + SINGLE_ELIMINATION_BO_3 = 2 + SINGLE_ELIMINATION_BO_5 = 3 + DOUBLE_ELIMINATION_BO_1 = 4 + DOUBLE_ELIMINATION_BO_3 = 5 + DOUBLE_ELIMINATION_BO_5 = 6 + + +class TournamentStatus(Enum): + CLOSED = 1 + OPEN = 2 + COMPLETED = 3 + CANCELED = 4 + INVITE_ONLY = 5 # For Show-matches + ONGOING = 6 + + +class TournamentError(Exception): + def __init__(self, message: str) -> None: + super().__init__(message) + + +class ParticipantType(Enum): + PLAYER = 1 + TEAM = 2 # ToDo: Teams are not yet supported + + +class Bracket(Enum): + UPPER = 1 + LOWER = 2 + FINAL = 3 + + +class MatchStatus(Enum): + WAITING = 1 # Participants incomplete + PENDING = 2 # Match is ready to be played + DELAYED = 3 # Same as PENDING, but with flag for UI + COMPLETED = 4 # Match has been played + CANCELED = 5 # Match got canceled, "bye" for followup \ No newline at end of file diff --git a/testing/unittests/TournamentDomainTests.py b/testing/unittests/TournamentDomainTests.py new file mode 100644 index 0000000..bf9518a --- /dev/null +++ b/testing/unittests/TournamentDomainTests.py @@ -0,0 +1,66 @@ +import unittest +from datetime import datetime + +from src.ezgg_lan_manager.types.TournamentBase import ParticipantType +from src.ezgg_lan_manager.types.Tournament import generate_new_tournament, GameTitle, TournamentFormat, TournamentStatus, TournamentError, Participant, MatchStatus + + +class TournamentDomainTests(unittest.TestCase): + def setUp(self): + # Generic Tournament config + self.name = "Tetris 1vs1" + self.game_title = GameTitle("Tetris99", "Some Description", "https://de.wikipedia.org/wiki/Tetris_99") + self.format_ = TournamentFormat.SINGLE_ELIMINATION_BO_3 + self.start_time = datetime(year=2100, month=6, day=23, hour=16, minute=30, second=0) + self.initial_status = TournamentStatus.CLOSED + + # Generic Participants + self.participant_a = Participant(1, "CoolUserName", ParticipantType.PLAYER) + self.participant_b = Participant(2, "CrazyUserName", ParticipantType.PLAYER) + self.participant_c = Participant(3, "FunnyUserName", ParticipantType.PLAYER) + + def test_tournament_without_participants_can_not_be_started(self) -> None: + tournament_under_test = generate_new_tournament(self.name, self.game_title, self.format_, self.start_time, self.initial_status) + with self.assertRaises(TournamentError): + tournament_under_test.start() + + def test_adding_the_same_participant_twice_leads_to_exception(self) -> None: + tournament_under_test = generate_new_tournament(self.name, self.game_title, self.format_, self.start_time, self.initial_status) + tournament_under_test.add_participant(self.participant_a) + with self.assertRaises(TournamentError): + tournament_under_test.add_participant(self.participant_a) + + def test_single_elimination_bo3_tournament_gets_generated_correctly(self) -> None: + tournament_under_test = generate_new_tournament(self.name, self.game_title, self.format_, self.start_time, self.initial_status) + + tournament_under_test.add_participant(self.participant_a) + tournament_under_test.add_participant(self.participant_b) + tournament_under_test.add_participant(self.participant_c) + tournament_under_test.start() + + # Assert Tournament was switched to ONGOING + self.assertEqual(TournamentStatus.ONGOING, tournament_under_test.status) + + matches_in_tournament = sorted(tournament_under_test.matches, key=lambda m: m.match_id) + + # First match + fm = matches_in_tournament[0] + self.assertEqual(fm.status, MatchStatus.PENDING) + self.assertEqual(fm.participants[0].participant_id, self.participant_a.id) + self.assertEqual(fm.participants[0].slot_number, 1) + self.assertEqual(fm.participants[1].participant_id, self.participant_b.id) + self.assertEqual(fm.participants[1].slot_number, 2) + + # Second match (Bye) + sm = matches_in_tournament[1] + self.assertEqual(sm.status, MatchStatus.COMPLETED) + self.assertEqual(sm.participants[0].participant_id, self.participant_c.id) + self.assertEqual(sm.participants[0].slot_number, 1) + self.assertEqual(sm.participants[0].participant_id, sm.winner) + + # Third match (Final) + sm = matches_in_tournament[2] + self.assertEqual(sm.status, MatchStatus.WAITING) + self.assertEqual(sm.participants[0].participant_id, self.participant_c.id) + self.assertEqual(sm.participants[0].slot_number, 1) + self.assertIsNone(sm.winner) \ No newline at end of file