334 lines
10 KiB
Python
334 lines
10 KiB
Python
# backend/app/logic.py
|
|
import math
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import models
|
|
from .constants import (
|
|
BracketType,
|
|
MatchSourceType,
|
|
MatchStatus,
|
|
TournamentTypes,
|
|
WinnerSide,
|
|
)
|
|
|
|
|
|
def get_seeded_positions(num_slots, teams):
|
|
seeds = [1, 2]
|
|
while len(seeds) < num_slots:
|
|
next_seeds = []
|
|
for s in seeds:
|
|
next_seeds.append(s)
|
|
next_seeds.append(2 * len(seeds) + 1 - s)
|
|
seeds = next_seeds
|
|
return [teams[s - 1] if s <= len(teams) else None for s in seeds]
|
|
|
|
|
|
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
|
|
|
|
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.display_num = 0
|
|
|
|
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.src_p1_type: Optional[MatchSourceType] = None
|
|
self.src_p2_type: Optional[MatchSourceType] = None
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.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,
|
|
}
|
|
|
|
wb_rounds = power
|
|
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)):
|
|
n = NodeRef(BracketType.WINNERS, r)
|
|
wb_layers[r].append(n)
|
|
|
|
# Link Winners
|
|
for r in range(1, wb_rounds):
|
|
for i, node in enumerate(wb_layers[r]):
|
|
target = wb_layers[r + 1][i // 2]
|
|
node.next_win = target
|
|
if i % 2 == 0:
|
|
target.src_p1 = node
|
|
target.src_p1_type = MatchSourceType.WINNER
|
|
else:
|
|
target.src_p2 = node
|
|
target.src_p2_type = MatchSourceType.WINNER
|
|
|
|
lb_layers = {}
|
|
if t.type == TournamentTypes.DOUBLE and size >= 4:
|
|
lb_rounds = (wb_rounds - 1) * 2
|
|
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):
|
|
n = NodeRef(BracketType.LOSERS, r)
|
|
lb_layers[r].append(n)
|
|
if r % 2 == 0:
|
|
current_count //= 2
|
|
|
|
for r in range(1, lb_rounds):
|
|
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.src_p1 = node
|
|
target.src_p1_type = MatchSourceType.WINNER
|
|
else:
|
|
if i % 2 == 0:
|
|
target.src_p1 = node
|
|
target.src_p1_type = MatchSourceType.WINNER
|
|
else:
|
|
target.src_p2 = node
|
|
target.src_p2_type = MatchSourceType.WINNER
|
|
|
|
# Link Drop-down (Winners -> Losers)
|
|
for r in range(1, wb_rounds):
|
|
drop_round = 1 if r == 1 else (r - 1) * 2
|
|
wb_layer_nodes = wb_layers[r]
|
|
lb_layer_nodes = lb_layers[drop_round]
|
|
|
|
for i, wb_node in enumerate(wb_layer_nodes):
|
|
target = None
|
|
if r == 1:
|
|
target = lb_layer_nodes[i // 2]
|
|
else:
|
|
if i < len(lb_layer_nodes):
|
|
target = lb_layer_nodes[i]
|
|
else:
|
|
target = lb_layer_nodes[-1]
|
|
|
|
wb_node.next_loss = target
|
|
|
|
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:
|
|
target.src_p2 = wb_node
|
|
target.src_p2_type = MatchSourceType.LOSER
|
|
|
|
# Finals
|
|
final_node = NodeRef(BracketType.FINALS, 1)
|
|
wb_final = wb_layers[wb_rounds][0]
|
|
lb_final = lb_layers[lb_rounds][0]
|
|
|
|
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
|
|
)
|
|
|
|
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 advance_flow(db: Session, t: models.Tournament):
|
|
changes = True
|
|
while changes:
|
|
changes = False
|
|
|
|
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
|
|
|
|
# 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
|
|
|
|
# 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
|
|
)
|
|
|
|
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
|
|
|
|
|
|
def _move_team(db, t, team_id, target_node_id, source_type):
|
|
if not target_node_id or not team_id:
|
|
return False
|
|
|
|
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
|
|
|
|
court_timers: dict[int, datetime] = {c.id: current_time for c in courts}
|
|
|
|
for node in nodes:
|
|
if node.planned_start_time:
|
|
continue
|
|
|
|
best_court_id = min(court_timers, key=lambda k: court_timers[k])
|
|
start = court_timers[best_court_id]
|
|
|
|
node.planned_court_id = best_court_id
|
|
node.planned_start_time = start
|
|
|
|
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()
|