Added seperate bracket/node logic

This commit is contained in:
2026-02-12 16:03:10 +01:00 Verified
parent 7cbb7faca8
commit bb68e326cd
13 changed files with 540 additions and 530 deletions
+251 -259
View File
@@ -1,10 +1,19 @@
# backend/app/logic.py
import math
from datetime import datetime, timedelta
from typing import List, Dict, Any
from typing import Optional
from uuid import uuid4
from .constants import BracketType, MatchSourceType, MatchStatus, TournamentTypes
from .models import Tournament
from sqlalchemy.orm import Session
from . import models
from .constants import (
BracketType,
MatchSourceType,
MatchStatus,
TournamentTypes,
WinnerSide,
)
def get_seeded_positions(num_slots, teams):
@@ -15,327 +24,310 @@ def get_seeded_positions(num_slots, teams):
next_seeds.append(s)
next_seeds.append(2 * len(seeds) + 1 - s)
seeds = next_seeds
return [teams[s - 1] if s <= len(teams) else "BYE" for s in seeds]
return [teams[s - 1] if s <= len(teams) else None for s in seeds]
def generate_structure(
teams: List[str], type: TournamentTypes = TournamentTypes.DOUBLE
) -> List[Dict[str, Any]]:
def generate_bracket_nodes(t: models.Tournament) -> list[dict]:
"""
Generates the skeleton (BracketNodes). Does NOT create Matches.
"""
teams = t.teams
count = len(teams)
if count < 2:
return []
power = math.ceil(math.log2(count)) if count > 0 else 1
size = 2**power
seeded_teams = get_seeded_positions(size, teams)
class Node:
def __init__(self, id, bracket: BracketType, round_n: int):
self.id = str(id)
nodes = []
display_counter = 1
def make_id():
return str(uuid4())
class NodeRef:
def __init__(self, bracket, round_n):
self.id = make_id()
self.bracket = bracket
self.round = round_n
self.p1: str | None = None
self.p2: str | None = None
self.display_num = 0
self.winner_next_match_id: str | None = None
self.loser_next_match_id: str | None = None
self.previous_match_p1_id: str | None = None
self.previous_match_p2_id: str | None = None
self.next_win: Optional["NodeRef"] = None
self.next_loss: Optional["NodeRef"] = None
self.src_p1: Optional["NodeRef"] = None
self.src_p2: Optional["NodeRef"] = None
self.source_p1_type: MatchSourceType | None = None
self.source_p2_type: MatchSourceType | None = None
self.src_p1_type: Optional[MatchSourceType] = None
self.src_p2_type: Optional[MatchSourceType] = None
def to_dict(self):
return {
"id": self.id,
# Pass Enum OBJECTS, not strings. SQLAlchemy handles the rest.
"bracket": self.bracket,
"round": self.round,
"p1_name": self.p1,
"p2_name": self.p2,
"status": MatchStatus.PENDING,
"previous_match_p1_id": self.previous_match_p1_id,
"previous_match_p2_id": self.previous_match_p2_id,
"source_p1_type": self.source_p1_type,
"source_p2_type": self.source_p2_type,
"winner_next_match_id": self.winner_next_match_id,
"loser_next_match_id": self.loser_next_match_id,
"bracket_type": self.bracket,
"round_number": self.round,
"display_number": self.display_num,
"winner_next_node_id": self.next_win.id if self.next_win else None,
"loser_next_node_id": self.next_loss.id if self.next_loss else None,
"source_p1_node_id": self.src_p1.id if self.src_p1 else None,
"source_p2_node_id": self.src_p2.id if self.src_p2 else None,
"source_p1_type": self.src_p1_type,
"source_p2_type": self.src_p2_type,
}
nodes: List[Node] = []
match_counter = 1
def create_node(bracket: BracketType, round_n: int):
nonlocal match_counter
n = Node(match_counter, bracket, round_n)
match_counter += 1
nodes.append(n)
return n
# --- Winners Bracket ---
wb_rounds = power
wb_matches = {r: [] for r in range(1, wb_rounds + 1)}
wb_layers = {r: [] for r in range(1, wb_rounds + 1)}
for r in range(1, wb_rounds + 1):
for _ in range(size // (2**r)):
wb_matches[r].append(create_node(BracketType.WINNERS, r))
n = NodeRef(BracketType.WINNERS, r)
wb_layers[r].append(n)
# Link Winners
for r in range(1, wb_rounds):
for i, m in enumerate(wb_matches[r]):
target = wb_matches[r + 1][i // 2]
m.winner_next_match_id = target.id
for i, node in enumerate(wb_layers[r]):
target = wb_layers[r + 1][i // 2]
node.next_win = target
if i % 2 == 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
target.src_p1 = node
target.src_p1_type = MatchSourceType.WINNER
else:
target.previous_match_p2_id = m.id
target.source_p2_type = MatchSourceType.WINNER
target.src_p2 = node
target.src_p2_type = MatchSourceType.WINNER
for i, m in enumerate(wb_matches[1]):
m.p1 = seeded_teams[i * 2]
m.p2 = seeded_teams[i * 2 + 1]
# --- Losers Bracket ---
if type == TournamentTypes.DOUBLE and size >= 4:
lb_layers = {}
if t.type == TournamentTypes.DOUBLE and size >= 4:
lb_rounds = (wb_rounds - 1) * 2
lb_matches = {r: [] for r in range(1, lb_rounds + 1)}
current_count = size // 4
lb_layers = {r: [] for r in range(1, lb_rounds + 1)}
current_count = size // 4
for r in range(1, lb_rounds + 1):
for _ in range(current_count):
lb_matches[r].append(create_node(BracketType.LOSERS, r))
n = NodeRef(BracketType.LOSERS, r)
lb_layers[r].append(n)
if r % 2 == 0:
current_count //= 2
# Link Losers Internal
for r in range(1, lb_rounds):
for i, m in enumerate(lb_matches[r]):
target = (
lb_matches[r + 1][i] if r % 2 != 0 else lb_matches[r + 1][i // 2]
)
m.winner_next_match_id = target.id
for i, node in enumerate(lb_layers[r]):
target = lb_layers[r + 1][i] if r % 2 != 0 else lb_layers[r + 1][i // 2]
node.next_win = target
if r % 2 != 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
target.src_p1 = node
target.src_p1_type = MatchSourceType.WINNER
else:
if i % 2 == 0:
target.previous_match_p1_id = m.id
target.source_p1_type = MatchSourceType.WINNER
target.src_p1 = node
target.src_p1_type = MatchSourceType.WINNER
else:
target.previous_match_p2_id = m.id
target.source_p2_type = MatchSourceType.WINNER
target.src_p2 = node
target.src_p2_type = MatchSourceType.WINNER
# Link Losers Drop-down
# Link Drop-down (Winners -> Losers)
for r in range(1, wb_rounds):
drop_round = 1 if r == 1 else (r - 1) * 2
wb_layer = wb_matches[r]
lb_layer = lb_matches[drop_round]
wb_layer_nodes = wb_layers[r]
lb_layer_nodes = lb_layers[drop_round]
for i, wb_m in enumerate(wb_layer):
target = (
lb_layer[i // 2]
if r == 1
else (lb_layer[i] if i < len(lb_layer) else lb_layer[-1])
)
slot = "p1" if (r == 1 and i % 2 == 0) else "p2"
wb_m.loser_next_match_id = target.id
if slot == "p1":
target.previous_match_p1_id = wb_m.id
target.source_p1_type = MatchSourceType.LOSER
for i, wb_node in enumerate(wb_layer_nodes):
target = None
if r == 1:
target = lb_layer_nodes[i // 2]
else:
target.previous_match_p2_id = wb_m.id
target.source_p2_type = MatchSourceType.LOSER
if i < len(lb_layer_nodes):
target = lb_layer_nodes[i]
else:
target = lb_layer_nodes[-1]
# Finals Linking
wb_final = wb_matches[wb_rounds][0]
lb_final = lb_matches[lb_rounds][0]
wb_node.next_loss = target
wb_final.loser_next_match_id = lb_final.id
lb_final.previous_match_p2_id = wb_final.id
lb_final.source_p2_type = MatchSourceType.LOSER
final = create_node(BracketType.FINALS, 1)
wb_final.winner_next_match_id = final.id
lb_final.winner_next_match_id = final.id
final.previous_match_p1_id = wb_final.id
final.source_p1_type = MatchSourceType.WINNER
final.previous_match_p2_id = lb_final.id
final.source_p2_type = MatchSourceType.WINNER
return [n.to_dict() for n in nodes]
def refresh_bracket(t_obj: Tournament):
matches_map = {m.id: m for m in t_obj.matches}
for _ in range(20):
for m in t_obj.matches:
def resolve(src_id, type_):
if not src_id or src_id not in matches_map:
return None
src = matches_map[src_id]
if type_ == MatchSourceType.WINNER:
return src.winner
if type_ == MatchSourceType.LOSER:
if src.winner == "BYE":
return "BYE"
if src.winner:
return src.p1_name if src.winner == src.p2_name else src.p2_name
return None
return None
if m.previous_match_p1_id:
m.p1_name = resolve(m.previous_match_p1_id, m.source_p1_type)
if m.previous_match_p2_id:
m.p2_name = resolve(m.previous_match_p2_id, m.source_p2_type)
# BYE Auto-Win
if not m.winner and (m.p1_name == "BYE" or m.p2_name == "BYE"):
if m.p1_name == "BYE" and m.p2_name == "BYE":
m.winner = "BYE"
elif m.p1_name == "BYE":
m.winner = m.p2_name
slot = WinnerSide.P1 if (r == 1 and i % 2 == 0) else WinnerSide.P2
if slot == WinnerSide.P1:
target.src_p1 = wb_node
target.src_p1_type = MatchSourceType.LOSER
else:
m.winner = m.p1_name
m.status = MatchStatus.FINISHED
target.src_p2 = wb_node
target.src_p2_type = MatchSourceType.LOSER
# Reset Logic
if m.status == MatchStatus.FINISHED and m.winner != "BYE":
has_p1 = bool(m.p1_name)
has_p2 = bool(m.p2_name)
if (
not has_p1
or not has_p2
or (m.winner != m.p1_name and m.winner != m.p2_name)
):
m.winner = None
m.status = MatchStatus.PENDING
m.sets = []
# Finals
final_node = NodeRef(BracketType.FINALS, 1)
wb_final = wb_layers[wb_rounds][0]
lb_final = lb_layers[lb_rounds][0]
# Numbering
display_counter = 1
sorted_matches = sorted(
t_obj.matches, key=lambda x: int(x.id) if x.id.isdigit() else 999
wb_final.next_loss = lb_final
wb_final.next_win = final_node
lb_final.next_win = final_node
final_node.src_p1 = wb_final
final_node.src_p1_type = MatchSourceType.WINNER
final_node.src_p2 = lb_final
final_node.src_p2_type = MatchSourceType.WINNER
all_nodes = []
for r in sorted(wb_layers.keys()):
all_nodes.extend(wb_layers[r])
for r in sorted(lb_layers.keys()):
all_nodes.extend(lb_layers[r])
all_nodes.append(final_node)
else:
all_nodes = []
for r in sorted(wb_layers.keys()):
all_nodes.extend(wb_layers[r])
for n in all_nodes:
n.display_num = display_counter
display_counter += 1
nodes.append(n.to_dict())
return nodes
def initialize_seeding(db: Session, t: models.Tournament):
seeded_teams = get_seeded_positions(
2 ** math.ceil(math.log2(t.team_count)), t.teams
)
for m in sorted_matches:
if m.winner == "BYE" or m.p1_name == "BYE" or m.p2_name == "BYE":
m.number = None
else:
m.number = display_counter
display_counter += 1
# NO LABEL GENERATION HERE - FRONTEND HANDLES IT
r1_nodes = [
n
for n in t.nodes
if n.bracket_type == BracketType.WINNERS and n.round_number == 1
]
r1_nodes.sort(key=lambda x: x.display_number)
for i, node in enumerate(r1_nodes):
t1 = seeded_teams[i * 2]
t2 = seeded_teams[i * 2 + 1]
node.p1_team_id = t1.id if t1 else None
node.p2_team_id = t2.id if t2 else None
advance_flow(db, t)
def update_schedule(t_obj: Tournament):
match_map = {m.id: m for m in t_obj.matches}
depth_cache = {}
def advance_flow(db: Session, t: models.Tournament):
changes = True
while changes:
changes = False
def get_depth(mid):
if mid not in match_map:
return 0
if mid in depth_cache:
return depth_cache[mid]
m = match_map[mid]
d = 1 + max(
get_depth(m.winner_next_match_id) if m.winner_next_match_id else 0,
get_depth(m.loser_next_match_id) if m.loser_next_match_id else 0,
)
depth_cache[mid] = d
return d
for node in t.nodes:
if node.round_number == 1 and node.bracket_type == BracketType.WINNERS:
if node.p1_team_id and not node.p2_team_id:
if _move_team(
db,
t,
node.p1_team_id,
node.winner_next_node_id,
MatchSourceType.WINNER,
):
pass
elif node.p2_team_id and not node.p1_team_id:
if _move_team(
db,
t,
node.p2_team_id,
node.winner_next_node_id,
MatchSourceType.WINNER,
):
pass
criticality_map = {}
for m in t_obj.matches:
criticality_map[m.id] = get_depth(m.id)
# 2. MATCH CREATION
if node.p1_team_id and node.p2_team_id:
if not node.match:
m_id = str(uuid4())
new_match = models.Match(
id=m_id,
tournament_id=t.id,
node_id=node.id,
p1_team_id=node.p1_team_id,
p2_team_id=node.p2_team_id,
status=MatchStatus.PENDING,
court_id=node.planned_court_id,
start_time=node.planned_start_time,
)
db.add(new_match)
db.commit()
db.refresh(node)
changes = True
start_time = t_obj.timestamp
duration = t_obj.duration
# 3. MATCH RESULT PROPAGATION
elif (
node.match.status == MatchStatus.FINISHED
and node.match.winner_team_id
):
winner_id = node.match.winner_team_id
loser_id = (
node.match.p1_team_id
if winner_id == node.match.p2_team_id
else node.match.p2_team_id
)
finish_times: Dict[str, datetime] = {}
court_timers: Dict[str, datetime] = {c.name: start_time for c in t_obj.courts}
if _move_team(
db,
t,
winner_id,
node.winner_next_node_id,
MatchSourceType.WINNER,
):
changes = True
if _move_team(
db, t, loser_id, node.loser_next_node_id, MatchSourceType.LOSER
):
changes = True
unscheduled = []
# 1. Initialize
for m in t_obj.matches:
if m.winner == "BYE" or m.p1_name == "BYE" or m.p2_name == "BYE":
finish_times[m.id] = start_time
m.status = MatchStatus.FINISHED
elif m.status == MatchStatus.FINISHED:
match_start = m.timestamp if m.timestamp else start_time
fin = match_start + timedelta(minutes=duration)
finish_times[m.id] = fin
if m.court_name and m.court_name in court_timers:
if fin > court_timers[m.court_name]:
court_timers[m.court_name] = fin
else:
m.timestamp = None
m.court_name = None
m.status = MatchStatus.PENDING
unscheduled.append(m)
def _move_team(db, t, team_id, target_node_id, source_type):
if not target_node_id or not team_id:
return False
if not court_timers:
target = next((n for n in t.nodes if n.id == target_node_id), None)
if not target:
return False
updated = False
if target.source_p1_type == source_type and not target.p1_team_id:
target.p1_team_id = team_id
updated = True
elif target.source_p2_type == source_type and not target.p2_team_id:
target.p2_team_id = team_id
updated = True
elif not target.p1_team_id:
target.p1_team_id = team_id
updated = True
elif not target.p2_team_id:
target.p2_team_id = team_id
updated = True
if updated:
db.add(target)
db.commit()
return updated
def update_schedule_times(db: Session, t: models.Tournament):
nodes = sorted(t.nodes, key=lambda n: n.display_number)
current_time = t.timestamp
courts = t.courts
if not courts:
return
# 2. Schedule
loop = len(t_obj.matches) * 2
while unscheduled and loop > 0:
loop -= 1
best_court = min(court_timers, key=lambda k: court_timers[k])
current_time_slot = court_timers[best_court]
court_timers: dict[int, datetime] = {c.id: current_time for c in courts}
ready = []
for m in unscheduled:
p1_r = (
finish_times.get(m.previous_match_p1_id, start_time)
if m.previous_match_p1_id
else start_time
)
p2_r = (
finish_times.get(m.previous_match_p2_id, start_time)
if m.previous_match_p2_id
else start_time
)
for node in nodes:
if node.planned_start_time:
continue
if max(p1_r, p2_r) <= current_time_slot:
ready.append(m)
best_court_id = min(court_timers, key=lambda k: court_timers[k])
start = court_timers[best_court_id]
if ready:
ready.sort(key=lambda x: (-criticality_map.get(x.id, 0), x.round))
cand = ready[0]
cand.court_name = best_court
cand.timestamp = current_time_slot
cand.status = MatchStatus.SCHEDULED
node.planned_court_id = best_court_id
node.planned_start_time = start
fin = current_time_slot + timedelta(minutes=duration)
finish_times[cand.id] = fin
court_timers[best_court] = fin
unscheduled.remove(cand)
else:
next_wake = None
for m in unscheduled:
p1_r = (
finish_times.get(m.previous_match_p1_id, start_time)
if m.previous_match_p1_id
else start_time
)
p2_r = (
finish_times.get(m.previous_match_p2_id, start_time)
if m.previous_match_p2_id
else start_time
)
ready_at = max(p1_r, p2_r)
if ready_at > current_time_slot:
if next_wake is None or ready_at < next_wake:
next_wake = ready_at
if next_wake:
court_timers[best_court] = next_wake
else:
break
court_timers[best_court_id] = start + timedelta(minutes=t.duration)
if node.match and node.match.status == MatchStatus.PENDING:
node.match.court_id = best_court_id
node.match.timestamp = start
db.commit()