Release 0.2.0 #34

Merged
Typhus merged 3 commits from main into prod 2026-02-03 23:04:39 +00:00
31 changed files with 1720 additions and 32 deletions

View File

@ -1 +1 @@
0.1.0
0.2.0

Binary file not shown.

144
sql/tournament_patch.sql Normal file
View File

@ -0,0 +1,144 @@
-- Apply this patch after using create_database.sql to extend the schema to support tournaments from version 0.2.0
-- WARNING: Executing this on a post 0.2.0 database will delete all data related to tournaments !!!
DROP TABLE IF EXISTS `game_titles`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `game_titles` (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL,
web_link VARCHAR(512) NOT NULL,
image_name VARCHAR(255) NOT NULL,
UNIQUE KEY uq_game_title_name (name)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;
DROP TABLE IF EXISTS `tournaments`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `tournaments` (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL,
game_title_id INT NOT NULL,
format VARCHAR(20) NOT NULL, -- SE_BO1, DE_BO3, ...
start_time DATETIME NOT NULL,
status VARCHAR(20) NOT NULL, -- OPEN, CLOSED, ONGOING, ...
max_participants INT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_tournament_game
FOREIGN KEY (game_title_id)
REFERENCES game_titles(id)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;
CREATE INDEX idx_tournaments_game_title
ON tournaments(game_title_id);
DROP TABLE IF EXISTS `tournament_participants`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `tournament_participants` (
id INT AUTO_INCREMENT PRIMARY KEY,
tournament_id INT NOT NULL,
user_id INT NOT NULL,
participant_type VARCHAR(10) NOT NULL DEFAULT 'PLAYER',
seed INT NULL,
joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uq_tournament_user (tournament_id, user_id),
CONSTRAINT fk_tp_tournament
FOREIGN KEY (tournament_id)
REFERENCES tournaments(id)
ON DELETE CASCADE,
CONSTRAINT fk_tp_user
FOREIGN KEY (user_id)
REFERENCES users(user_id)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;
CREATE INDEX idx_tp_tournament
ON tournament_participants(tournament_id);
CREATE INDEX idx_tp_user
ON tournament_participants(user_id);
DROP TABLE IF EXISTS `tournament_rounds`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `tournament_rounds` (
id INT AUTO_INCREMENT PRIMARY KEY,
tournament_id INT NOT NULL,
bracket VARCHAR(10) NOT NULL, -- UPPER, LOWER, FINAL
round_index INT NOT NULL,
UNIQUE KEY uq_round (tournament_id, bracket, round_index),
CONSTRAINT fk_round_tournament
FOREIGN KEY (tournament_id)
REFERENCES tournaments(id)
ON DELETE CASCADE
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;
CREATE INDEX idx_rounds_tournament
ON tournament_rounds(tournament_id);
DROP TABLE IF EXISTS `matches`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `matches` (
id INT AUTO_INCREMENT PRIMARY KEY,
tournament_id INT NOT NULL,
round_id INT NOT NULL,
match_index INT NOT NULL,
status VARCHAR(15) NOT NULL, -- WAITING, PENDING, COMPLETED, ...
best_of INT NOT NULL, -- 1, 3, 5
scheduled_time DATETIME NULL,
completed_at DATETIME NULL,
UNIQUE KEY uq_match (round_id, match_index),
CONSTRAINT fk_match_tournament
FOREIGN KEY (tournament_id)
REFERENCES tournaments(id)
ON DELETE CASCADE,
CONSTRAINT fk_match_round
FOREIGN KEY (round_id)
REFERENCES tournament_rounds(id)
ON DELETE CASCADE
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;
CREATE INDEX idx_matches_tournament
ON matches(tournament_id);
CREATE INDEX idx_matches_round
ON matches(round_id);
DROP TABLE IF EXISTS `match_participants`;
/*!40101 SET @saved_cs_client = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `match_participants` (
match_id INT NOT NULL,
participant_id INT NOT NULL,
score INT NULL,
is_winner TINYINT(1) NULL,
PRIMARY KEY (match_id, participant_id),
CONSTRAINT fk_mp_match
FOREIGN KEY (match_id)
REFERENCES matches(id)
ON DELETE CASCADE,
CONSTRAINT fk_mp_participant
FOREIGN KEY (participant_id)
REFERENCES tournament_participants(id)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */;

View File

@ -30,7 +30,7 @@ if __name__ == "__main__":
corner_radius_large=0,
font=Font(from_root("src/ezgg_lan_manager/assets/fonts/joystix.otf"))
)
default_attachments = [LocalData()]
default_attachments: list = [LocalData()]
default_attachments.extend(init_services())
lan_info = default_attachments[3].get_lan_info()
@ -161,6 +161,16 @@ if __name__ == "__main__":
name="DbErrorPage",
url_segment="db-error",
build=pages.DbErrorPage,
),
ComponentPage(
name="TournamentDetailsPage",
url_segment="tournament",
build=pages.TournamentDetailsPage,
),
ComponentPage(
name="TournamentRulesPage",
url_segment="tournament-rules",
build=pages.TournamentRulesPage,
)
],
theme=theme,

View File

@ -13,11 +13,12 @@ from src.ezgg_lan_manager.services.NewsService import NewsService
from src.ezgg_lan_manager.services.ReceiptPrintingService import ReceiptPrintingService
from src.ezgg_lan_manager.services.SeatingService import SeatingService
from src.ezgg_lan_manager.services.TicketingService import TicketingService
from src.ezgg_lan_manager.services.TournamentService import TournamentService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types import *
# Inits services in the correct order
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService]:
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService, TournamentService]:
logging.basicConfig(level=logging.DEBUG)
configuration_service = ConfigurationService(from_root("config.toml"))
db_service = DatabaseService(configuration_service.get_database_configuration())
@ -30,6 +31,7 @@ def init_services() -> tuple[AccountingService, CateringService, ConfigurationSe
receipt_printing_service = ReceiptPrintingService(seating_service, configuration_service.get_receipt_printing_configuration(), configuration_service.DEV_MODE_ACTIVE)
catering_service = CateringService(db_service, accounting_service, user_service, receipt_printing_service)
local_data_service = LocalDataService()
tournament_service = TournamentService(db_service, user_service)
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service, tournament_service

Binary file not shown.

After

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 23 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 14 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 6.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 65 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 60 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 17 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 44 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 9.8 KiB

View File

@ -0,0 +1,35 @@
from typing import Optional
from rio import Component, Row, Text, TextStyle, Color
class TournamentDetailsInfoRow(Component):
key: str
value: str
key_color: Optional[Color] = None
value_color: Optional[Color] = None
def build(self) -> Component:
return Row(
Text(
text=self.key,
style=TextStyle(
fill=self.key_color if self.key_color is not None else self.session.theme.background_color,
font_size=1
),
margin_bottom=0.5,
align_x=0
),
Text(
text=self.value,
style=TextStyle(
fill=self.value_color if self.value_color is not None else self.session.theme.background_color,
font_size=1
),
margin_bottom=0.5,
align_x=1
),
margin_left=4,
margin_right=4
)

View File

@ -0,0 +1,60 @@
from typing import Literal, Callable
from rio import Component, PointerEventListener, Rectangle, Image, Text, Tooltip, TextStyle, Color, Icon, Row, PointerEvent
from from_root import from_root
from src.ezgg_lan_manager.types.TournamentBase import TournamentStatus
class TournamentPageRow(Component):
tournament_id: int
tournament_name: str
game_image_name: str
current_participants: int
max_participants: int
tournament_status: TournamentStatus
clicked_cb: Callable
def handle_click(self, _: PointerEvent) -> None:
self.clicked_cb(self.tournament_id)
def determine_tournament_status_icon_color_and_text(self) -> tuple[str, Literal["success", "warning", "danger"], str]:
if self.tournament_status == TournamentStatus.OPEN:
return "material/lock_open", "success", "Anmeldung geöffnet"
elif self.tournament_status == TournamentStatus.CLOSED:
return "material/lock", "danger", "Anmeldung geschlossen"
elif self.tournament_status == TournamentStatus.ONGOING:
return "material/autoplay", "warning", "Turnier läuft"
elif self.tournament_status == TournamentStatus.COMPLETED:
return "material/check_circle", "success", "Turnier beendet"
elif self.tournament_status == TournamentStatus.CANCELED:
return "material/cancel", "danger", "Turnier abgesagt"
elif self.tournament_status == TournamentStatus.INVITE_ONLY:
return "material/person_cancel", "warning", "Teilnahme nur per Einladung"
else:
raise RuntimeError(f"Unknown tournament status: {str(self.tournament_status)}")
def build(self) -> Component:
icon_name, color, text = self.determine_tournament_status_icon_color_and_text()
return PointerEventListener(
content=Rectangle(
content=Row(
Image(image=from_root(f"src/ezgg_lan_manager/assets/img/games/{self.game_image_name}")),
Text(self.tournament_name, style=TextStyle(fill=self.session.theme.background_color, font_size=1)),
Text(f"{self.current_participants}/{self.max_participants}", style=TextStyle(fill=self.session.theme.background_color, font_size=1), justify="right", margin_right=0.5),
Tooltip(anchor=Icon(icon_name, min_width=1, min_height=1, fill=color), position="top",
tip=Text(text, style=TextStyle(fill=self.session.theme.background_color, font_size=0.7))),
proportions=[1, 4, 1, 1],
margin=.5
),
fill=self.session.theme.hud_color,
margin=1,
margin_bottom=0,
stroke_color=Color.TRANSPARENT,
stroke_width=0.2,
hover_stroke_color=self.session.theme.background_color,
cursor="pointer"
),
on_press=self.handle_click
)

View File

@ -1,20 +1,85 @@
import logging
from datetime import datetime
from typing import Optional
from rio import Column, Component, event, TextStyle, Text, Spacer
from from_root import from_root
from rio import Column, Component, event, TextStyle, Text, Spacer, Row, Image, Tooltip, IconButton, Popup, Rectangle, Dropdown, ThemeContextSwitcher, Button
from src.ezgg_lan_manager import ConfigurationService
from src.ezgg_lan_manager import ConfigurationService, TournamentService, UserService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.types.DateUtil import weekday_to_display_text
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import TournamentStatus
logger = logging.getLogger(__name__.split(".")[-1])
class ManageTournamentsPage(Component):
tournaments: list[Tournament] = []
remove_participant_popup_open: bool = False
cancel_options: dict[str, Optional[Participant]] = {"": None}
tournament_id_selected_for_participant_removal: Optional[int] = None
participant_selected_for_removal: Optional[Participant] = None
@event.on_populate
async def on_populate(self) -> None:
self.tournaments = await self.session[TournamentService].get_tournaments()
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Turnier Verwaltung")
async def on_start_pressed(self, tournament_id: int) -> None:
logger.info(f"Starting tournament with ID {tournament_id}")
await self.session[TournamentService].start_tournament(tournament_id)
async def on_cancel_pressed(self, tournament_id: int) -> None:
logger.info(f"Canceling tournament with ID {tournament_id}")
await self.session[TournamentService].cancel_tournament(tournament_id)
async def on_remove_participant_pressed(self, tournament_id: int) -> None:
tournament = await self.session[TournamentService].get_tournament_by_id(tournament_id)
if tournament is None:
return
users = await self.session[UserService].get_all_users()
try:
self.cancel_options = {next(filter(lambda u: u.user_id == p.id, users)).user_name: p for p in tournament.participants}
except StopIteration as e:
logger.error(f"Error trying to find user for participant: {e}")
self.tournament_id_selected_for_participant_removal = tournament_id
self.remove_participant_popup_open = True
async def on_remove_participant_confirm_pressed(self) -> None:
if self.participant_selected_for_removal is not None and self.tournament_id_selected_for_participant_removal is not None:
logger.info(f"Removing participant with ID {self.participant_selected_for_removal.id} from tournament with ID {self.tournament_id_selected_for_participant_removal}")
await self.session[TournamentService].unregister_user_from_tournament(self.participant_selected_for_removal.id, self.tournament_id_selected_for_participant_removal)
await self.on_remove_participant_cancel_pressed()
async def on_remove_participant_cancel_pressed(self) -> None:
self.tournament_id_selected_for_participant_removal = None
self.participant_selected_for_removal = None
self.remove_participant_popup_open = False
def build(self) -> Component:
tournament_rows = []
for tournament in self.tournaments:
start_time_color = self.session.theme.background_color
if tournament.start_time < datetime.now() and tournament.status == TournamentStatus.OPEN:
start_time_color = self.session.theme.warning_color
tournament_rows.append(
Row(
Image(image=from_root(f"src/ezgg_lan_manager/assets/img/games/{tournament.game_title.image_name}"), min_width=1.5, margin_right=1),
Text(tournament.name, style=TextStyle(fill=self.session.theme.background_color, font_size=0.8), justify="left", margin_right=1.5),
Text(f"{weekday_to_display_text(tournament.start_time.weekday())[:2]}.{tournament.start_time.strftime('%H:%M')} Uhr", style=TextStyle(fill=start_time_color, font_size=0.8), justify="left", margin_right=1),
Spacer(),
Tooltip(anchor=IconButton("material/play_arrow", min_size=2, margin_right=0.5, on_press=lambda: self.on_start_pressed(tournament.id)), tip="Starten"),
Tooltip(anchor=IconButton("material/cancel_schedule_send", min_size=2, margin_right=0.5, on_press=lambda: self.on_cancel_pressed(tournament.id)), tip="Absagen"),
Tooltip(anchor=IconButton("material/person_cancel", min_size=2, on_press=lambda: self.on_remove_participant_pressed(tournament.id)), tip="Spieler entfernen"),
margin=1
)
)
return Column(
MainViewContentBox(
Popup(
anchor=MainViewContentBox(
Column(
Text(
text="Turnier Verwaltung",
@ -25,8 +90,28 @@ class ManageTournamentsPage(Component):
margin_top=2,
margin_bottom=2,
align_x=0.5
),
*tournament_rows
)
)
),
content=Rectangle(
content=Row(
ThemeContextSwitcher(
content=Dropdown(options=self.cancel_options, min_width=20, selected_value=self.bind().participant_selected_for_removal), color=self.session.theme.hud_color
),
Button(content="REMOVE", shape="rectangle", grow_x=False, on_press=self.on_remove_participant_confirm_pressed),
Button(content="CANCEL", shape="rectangle", grow_x=False, on_press=self.on_remove_participant_cancel_pressed),
margin=0.5
),
min_width=30,
min_height=4,
fill=self.session.theme.primary_color,
margin_top=3.5,
stroke_width=0.3,
stroke_color=self.session.theme.neutral_color,
),
is_open=self.remove_participant_popup_open,
color="none"
),
Spacer()
)

View File

@ -0,0 +1,252 @@
from typing import Optional, Union, Literal
from from_root import from_root
from rio import Column, Component, event, TextStyle, Text, Row, Image, Spacer, ProgressCircle, Button, Checkbox, ThemeContextSwitcher, Link, Revealer, PointerEventListener, \
PointerEvent, Rectangle, Color
from src.ezgg_lan_manager import ConfigurationService, TournamentService, UserService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.components.TournamentDetailsInfoRow import TournamentDetailsInfoRow
from src.ezgg_lan_manager.types.DateUtil import weekday_to_display_text
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.SessionStorage import SessionStorage
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import TournamentStatus, tournament_status_to_display_text, tournament_format_to_display_texts
from src.ezgg_lan_manager.types.User import User
class TournamentDetailsPage(Component):
tournament: Optional[Union[Tournament, str]] = None
rules_accepted: bool = False
user: Optional[User] = None
loading: bool = False
participant_revealer_open: bool = False
current_tournament_user_list: list[User] = [] # ToDo: Integrate Teams
# State for message above register button
message: str = ""
is_success: bool = False
@event.on_populate
async def on_populate(self) -> None:
try:
tournament_id = int(self.session.active_page_url.query_string.split("id=")[-1])
except (ValueError, AttributeError, TypeError):
tournament_id = None
if tournament_id is not None:
self.tournament = await self.session[TournamentService].get_tournament_by_id(tournament_id)
if self.tournament is not None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - {self.tournament.name}")
self.current_tournament_user_list = await self.session[TournamentService].get_users_from_participant_list(self.tournament.participants)
else:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Turniere")
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
self.loading_done()
def open_close_participant_revealer(self, _: PointerEvent) -> None:
self.participant_revealer_open = not self.participant_revealer_open
async def register_pressed(self) -> None:
self.loading = True
if not self.user:
return
try:
await self.session[TournamentService].register_user_for_tournament(self.user.user_id, self.tournament.id)
self.is_success = True
self.message = f"Erfolgreich angemeldet!"
except Exception as e:
self.is_success = False
self.message = f"Fehler: {e}"
self.loading = False
await self.on_populate()
async def unregister_pressed(self) -> None:
self.loading = True
if not self.user:
return
try:
await self.session[TournamentService].unregister_user_from_tournament(self.user.user_id, self.tournament.id)
self.is_success = True
self.message = f"Erfolgreich abgemeldet!"
except Exception as e:
self.is_success = False
self.message = f"Fehler: {e}"
self.loading = False
await self.on_populate()
async def tree_button_clicked(self) -> None:
pass # ToDo: Implement tournament tree view
def loading_done(self) -> None:
if self.tournament is None:
self.tournament = "Turnier konnte nicht gefunden werden"
self.session[SessionStorage].subscribe_to_logged_in_or_out_event(str(self.__class__), self.on_populate)
def build(self) -> Component:
if self.tournament is None:
content = Column(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=0,
margin_bottom=0
),
min_height=10
)
elif isinstance(self.tournament, str):
content = Row(
Text(
text=self.tournament,
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1
),
margin_top=2,
margin_bottom=2,
align_x=0.5
)
)
else:
tournament_status_color = self.session.theme.background_color
tree_button = Spacer(grow_x=False, grow_y=False)
if self.tournament.status == TournamentStatus.OPEN:
tournament_status_color = self.session.theme.success_color
elif self.tournament.status == TournamentStatus.CLOSED:
tournament_status_color = self.session.theme.danger_color
elif self.tournament.status == TournamentStatus.ONGOING or self.tournament.status == TournamentStatus.COMPLETED:
tournament_status_color = self.session.theme.warning_color
tree_button = Button(
content="Turnierbaum anzeigen",
shape="rectangle",
style="minor",
color="hud",
margin_left=4,
margin_right=4,
margin_top=1,
on_press=self.tree_button_clicked
)
# ToDo: Integrate Teams logic
ids_of_participants = [p.id for p in self.tournament.participants]
color_key: Literal["hud", "danger"] = "hud"
on_press_function = self.register_pressed
if self.user and self.user.user_id in ids_of_participants: # User already registered for tournament
button_text = "Abmelden"
button_sensitive_hook = True # User has already accepted the rules previously
color_key = "danger"
on_press_function = self.unregister_pressed
elif self.user and self.user.user_id not in ids_of_participants:
button_text = "Anmelden"
button_sensitive_hook = self.rules_accepted
else:
# This should NEVER happen
button_text = "Anmelden"
button_sensitive_hook = False
if self.tournament.status != TournamentStatus.OPEN or self.tournament.is_full:
button_sensitive_hook = False # Override button controls if tournament is not open anymore or full
if self.user:
accept_rules_row = Row(
ThemeContextSwitcher(content=Checkbox(is_on=self.bind().rules_accepted, margin_left=4), color=self.session.theme.hud_color),
Text("Ich akzeptiere die ", margin_left=1, style=TextStyle(fill=self.session.theme.background_color, font_size=0.8), overflow="nowrap", justify="right"),
Link(Text("Turnierregeln", margin_right=4, style=TextStyle(fill=self.session.theme.background_color, font_size=0.8, italic=True), overflow="nowrap", justify="left"), "./tournament-rules", open_in_new_tab=True)
)
button = Button(
content=button_text,
shape="rectangle",
style="major",
color=color_key,
margin_left=2,
margin_right=2,
is_sensitive=button_sensitive_hook,
on_press=on_press_function,
is_loading=self.loading
)
else:
# No UI here if user not logged in
accept_rules_row, button = Spacer(), Spacer()
content = Column(
Row(
Image(image=from_root(f"src/ezgg_lan_manager/assets/img/games/{self.tournament.game_title.image_name}"), margin_right=1),
Text(
text=self.tournament.name,
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=1,
margin_bottom=1,
align_x=0.5
),
margin_right=6,
margin_left=6
),
Spacer(min_height=1),
TournamentDetailsInfoRow("Status", tournament_status_to_display_text(self.tournament.status), value_color=tournament_status_color),
TournamentDetailsInfoRow("Startzeit", f"{weekday_to_display_text(self.tournament.start_time.weekday())}, {self.tournament.start_time.strftime('%H:%M')} Uhr"),
TournamentDetailsInfoRow("Format", tournament_format_to_display_texts(self.tournament.format)[0]),
TournamentDetailsInfoRow("Best of", tournament_format_to_display_texts(self.tournament.format)[1]),
PointerEventListener(
content=Rectangle(
content=TournamentDetailsInfoRow(
"Teilnehmer ▴" if self.participant_revealer_open else "Teilnehmer ▾",
f"{len(self.tournament.participants)} / {self.tournament.max_participants}",
value_color=self.session.theme.danger_color if self.tournament.is_full else self.session.theme.background_color,
key_color=self.session.theme.secondary_color
),
fill=Color.TRANSPARENT,
cursor="pointer"
),
on_press=self.open_close_participant_revealer
),
Revealer(
header=None,
content=Text(
"\n".join([u.user_name for u in self.current_tournament_user_list]), # ToDo: Integrate Teams
style=TextStyle(fill=self.session.theme.background_color)
),
is_open=self.participant_revealer_open,
margin_left=4,
margin_right=4
),
tree_button,
Row(
Text(
text="Info",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=1,
margin_bottom=1,
align_x=0.5
)
),
# FixMe: Use rio.Markdown with correct TextStyle instead to allow basic text formatting from DB-side.
Text(self.tournament.description, margin_left=2, margin_right=2, style=TextStyle(fill=self.session.theme.background_color, font_size=1), overflow="wrap"),
Spacer(min_height=2),
accept_rules_row,
Spacer(min_height=0.5),
Text(self.message, margin_left=2, margin_right=2, style=TextStyle(fill=self.session.theme.success_color if self.is_success else self.session.theme.danger_color, font_size=1), overflow="wrap", justify="center"),
Spacer(min_height=0.5),
button
)
return Column(
MainViewContentBox(
Column(
Spacer(min_height=1),
content,
Spacer(min_height=1)
)
),
align_y=0
)

