"""Match schedule library."""
from __future__ import annotations
import bisect
import datetime
from collections.abc import Iterable, Iterator, Mapping, Sequence
from pathlib import Path
from typing import Any, TypeVar
from typing_extensions import TypedDict
import dateutil.tz
from league_ranker import RankedPosition
from . import yaml_loader
from .arenas import Arena
from .knockout_scheduler import (
AutoKnockoutScheduleData,
KnockoutBracket,
KnockoutRound,
KnockoutScheduler,
modernise_knockout_config_if_needed,
StaticKnockoutScheduleData,
StaticScheduler,
StructuredKnockoutScheduleData,
StructuredScheduler,
)
from .knockout_scheduler.base_scheduler import BaseKnockoutScheduler
from .match_period import Delay, Match, MatchPeriod, MatchSlot, MatchType
from .match_period_clock import MatchPeriodClock
from .scores import Scores
from .teams import Team
from .types import (
ArenaName,
DelayData,
ExtraSpacingData,
KnockoutData,
LeagueMatches,
MatchNumber,
MatchSlotLengthsData,
ScheduleData,
ShepherdName,
StagingTimingsData,
TLA,
)
TSchedule = TypeVar('TSchedule', bound='MatchSchedule')
[docs]
class StagingOffsets(TypedDict):
opens: datetime.timedelta
closes: datetime.timedelta
duration: datetime.timedelta
signal_shepherds: Mapping[ShepherdName, datetime.timedelta]
signal_teams: datetime.timedelta
[docs]
class StagingTimes(TypedDict):
opens: datetime.datetime
closes: datetime.datetime
duration: datetime.timedelta
signal_shepherds: Mapping[ShepherdName, datetime.datetime]
signal_teams: datetime.datetime
[docs]
class WrongNumberOfTeams(Exception):
def __init__(
self,
match_n: int,
arena_name: str,
teams: Sequence[TLA | None],
num_teams_per_arena: int,
) -> None:
super().__init__(match_n, arena_name, teams, num_teams_per_arena)
self.match_n = match_n
self.arena_name = arena_name
self.teams = teams
self.num_teams_per_arena = num_teams_per_arena
def __str__(self) -> str:
return "Match {}{} has {} teams but must have {}".format(
self.arena_name,
self.match_n,
len(self.teams),
self.num_teams_per_arena,
)
[docs]
def parse_ranges(ranges: str) -> set[int]:
"""
Parse a comma separated list of numbers which may include ranges
specified as hyphen-separated numbers.
From https://stackoverflow.com/questions/6405208
"""
result: list[int] = []
for part in ranges.split(','):
if '-' in part:
a_, b_ = part.split('-')
a, b = int(a_), int(b_)
result.extend(range(a, b + 1))
else:
a = int(part)
result.append(a)
return set(result)
[docs]
def get_timezone(name: str) -> datetime.tzinfo:
tzinfo = dateutil.tz.gettz(name)
if tzinfo is None:
raise ValueError(f"Failed to load timezone info for {name!r}")
return tzinfo
[docs]
class MatchSchedule:
"""
A match schedule.
"""
[docs]
@classmethod
def create(
cls: type[TSchedule],
config_fname: Path,
league_fname: Path,
knockout_fname: Path,
scores: Scores,
arenas: Mapping[ArenaName, Arena],
num_teams_per_arena: int,
teams: Mapping[TLA, Team],
) -> TSchedule:
"""
Create a new match schedule around the given config data.
:param Path config_fname: The filename of the main config file.
:param Path league_fname: The filename of the file containing the league matches.
:param `.Scores` scores: The scores for the competition.
:param dict arenas: A mapping of arena ids to :class:`.Arena` instances.
:param int num_teams_per_arena: The usual number of teams per arena.
:param dict teams: A mapping of TLAs to :class:`.Team` instances.
"""
y: ScheduleData = yaml_loader.load(config_fname)
league: LeagueMatches = yaml_loader.load(league_fname)['matches']
schedule = cls(y, league, teams, num_teams_per_arena)
knockout_config = modernise_knockout_config_if_needed(y['knockout'])
k: BaseKnockoutScheduler[Any]
if knockout_config['scheduler'] == 'static':
k = StaticScheduler(
schedule,
scores,
arenas,
num_teams_per_arena,
teams,
config=StaticKnockoutScheduleData({
'match_periods': y['match_periods'],
'brackets': knockout_config.get('brackets', ()),
'static_knockout': StaticScheduler.modernise_config_if_needed(
y['static_knockout'],
),
}),
)
elif knockout_config['scheduler'] == 'structured':
structured_knockout: KnockoutData = yaml_loader.load(knockout_fname)
k = StructuredScheduler(
schedule,
scores,
arenas,
num_teams_per_arena,
teams,
config=StructuredKnockoutScheduleData({
'match_periods': y['match_periods'],
'brackets': knockout_config.get('brackets', ()),
'round_spacing': knockout_config['round_spacing'],
'structure': structured_knockout['structured_knockout'],
}),
)
elif knockout_config['scheduler'] == 'automatic':
k = KnockoutScheduler(
schedule,
scores,
arenas,
num_teams_per_arena,
teams,
config=AutoKnockoutScheduleData({
'match_periods': y['match_periods'],
'brackets': knockout_config.get('brackets', ()),
'knockout': knockout_config,
}),
)
else:
raise ValueError(
f"Unsupported knockout scheduler type {knockout_config['scheduler']!r}",
)
k.add_knockouts()
k.validate_brackets()
schedule.knockout_brackets = k.knockout_brackets
schedule.knockout_rounds = k.knockout_rounds
schedule.match_periods.append(k.period)
if 'tiebreaker' in y:
schedule.add_tiebreaker(scores, y['tiebreaker'])
return schedule
def __init__(
self,
y: ScheduleData,
league: LeagueMatches,
teams: Mapping[TLA, Team],
num_teams_per_arena: int,
) -> None:
self._num_corners = num_teams_per_arena
self.teams = teams
"""A mapping of TLAs to :class:`.Team` instances."""
self.match_periods: list[MatchPeriod] = []
r"""
A list of the :class:`.MatchPeriod`\s which contain the matches
for the competition.
"""
self.knockout_brackets: Sequence[KnockoutBracket]
"""
A list of brackets which make up the knockout.
This currently has no bearing on the actual matches and is purely a
display consideration.
"""
self.knockout_rounds: Sequence[KnockoutRound] = []
"""
A list of the knockout matches by round. Each entry in the list
represents a round of knockout matches, such that `knockout_rounds[-1]`
contains a list with only one match -- the final.
"""
for e in y['match_periods']['league']:
if 'max_end_time' in e:
max_end_time = e['max_end_time']
else:
max_end_time = e['end_time']
period = MatchPeriod(
e['start_time'],
e['end_time'],
max_end_time,
e['description'],
[],
MatchType.league,
)
self.match_periods.append(period)
self._load_match_slot_lengths(y['match_slot_lengths'])
self._load_staging_times(y['staging'])
self._build_extra_spacing(y['league']['extra_spacing'])
self._build_delaylist(y['delays'])
self.matches: list[MatchSlot] = []
"""
A list of match slots in the schedule. Each match slot is a dict of
arena to the :class:`.Match` occurring in that arena.
"""
self.n_planned_league_matches = 0
"""The number of planned league matches."""
self._build_matchlist(league)
self.timezone = get_timezone(y.get('timezone', 'UTC'))
self.n_league_matches = self.n_matches()
def _load_match_slot_lengths(self, yamldata: MatchSlotLengthsData) -> None:
durations = {
key: datetime.timedelta(0, value) # type: ignore[arg-type]
for key, value in yamldata.items()
}
pre = durations['pre']
post = durations['post']
match = durations['match']
total = durations['total']
if total != pre + post + match:
raise ValueError("Match slot lengths are inconsistent.")
self.match_slot_lengths = durations
self.match_duration: datetime.timedelta = total
def _load_staging_times(self, yamldata: StagingTimingsData) -> None:
def to_timedeltas(item: Any) -> Any:
if isinstance(item, dict):
return {
key: to_timedeltas(value)
for key, value in item.items()
}
else:
return datetime.timedelta(seconds=item)
durations: StagingOffsets = to_timedeltas(yamldata)
opens = durations['opens']
closes = durations['closes']
duration = durations['duration']
if duration != opens - closes:
raise ValueError("Staging timings are inconsistent.")
for other in ('signal_teams', 'signal_shepherds'):
if other not in durations:
raise ValueError(f"Staging times missing {other!r} key.")
self.staging_times = durations
[docs]
def get_staging_times(self, match: Match) -> StagingTimes:
pre = self.match_slot_lengths['pre']
match_start = match.start_time + pre
offsets = self.staging_times
signal_shepherds: dict[ShepherdName, datetime.datetime] = {
area: match_start - offset
for area, offset in offsets['signal_shepherds'].items()
}
return {
'opens': match_start - offsets['opens'],
'closes': match_start - offsets['closes'],
'duration': self.staging_times['duration'],
'signal_shepherds': signal_shepherds,
'signal_teams': match_start - offsets['signal_teams'],
}
def _build_extra_spacing(self, yamldata: list[ExtraSpacingData] | None) -> None:
spacing: dict[MatchNumber, datetime.timedelta] = {}
if not yamldata:
self._spacing = spacing
return
for info in yamldata:
match_numbers = parse_ranges(info['match_numbers'])
duration = datetime.timedelta(seconds=info['duration'])
for num in match_numbers:
assert num not in spacing
spacing[MatchNumber(num)] = duration
self._spacing = spacing
def _build_delaylist(self, yamldata: list[DelayData] | None) -> None:
delays: list[Delay] = []
if yamldata is None:
# No delays, hurrah
self.delays = delays
return
for info in yamldata:
d = Delay(datetime.timedelta(seconds=info['delay']), info['time'])
delays.append(d)
delays.sort(key=lambda x: x.time)
self.delays = delays
def _recover_time(self, recovered_time: Delay) -> None:
"""
Callback to record that previously delayed time has been recovered.
This is expected to be passed to `MatchPeriodClock.apply_spacing` as
part of either the knockouts or league scheduling. There it is called as
to indicate that a flexible gap in the schedule has been shrunk,
recovering time previously lost to delays.
This is necessary to ensure that the central list of delays (held on
this instance) is in sync with the actual timings, so that calls to
`delay_at` return values which correspond correctly.
The passed delay value should:
- be at the `time` of the earlier end of the time window which was
shortened
- have a *negative* `delay` value, with magnitude of the amount of time
recovered
"""
if recovered_time.delay >= datetime.timedelta(0):
raise ValueError(
f"Recovered duration should be negative. (Got {recovered_time!r})",
)
bisect.insort(self.delays, recovered_time, key=lambda x: x.time)
[docs]
def remove_drop_outs(
self,
teams: Iterable[TLA | None],
since_match: MatchNumber,
) -> list[TLA | None]:
"""
Take a list of TLAs and replace the teams that have dropped out with
``None`` values.
:param list teams: A list of TLAs.
:param int since_match: The match number to check for drop outs from.
:return: A new list containing the appropriate teams.
"""
new_teams: list[TLA | None] = []
for tla in teams:
if tla is None:
new_teams.append(None)
else:
if self.teams[tla].is_still_around(since_match):
new_teams.append(tla)
else:
new_teams.append(None)
return new_teams
def _build_matchlist(self, yamldata: LeagueMatches | None) -> None:
"""
Build the match list.
:param dict yamldata: The matches data from the league file, formatted
as a dict of match numbers to dicts of arena to lists of TLAs.
"""
if yamldata is None:
return
match_numbers = sorted(yamldata.keys())
self.n_planned_league_matches = len(match_numbers)
if tuple(match_numbers) != tuple(range(len(match_numbers))):
raise Exception("Matches are not a complete 0-N range")
# Effectively just the .values(), except that it's ordered by number
raw_matches = [yamldata[m] for m in match_numbers]
match_n = MatchNumber(0)
for period in self.match_periods:
# Fill this period with matches
clock = MatchPeriodClock(period, self.delays)
# No extra spacing for matches at the start of a period
# Fill this match period with matches
for start in clock.iterslots(self.match_duration):
try:
arenas = raw_matches.pop(0)
except IndexError:
# no more matches left
break
match_slot = self._create_league_match_slot(start, arenas, match_n)
period.matches.append(match_slot)
self.matches.append(match_slot)
match_n = MatchNumber(match_n + 1)
extra_spacing = self._spacing.get(match_n)
if extra_spacing:
clock.advance_time(extra_spacing)
def _create_league_match_slot(
self,
start_time: datetime.datetime,
arenas: Mapping[ArenaName, Sequence[TLA | None]],
match_n: MatchNumber,
) -> MatchSlot:
"""
Returns a dict of arena to :class:`.Match` for the given start time,
arenas dict and match number.
"""
match_slot: dict[ArenaName, Match] = {}
end_time = start_time + self.match_duration
for arena_name, teams in arenas.items():
teams = self.remove_drop_outs(teams, match_n)
display_name = f"Match {match_n}"
if len(teams) != self._num_corners:
raise WrongNumberOfTeams(match_n, arena_name, teams, self._num_corners)
match = Match(
match_n,
display_name,
arena_name,
teams,
start_time,
end_time,
MatchType.league,
use_resolved_ranking=False,
)
match_slot[arena_name] = match
return MatchSlot(match_slot)
[docs]
def delay_at(self, date: datetime.datetime) -> datetime.timedelta:
"""
Calculates the active delay at a given ``date``. Intended for use
only in exposing the current delay value -- scheduling should be
done using a :class:`.MatchPeriodClock` instead.
:param datetime date: The date to find the delay for.
:return: A :class:`datetime.timedelta` specifying the active delay.
"""
total = datetime.timedelta()
period = self.period_at(date)
if not period:
# No current period, no delays active
return total
delays = MatchPeriodClock.delays_for_period(period, self.delays)
for delay in delays:
if delay.time > date:
break
total += delay.delay
return total
[docs]
def matches_at(self, date: datetime.datetime) -> Iterator[Match]:
"""
Get all the matches that occur around a specific ``date``.
:param datetime date: The date at which matches occur.
:return: An iterable list of matches.
"""
for slot in self.matches:
for match in slot.values():
if match.start_time <= date < match.end_time:
yield match
[docs]
def period_at(self, date: datetime.datetime) -> MatchPeriod | None:
"""
Get the match period that occur around a specific ``date``.
:param datetime date: The date at which period occurs.
:return: The period at that time or ``None``.
"""
for period in self.match_periods:
if period.start_time <= date < period.max_end_time:
return period
return None
[docs]
def n_matches(self) -> int:
"""
Get the number of matches.
:return: The number of matches.
"""
return len(self.matches)
@property
def final_match(self) -> Match:
"""
Get the :class:`.Match` for the last match of the competition.
This is the info for the 'finals' of the competition (i.e: the
last of the knockout matches) unless there is a tiebreaker.
"""
last_match_slot = self.matches[-1]
last_matches = list(last_match_slot.values())
assert len(last_matches) == 1, last_match_slot
return last_matches[0]
[docs]
def add_tiebreaker(self, scores: Scores, time: datetime.datetime) -> None:
"""
Add a tie breaker to the league if required. Also set a ``tiebreaker``
attribute if necessary.
:param `.Scores` scores: The scores for the competition.
:param datetime.datetime time: The time to have the tiebreaker match.
"""
finals_info = self.knockout_rounds[-1][0]
finals_key = (finals_info.arena, finals_info.num)
try:
finals_positions = scores.knockout.game_positions[finals_key]
except KeyError:
return
winners = finals_positions.get(RankedPosition(1))
if not winners:
raise AssertionError("The only winning move is not to play.")
if len(winners) > 1: # Act surprised!
# Start with the winning teams in the same order as in the finals
tiebreaker_teams = [
team if team in winners else None
for team in finals_info.teams
]
# Use a static permutation
permutation = [3, 2, 0, 1]
tiebreaker_teams = [
tiebreaker_teams[permutation[n]]
for n in permutation
]
# Inject new match
end_time = time + self.match_duration
num = self.n_matches()
arena = finals_info.arena
match = Match(
num=MatchNumber(num),
display_name=f"Tiebreaker (#{num})",
arena=arena,
teams=tiebreaker_teams,
type=MatchType.tiebreaker,
start_time=time,
end_time=end_time,
use_resolved_ranking=False,
)
slot = MatchSlot({arena: match})
self.matches.append(slot)
match_period = MatchPeriod(
time,
end_time,
end_time,
'Tiebreaker',
[slot],
MatchType.tiebreaker,
)
self.match_periods.append(match_period)
self.tiebreaker = match
@property
def datetime_now(self) -> datetime.datetime:
"""Get the current date and time, with the correct timezone."""
return datetime.datetime.now(self.timezone)