"""An automatic seeded knockout schedule."""
from __future__ import annotations
import dataclasses
import datetime
import math
from collections.abc import Iterable, Mapping, Sized
from typing_extensions import Self
from ..match_period import KnockoutMatch, Match, MatchSlot, MatchType
from ..match_period_clock import MatchPeriodClock, OutOfTimeException, Spacing
from ..scores import Scores
from ..teams import Team
from ..types import (
ArenaName,
KnockoutRoundSpacingData,
MatchNumber,
ScheduleAutomaticKnockoutData,
ScheduleKnockoutRoundSpacingData,
TLA,
)
from . import seeding, stable_random
from .base_scheduler import (
BaseKnockoutScheduleData,
BaseKnockoutScheduler,
DEFAULT_KNOCKOUT_BRACKET_NAME,
)
from .types import ScheduleHost
@dataclasses.dataclass(frozen=True, slots=True)
class RoundsSpacing:
default: Spacing
overrides: Mapping[int, Spacing]
@classmethod
def parse(cls, spacing_data: ScheduleKnockoutRoundSpacingData) -> Self:
def build(spacing_entry: KnockoutRoundSpacingData) -> Spacing:
return Spacing(
delay_flex=datetime.timedelta(seconds=spacing_entry['delay_flex']),
minimum=datetime.timedelta(seconds=spacing_entry['minimum']),
nominal=datetime.timedelta(seconds=spacing_entry['nominal']),
)
default = build(spacing_data['default'])
overrides = {
num: build(entry)
for num, entry in spacing_data.get('overrides', {}).items()
}
return cls(default, overrides)
def __post_init__(self) -> None:
if not (
all(x < 0 for x in self.overrides.keys()) or
all(x > 0 for x in self.overrides.keys())
):
raise ValueError(
"Invalid overrides configuration -- "
"all entries must be positive or all must be negative. "
f"Got {set(self.overrides.keys())}.",
)
def get(self, *, round_num: int, rounds_remaining: int) -> Spacing:
if (spacing := self.overrides.get(round_num)) is not None:
return spacing
if (spacing := self.overrides.get(-rounds_remaining)) is not None:
return spacing
return self.default
class AutoKnockoutScheduleData(BaseKnockoutScheduleData):
knockout: ScheduleAutomaticKnockoutData
[docs]
class KnockoutScheduler(BaseKnockoutScheduler[AutoKnockoutScheduleData]):
"""
A class that can be used to generate a knockout schedule based on seeding.
Due to the way the seeding logic works, this class is suitable only when
games feature four slots for competitors, with the top two progressing to
the next round.
:param schedule: The league schedule.
:param scores: The scores.
:param dict arenas: The arenas.
:param int num_teams_per_arena: The usual number of teams per arena.
:param dict teams: The teams.
:param config: Custom configuration for the knockout scheduler.
"""
num_teams_per_arena = 4
"""
Constant value due to the way the automatic seeding works.
"""
def __init__(
self,
schedule: ScheduleHost,
scores: Scores,
arenas: Iterable[ArenaName],
num_teams_per_arena: int,
teams: Mapping[TLA, Team],
config: AutoKnockoutScheduleData,
) -> None:
if num_teams_per_arena != self.num_teams_per_arena:
raise ValueError(
"The automatic knockout scheduler can only be used for {} teams"
" per arena (and not {})".format(
self.num_teams_per_arena,
num_teams_per_arena,
),
)
super().__init__(schedule, scores, arenas, num_teams_per_arena, teams, config)
self.R = stable_random.Random()
self.clock = MatchPeriodClock(self.period, self.schedule.delays)
def _add_round_of_matches(
self,
matches: list[list[TLA]],
arenas: Iterable[ArenaName],
rounds_remaining: int,
) -> None:
"""
Add a whole round of matches.
:param list matches: A list of lists of teams for each match.
"""
knockout_round = self._append_knockout_round(rounds_remaining)
round_num = 0
while len(matches):
# Deliberately not using iterslots since we need to ensure
# that the time advances even after we've run out of matches
start_time = self.clock.current_time
end_time = start_time + self.schedule.match_duration
new_matches = {}
for arena in arenas:
teams: list[TLA | None] = list(matches.pop(0))
if len(teams) < self.num_teams_per_arena:
# Fill empty zones with None
teams += [None] * (self.num_teams_per_arena - len(teams))
# Randomise the zones
self.R.shuffle(teams)
num = MatchNumber(len(self.schedule.matches))
display_name = self.get_match_display_name(
rounds_remaining,
round_num,
num,
)
match = KnockoutMatch(
num,
display_name,
arena,
teams,
start_time,
end_time,
MatchType.knockout,
knockout_bracket=DEFAULT_KNOCKOUT_BRACKET_NAME,
# Just the finals don't use the resolved ranking
use_resolved_ranking=rounds_remaining != 0,
)
knockout_round.append(match)
new_matches[arena] = match
if len(matches) == 0:
break
self.clock.advance_time(self.schedule.match_duration)
self.schedule.matches.append(MatchSlot(new_matches))
self.period.matches.append(MatchSlot(new_matches))
round_num += 1
[docs]
def get_winners(self, game: Match) -> list[TLA]:
"""
Find the parent match's winners.
:param game: A game.
"""
ranking = self.get_ranking(game)
return ranking[:2]
def _add_round(self, arenas: Iterable[ArenaName], rounds_remaining: int) -> None:
prev_round = self.knockout_rounds[-1]
matches = []
for i in range(0, len(prev_round), 2):
winners = []
for parent in prev_round[i:i + 2]:
winners += self.get_winners(parent)
matches.append(winners)
self._add_round_of_matches(matches, arenas, rounds_remaining)
def _add_first_round(self, conf_arity: int | None = None) -> None:
teams = self._get_seeds()
arity = len(teams)
if conf_arity is not None and conf_arity < arity:
arity = conf_arity
# Seed the random generator with the seeded team list
# This makes it unpredictable which teams will be in which zones
# until the league scores have been established
self.R.seed(''.join(teams).encode('utf-8'))
matches = []
for seeds in seeding.first_round_seeding(arity):
match_teams = [teams[seed] for seed in seeds]
matches.append(match_teams)
rounds_remaining = self.get_rounds_remaining(matches)
self._add_round_of_matches(matches, self.arenas, rounds_remaining)
[docs]
@staticmethod
def get_rounds_remaining(prev_matches: Sized) -> int:
return int(math.log(len(prev_matches), 2))
def _apply_spacing(self, spacing: Spacing) -> None:
self.clock.apply_spacing(
spacing=spacing,
recover_time=self.schedule._recover_time,
)
def _add_knockouts(self) -> None:
knockout_conf = self.config['knockout']
rounds_spacing = RoundsSpacing.parse(knockout_conf['round_spacing'])
self._add_first_round(conf_arity=knockout_conf.get('arity'))
while len(self.knockout_rounds[-1]) > 1:
# Number of rounds remaining to be added
rounds_remaining = self.get_rounds_remaining(self.knockout_rounds[-1])
# Add the delay between rounds
self._apply_spacing(rounds_spacing.get(
round_num=len(self.knockout_rounds),
rounds_remaining=rounds_remaining,
))
arenas: Iterable[ArenaName]
if rounds_remaining <= knockout_conf['single_arena']['rounds']:
arenas = knockout_conf['single_arena']['arenas']
else:
arenas = self.arenas
self._add_round(arenas, rounds_remaining - 1)
[docs]
def add_knockouts(self) -> None:
try:
self._add_knockouts()
except OutOfTimeException as e:
raise OutOfTimeException(
"Ran out of time scheduling the knockouts. This usually indicates "
"that there are more teams than it is possible to schedule matches "
"for within the given period. Consider adjusting the number of teams "
"which progress to the knockouts or allowing more time.",
) from e