View File

@ -0,0 +1,52 @@
from rio import Column, Component, event, TextStyle, Text, Spacer
from src.ezgg_lan_manager import ConfigurationService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
RULES: list[str] = [
"Den Anweisungen der Turnierleitung ist stets Folge zu leisten.",
"Teilnehmer müssen aktiv dafür sorgen, dass Spiele ohne Verzögerungen stattfinden.",
"Unvollständige Teams werden ggf. zum Turnierstart entfernt.",
"Verzögerungen und Ausfälle sind er Turnierleitung sofort zu melden.",
"Jeder Spieler erstellt Screenshots am Rundenende zur Ergebnisdokumentation.",
"Der Verlierer trägt das Ergebnis ein, der Gewinner überprüft es.",
"Bei fehlendem oder falschem Ergebnis, ist sofort die Turnierorganisation zu informieren.",
"Von 02:0011:00 Uhr besteht keine Spielpflicht",
"Täuschung, Falschangaben sowie Bugusing und Cheaten führen zur sofortigen Disqualifikation."
]
class TournamentRulesPage(Component):
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Turnierregeln")
def build(self) -> Component:
return Column(
MainViewContentBox(
Column(
Text(
text="Turnierregeln",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=2,
margin_bottom=2,
align_x=0.5
),
*[Text(
f"{idx + 1}. {rule}",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=0.9
),
margin_bottom=0.8,
margin_left=1,
margin_right=1,
overflow="wrap"
) for idx, rule in enumerate(RULES)],
Spacer(min_height=1)
)
),
Spacer(grow_y=True)
)

