add tournament data model

This commit is contained in:
David Rodenkirchen 2026-01-27 18:24:30 +01:00 committed by David Rodenkirchen
parent b505191156
commit ff5d715a4e
7 changed files with 628 additions and 0 deletions

Binary file not shown.

View File

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

View File

@ -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"<Match id={self._match_id} round={self._round_number} bracket={self._bracket.name} "
f"status={self.status.name} participants=[{participants}] winner={self.winner}>")

View File

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

View File

@ -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()
)

View File

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

View File

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