View File

@ -1,15 +1,50 @@
from rio import Column, Component, event, TextStyle, Text
from rio import Column, Component, event, TextStyle, Text, Spacer, ProgressCircle
from src.ezgg_lan_manager import ConfigurationService
from src.ezgg_lan_manager import ConfigurationService, TournamentService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.components.TournamentPageRow import TournamentPageRow
from src.ezgg_lan_manager.types.Tournament import Tournament
class TournamentsPage(Component):
tournament_data: list[Tournament] = []
@event.on_populate
async def on_populate(self) -> None:
self.tournament_data = await self.session[TournamentService].get_tournaments()
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Turniere")
def tournament_clicked(self, tournament_id: int) -> None:
self.session.navigate_to(f"tournament?id={tournament_id}")
def build(self) -> Component:
tournament_page_rows = []
for tournament in self.tournament_data:
tournament_page_rows.append(
TournamentPageRow(
tournament.id,
tournament.name,
tournament.game_title.image_name,
len(tournament.participants),
tournament.max_participants,
tournament.status,
self.tournament_clicked
)
)
if len(self.tournament_data) == 0:
content = [Column(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=0,
margin_bottom=0
),
min_height=10
)]
else:
content = tournament_page_rows
return Column(
MainViewContentBox(
Column(
@ -20,18 +55,11 @@ class TournamentsPage(Component):
font_size=1.2
),
margin_top=2,
margin_bottom=0,
margin_bottom=2,
align_x=0.5
),
Text(
text="Aktuell ist noch kein Turnierplan hinterlegt.",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=0.9
),
margin=1,
overflow="wrap"
)
*content,
Spacer(min_height=1)
)
),
align_y=0

View File

@ -20,3 +20,5 @@ from .ManageUsersPage import ManageUsersPage
from .ManageCateringPage import ManageCateringPage
from .ManageTournamentsPage import ManageTournamentsPage
from .OverviewPage import OverviewPage
from .TournamentDetailsPage import TournamentDetailsPage
from .TournamentRulesPage import TournamentRulesPage

View File

@ -1,6 +1,7 @@
import logging
from datetime import date, datetime
from pprint import pprint
from typing import Optional
from decimal import Decimal
@ -11,8 +12,11 @@ from src.ezgg_lan_manager.types.CateringMenuItem import CateringMenuItem, Cateri
from src.ezgg_lan_manager.types.CateringOrder import CateringMenuItemsWithAmount, CateringOrderStatus
from src.ezgg_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
from src.ezgg_lan_manager.types.News import News
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.Seat import Seat
from src.ezgg_lan_manager.types.Ticket import Ticket
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import GameTitle, TournamentFormat, TournamentStatus, ParticipantType
from src.ezgg_lan_manager.types.Transaction import Transaction
from src.ezgg_lan_manager.types.User import User
@ -81,6 +85,54 @@ class DatabaseService:
last_updated_at=data[11]
)
@staticmethod
def _parse_tournament_format(format_as_string: str) -> TournamentFormat:
if format_as_string == "SE_BO_1":
return TournamentFormat.SINGLE_ELIMINATION_BO_1
elif format_as_string == "SE_BO_3":
return TournamentFormat.SINGLE_ELIMINATION_BO_3
elif format_as_string == "SE_BO_5":
return TournamentFormat.SINGLE_ELIMINATION_BO_5
elif format_as_string == "DE_BO_1":
return TournamentFormat.DOUBLE_ELIMINATION_BO_1
elif format_as_string == "DE_BO_3":
return TournamentFormat.DOUBLE_ELIMINATION_BO_3
elif format_as_string == "DE_BO_5":
return TournamentFormat.DOUBLE_ELIMINATION_BO_5
elif format_as_string == "FFA":
return TournamentFormat.FFA
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown TournamentFormat: {format_as_string}")
@staticmethod
def _parse_tournament_status(status_as_string: str) -> TournamentStatus:
if status_as_string == "CLOSED":
return TournamentStatus.CLOSED
elif status_as_string == "OPEN":
return TournamentStatus.OPEN
elif status_as_string == "COMPLETED":
return TournamentStatus.COMPLETED
elif status_as_string == "CANCELED":
return TournamentStatus.CANCELED
elif status_as_string == "INVITE_ONLY":
return TournamentStatus.INVITE_ONLY
elif status_as_string == "ONGOING":
return TournamentStatus.ONGOING
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown TournamentStatus: {status_as_string}")
@staticmethod
def _parse_participant_type(participant_type_as_string: str) -> ParticipantType:
if participant_type_as_string == "PLAYER":
return ParticipantType.PLAYER
elif participant_type_as_string == "TEAM":
return ParticipantType.TEAM
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown ParticipantType: {participant_type_as_string}")
async def get_user_by_name(self, user_name: str) -> Optional[User]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
@ -787,3 +839,131 @@ class DatabaseService:
return await self.remove_profile_picture(user_id)
except Exception as e:
logger.warning(f"Error deleting user profile picture: {e}")
async def get_all_tournaments(self) -> list[Tournament]:
logger.info(f"Polling Tournaments...")
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.DictCursor) as cursor:
try:
await cursor.execute(
"""
SELECT
/* =======================
Tournament
======================= */
t.id AS tournament_id,
t.name AS tournament_name,
t.description AS tournament_description,
t.format AS tournament_format,
t.start_time,
t.status AS tournament_status,
t.max_participants,
t.created_at,
/* =======================
Game Title
======================= */
gt.id AS game_title_id,
gt.name AS game_title_name,
gt.description AS game_title_description,
gt.web_link AS game_title_web_link,
gt.image_name AS game_title_image_name,
/* =======================
Tournament Participant
======================= */
tp.id AS participant_id,
tp.user_id,
tp.participant_type,
tp.seed,
tp.joined_at
FROM tournaments t
JOIN game_titles gt
ON gt.id = t.game_title_id
LEFT JOIN tournament_participants tp
ON tp.tournament_id = t.id
ORDER BY
t.id,
tp.seed IS NULL,
tp.seed;
"""
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.get_all_tournaments()
except Exception as e:
logger.warning(f"Error getting tournaments: {e}")
tournaments = []
current_tournament: Optional[Tournament] = None
for row in await cursor.fetchall():
if current_tournament is None or current_tournament.id != row["tournament_id"]:
if current_tournament is not None:
tournaments.append(current_tournament)
current_tournament = Tournament(
id_=row["tournament_id"],
name=row["tournament_name"],
description=row["tournament_description"],
game_title=GameTitle(
name=row["game_title_name"],
description=row["game_title_description"],
web_link=row["game_title_web_link"],
image_name=row["game_title_image_name"]
),
format_=self._parse_tournament_format(row["tournament_format"]),
start_time=row["start_time"],
status=self._parse_tournament_status(row["tournament_status"]),
participants=[Participant(id_=row["user_id"], participant_type=self._parse_participant_type(row["participant_type"]))] if row["user_id"] is not None else [],
matches=None, # ToDo: Implement
rounds=[], # ToDo: Implement
max_participants=row["max_participants"]
)
else:
current_tournament.add_participant(
Participant(id_=row["user_id"], participant_type=self._parse_participant_type(row["participant_type"]))
)
else:
tournaments.append(current_tournament)
return tournaments
async def add_participant_to_tournament(self, participant: Participant, tournament: Tournament) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO tournament_participants (tournament_id, user_id, participant_type) VALUES (%s, %s, %s);",
(tournament.id, participant.id, participant.participant_type.name)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.add_participant_to_tournament(participant, tournament)
except Exception as e:
logger.warning(f"Error adding participant to tournament: {e}")
async def remove_participant_from_tournament(self, participant: Participant, tournament: Tournament) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"DELETE FROM tournament_participants WHERE (tournament_id = %s AND user_id = %s);",
(tournament.id, participant.id)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.remove_participant_from_tournament(participant, tournament)
except Exception as e:
logger.warning(f"Error removing participant from tournament: {e}")

View File

@ -0,0 +1,72 @@
from typing import Optional
from src.ezgg_lan_manager.services.DatabaseService import DatabaseService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import ParticipantType, TournamentError
from src.ezgg_lan_manager.types.User import User
class TournamentService:
def __init__(self, db_service: DatabaseService, user_service: UserService) -> None:
self._db_service = db_service
self._user_service = user_service
# Crude cache mechanism. If performance suffers, maybe implement a queue with Single-Owner-Pattern or a Lock
self._cache: dict[int, Tournament] = {}
self._cache_dirty: bool = True # Setting this flag invokes cache update on next read
async def _update_cache(self) -> None:
tournaments = await self._db_service.get_all_tournaments()
for tournament in tournaments:
self._cache[tournament.id] = tournament
self._cache_dirty = False
async def register_user_for_tournament(self, user_id: int, tournament_id: int) -> None:
tournament = await self.get_tournament_by_id(tournament_id)
if not tournament:
raise TournamentError(f"No tournament with ID {tournament_id} was found")
participant = Participant(id_=user_id, participant_type=ParticipantType.PLAYER)
tournament.add_participant(participant)
await self._db_service.add_participant_to_tournament(participant, tournament)
self._cache_dirty = True
async def unregister_user_from_tournament(self, user_id: int, tournament_id: int) -> None:
tournament = await self.get_tournament_by_id(tournament_id)
if not tournament:
raise TournamentError(f"No tournament with ID {tournament_id} was found")
participant = next(filter(lambda p: p.id == user_id, tournament.participants), None)
if participant is not None:
tournament.remove_participant(participant)
await self._db_service.remove_participant_from_tournament(participant, tournament)
self._cache_dirty = True
async def get_tournaments(self) -> list[Tournament]:
if self._cache_dirty:
await self._update_cache()
return list(self._cache.values())
async def get_tournament_by_id(self, tournament_id: int) -> Optional[Tournament]:
if self._cache_dirty:
await self._update_cache()
return self._cache.get(tournament_id, None)
async def get_users_from_participant_list(self, participants: list[Participant]) -> list[User]:
all_users = await self._db_service.get_all_users()
participant_ids = [p.id for p in participants]
return list(filter(lambda u: u.user_id in participant_ids, all_users))
async def start_tournament(self, tournament_id: int):
tournament = await self.get_tournament_by_id(tournament_id)
if tournament:
tournament.start()
# ToDo: Write matches/round to database
self._cache_dirty = True
async def cancel_tournament(self, tournament_id: int):
tournament = await self.get_tournament_by_id(tournament_id)
if tournament:
tournament.cancel()
# ToDo: Update to database
self._cache_dirty = True

View File

@ -0,0 +1,15 @@
def weekday_to_display_text(weekday: int) -> str:
if weekday == 0:
return "Montag"
elif weekday == 1:
return "Dienstag"
elif weekday == 2:
return "Mittwoch"
elif weekday == 3:
return "Donnerstag"
elif weekday == 4:
return "Freitag"
elif weekday == 5:
return "Samstag"
else:
return "Sonntag"

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,160 @@
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, 1, 2]) -> None:
self._participant_id = participant_id
if slot_number not in (-1, 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, 1, 2]) -> None:
if slot == -1:
raise TournamentError("Normal match does not support slot -1")
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}>")
class FFAMatch(Match):
"""
Specialized match that supports infinite participants
"""
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:
super().__init__(match_id, tournament_id, round_number, bracket, best_of, status, next_match_win_lose_ids, match_has_ended_callback)
@property
def is_fully_seeded(self) -> bool:
return len(self._participants) > 1
def assign_participant(self, participant_id: int, slot: Literal[-1, 1, 2]) -> None:
if slot != -1:
raise TournamentError("FFAMatch does not support slot 1 and 2")
new_participant = MatchParticipant(participant_id, slot)
self._participants.append(new_participant)
def __repr__(self) -> str:
participants = ", ".join(
f"{p.participant_id}" 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,15 @@
from src.ezgg_lan_manager.types.TournamentBase import ParticipantType
class Participant:
def __init__(self, id_: int, participant_type: ParticipantType) -> None:
self._id = id_
self._participant_type = participant_type
@property
def id(self) -> int:
return self._id
@property
def participant_type(self) -> ParticipantType:
return self._participant_type

View File

@ -0,0 +1,368 @@
import uuid
from datetime import datetime
from typing import Optional
from math import ceil, log2
from src.ezgg_lan_manager.types.Match import Match, FFAMatch
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,
description: str,
game_title: GameTitle,
format_: TournamentFormat,
start_time: datetime,
status: TournamentStatus,
participants: list[Participant],
matches: Optional[tuple[Match]],
rounds: list[list[Match]],
max_participants: int) -> None:
self._id = id_
self._name = name
self._description = description
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
self._max_participants = max_participants
@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 []
@property
def max_participants(self) -> int:
return self._max_participants
@property
def description(self) -> str:
return self._description
@property
def is_full(self) -> bool:
return len(self._participants) >= self._max_participants
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 remove_participant(self, participant: Participant) -> None:
if participant.id not in (p.id for p in self._participants):
raise TournamentError(f"Participant with ID {participant.id} not registered for tournament")
# ToDo: Check if tournament already started => correctly resolve matches with now missing participant
self._participants.remove(participant)
def cancel(self):
self.status = TournamentStatus.CANCELED
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"
elif fmt.name.startswith("FFA"):
bracket = "FINAL"
else:
raise TournamentError(f"Unsupported tournament format: {fmt}")
if fmt.name.endswith("_BO_1") or fmt.name.endswith("FFA"):
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 == "FINAL":
rounds: list[list[Match]] = []
round_matches = []
match = FFAMatch(
match_id=match_id_counter,
tournament_id=self._id,
round_number=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
)
for participant in self.participants:
match.assign_participant(participant.id, -1)
round_matches.append(match)
rounds.append(round_matches)
self._matches = [match]
elif 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, description: str, game_title: GameTitle, format_: TournamentFormat, start_time: datetime, max_participants: int, initial_status: TournamentStatus = TournamentStatus.CLOSED) -> Tournament:
id_ = uuid.uuid4().int
return Tournament(
id_,
name,
description,
game_title,
format_,
start_time,
initial_status,
list(),
None,
list(),
max_participants
)

View File

@ -0,0 +1,87 @@
from dataclasses import dataclass
from enum import Enum
@dataclass
class GameTitle:
name: str
description: str
web_link: str
image_name: str # Name of the image in assets/img/games
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
FFA = 7
def tournament_format_to_display_texts(tournament_format: TournamentFormat) -> tuple[str, str]:
""" Returns tuple where idx 0 is SE/DE/FFA string and idx 1 is match count """
if tournament_format == TournamentFormat.SINGLE_ELIMINATION_BO_1:
return "Single Elimination", "1"
elif tournament_format == TournamentFormat.SINGLE_ELIMINATION_BO_3:
return "Single Elimination", "3"
elif tournament_format == TournamentFormat.SINGLE_ELIMINATION_BO_5:
return "Single Elimination", "5"
elif tournament_format == TournamentFormat.DOUBLE_ELIMINATION_BO_1:
return "Double Elimination", "1"
elif tournament_format == TournamentFormat.DOUBLE_ELIMINATION_BO_3:
return "Double Elimination", "3"
elif tournament_format == TournamentFormat.DOUBLE_ELIMINATION_BO_5:
return "Double Elimination", "5"
elif tournament_format == TournamentFormat.FFA:
return "Free for All", "1"
else:
raise RuntimeError(f"Unknown tournament status: {str(tournament_format)}")
class TournamentStatus(Enum):
CLOSED = 1
OPEN = 2
COMPLETED = 3
CANCELED = 4
INVITE_ONLY = 5 # For Show-matches
ONGOING = 6
def tournament_status_to_display_text(tournament_status: TournamentStatus) -> str:
if tournament_status == TournamentStatus.OPEN:
return "Offen"
elif tournament_status == TournamentStatus.CLOSED:
return "Geschlossen"
elif tournament_status == TournamentStatus.ONGOING:
return "Läuft"
elif tournament_status == TournamentStatus.COMPLETED:
return "Abgeschlossen"
elif tournament_status == TournamentStatus.CANCELED:
return "Abgesagt"
elif tournament_status == TournamentStatus.INVITE_ONLY:
return "Invite-only"
else:
raise RuntimeError(f"Unknown tournament status: {str(tournament_status)}")
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,82 @@
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.description = "Just play Tetris, yo"
self.game_title = GameTitle("Tetris99", "Some Description", "https://de.wikipedia.org/wiki/Tetris_99", "tetris.png")
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, ParticipantType.PLAYER)
self.participant_b = Participant(2, ParticipantType.PLAYER)
self.participant_c = Participant(3, ParticipantType.PLAYER)
def test_tournament_without_participants_can_not_be_started(self) -> None:
tournament_under_test = generate_new_tournament(self.name, self.description, self.game_title, self.format_, self.start_time, 32, 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.description, self.game_title, self.format_, self.start_time, 32, 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.description, self.game_title, self.format_, self.start_time, 32, 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)
def test_ffa_tournament_with_15_participants_gets_generated_correctly(self) -> None:
tournament_under_test = generate_new_tournament("Among Us", "It's Among Us", GameTitle("Among Us", "", "", ""), TournamentFormat.FFA, self.start_time, 32, TournamentStatus.OPEN)
for i in range(1, 16):
tournament_under_test.add_participant(Participant(i, ParticipantType.PLAYER))
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)
self.assertEqual(1, len(matches_in_tournament))
self.assertEqual(15, len(matches_in_tournament[0].participants))