# Copyright (C) 2023 by Animath
# SPDX-License-Identifier: GPL-3.0-or-later
import os
from asgiref.sync import sync_to_async
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.db.models import QuerySet
from django.urls import reverse_lazy
from django.utils.text import format_lazy, slugify
from django.utils.translation import gettext_lazy as _
from participation.models import Participation, Passage, Pool as PPool, Tournament
class Draw(models.Model):
"""
A draw instance is linked to a :model:`participation.Tournament` and contains all information
about a draw.
"""
tournament = models.OneToOneField(
Tournament,
on_delete=models.CASCADE,
verbose_name=_('tournament'),
help_text=_("The associated tournament.")
)
current_round = models.ForeignKey(
'Round',
on_delete=models.CASCADE,
null=True,
default=None,
related_name='+',
verbose_name=_('current round'),
help_text=_("The current round where teams select their problems."),
)
last_message = models.TextField(
blank=True,
default="",
verbose_name=_("last message"),
help_text=_("The last message that is displayed on the drawing interface.")
)
def get_absolute_url(self):
return reverse_lazy('draw:index') + f'#{slugify(self.tournament.name)}'
@property
def exportable(self) -> bool:
"""
True if any pool of the draw is exportable, ie. can be exported to the tournament interface.
This operation is synchronous.
"""
return any(pool.exportable for r in self.round_set.all() for pool in r.pool_set.all())
async def is_exportable(self) -> bool:
"""
True if any pool of the draw is exportable, ie. can be exported to the tournament interface.
This operation is asynchronous.
"""
return any([await pool.is_exportable() async for r in self.round_set.all() async for pool in r.pool_set.all()])
def get_state(self) -> str:
"""
The current state of the draw.
Can be:
* **DICE_SELECT_POULES** if we are waiting for teams to launch their dice to determine pools and passage order ;
* **DICE_ORDER_POULE** if we are waiting for teams to launch their dice to determine the problem draw order ;
* **WAITING_DRAW_PROBLEM** if we are waiting for a team to draw a problem ;
* **WAITING_CHOOSE_PROBLEM** if we are waiting for a team to accept or reject a problem ;
* **WAITING_FINAL** if this is the final tournament and we are between the two rounds ;
* **DRAW_ENDED** if the draw is ended.
Warning: the current round and the current team must be prefetched in an async context.
"""
if self.current_round.current_pool is None:
return 'DICE_SELECT_POULES'
elif self.current_round.current_pool.current_team is None:
return 'DICE_ORDER_POULE'
elif self.current_round.current_pool.current_team.accepted is not None:
if self.current_round.number < settings.NB_ROUNDS:
# The last step can be the last problem acceptation after the first round
# only for the final between the two rounds
return 'WAITING_FINAL'
else:
return 'DRAW_ENDED'
elif self.current_round.current_pool.current_team.purposed is None:
return 'WAITING_DRAW_PROBLEM'
else:
return 'WAITING_CHOOSE_PROBLEM'
get_state.short_description = _('State')
@property
def information(self):
"""
The information header on the draw interface, which is defined according to the
current state.
Warning: this property is synchronous.
"""
s = ""
if self.last_message:
s += self.last_message + "<br><br>"
match self.get_state():
case 'DICE_SELECT_POULES':
# Waiting for dices to determine pools and passage order
if self.current_round.number == 1:
# Specific information for the first round
s += _("We are going to start the problem draw.<br>"
"You can ask any question if something is not clear or wrong.<br><br>"
"We are going to first draw the pools and the passage order for the first round "
"with all the teams, then for each pool, we will draw the draw order and the problems.")
s += "<br><br>"
s += _("The captains, you can now all throw a 100-sided dice, by clicking on the big dice button. "
"The pools and the passage order during the first round will be the increasing order "
"of the dices, ie. the smallest dice will be the first to pass in pool A.")
case 'DICE_ORDER_POULE':
# Waiting for dices to determine the choice order
s += _("We are going to start the problem draw for the pool <strong>{pool}</strong>, "
"between the teams <strong>{teams}</strong>. "
"The captains can throw a 100-sided dice by clicking on the big dice button "
"to determine the order of draw. The team with the highest score will draw first.") \
.format(pool=self.current_round.current_pool,
teams=', '.join(td.participation.team.trigram
for td in self.current_round.current_pool.teamdraw_set.all()))
case 'WAITING_DRAW_PROBLEM':
# Waiting for a problem draw
td = self.current_round.current_pool.current_team
s += _("The team <strong>{trigram}</strong> is going to draw a problem. "
"Click on the urn in the middle to draw a problem.") \
.format(trigram=td.participation.team.trigram)
case 'WAITING_CHOOSE_PROBLEM':
# Waiting for the team that can accept or reject the problem
td = self.current_round.current_pool.current_team
s += _("The team <strong>{trigram}</strong> drew the problem <strong>{problem}: "
"{problem_name}</strong>.") \
.format(trigram=td.participation.team.trigram,
problem=td.purposed, problem_name=settings.PROBLEMS[td.purposed - 1]) + " "
if td.purposed in td.rejected:
# The problem was previously rejected
s += _("It already refused this problem before, so it can refuse it without penalty and "
"draw a new problem immediately, or change its mind.")
else:
# The problem can be rejected
s += _("It can decide to accept or refuse this problem.") + " "
if len(td.rejected) >= len(settings.PROBLEMS) - settings.RECOMMENDED_SOLUTIONS_COUNT:
s += _("Refusing this problem will add a new 25% penalty "
"on the coefficient of the oral defense.")
else:
s += _("There are still {remaining} refusals without penalty.").format(
remaining=len(settings.PROBLEMS) - settings.RECOMMENDED_SOLUTIONS_COUNT - len(td.rejected))
case 'WAITING_FINAL':
# We are between the two rounds of the final tournament
s += _("The draw for the second round will take place at the end of the first round. Good luck!")
case 'DRAW_ENDED':
# The draw is ended
s += _("The draw is ended. The solutions of the other teams can be found in the tab "
"\"My participation\".")
s += "<br><br>" if s else ""
rules_link = settings.RULES_LINK
s += _("For more details on the draw, the rules are available on "
"<a class=\"alert-link\" href=\"{link}\">{link}</a>.").format(link=rules_link)
return s
async def ainformation(self) -> str:
"""
Asynchronous version to get the information header content.
"""
return await sync_to_async(lambda: self.information)()
def __str__(self):
return str(format_lazy(_("Draw of tournament {tournament}"), tournament=self.tournament.name))
class Meta:
verbose_name = _('draw')
verbose_name_plural = _('draws')
class Round(models.Model):
"""
This model is attached to a :model:`draw.Draw` and represents the draw
for one round of the :model:`participation.Tournament`.
"""
draw = models.ForeignKey(
Draw,
on_delete=models.CASCADE,
verbose_name=_('draw'),
)
number = models.PositiveSmallIntegerField(
choices=[
(1, _('Round 1')),
(2, _('Round 2')),
(3, _('Round 3'))],
verbose_name=_('number'),
help_text=_("The number of the round, 1 or 2 (or 3 for ETEAM)"),
validators=[MinValueValidator(1), MaxValueValidator(3)],
)
current_pool = models.ForeignKey(
'Pool',
on_delete=models.SET_NULL,
null=True,
default=None,
related_name='+',
verbose_name=_('current pool'),
help_text=_("The current pool where teams select their problems."),
)
def get_absolute_url(self):
return reverse_lazy('draw:index') + f'#{slugify(self.draw.tournament.name)}'
@property
def team_draws(self) -> QuerySet["TeamDraw"]:
"""
Returns a query set ordered by pool and by passage index of all team draws.
"""
return self.teamdraw_set.order_by('pool__letter', 'passage_index').all()
async def next_pool(self):
"""
Returns the next pool of the round.
For example, after the pool A, we have the pool B.
"""
pool = self.current_pool
return await self.pool_set.aget(letter=pool.letter + 1)
def __str__(self):
return self.get_number_display()
def clean(self):
if self.number is not None and self.number > settings.NB_ROUNDS:
raise ValidationError({'number': _("The number of the round must be between 1 and {nb}.")
.format(nb=settings.NB_ROUNDS)})
return super().clean()
class Meta:
verbose_name = _('round')
verbose_name_plural = _('rounds')
ordering = ('draw__tournament__name', 'number',)
class Pool(models.Model):
"""
A Pool is a collection of teams in a :model:`draw.Round` of a `draw.Draw`.
It has a letter (eg. A, B, C or D) and a size, between 3 and 5.
After the draw, the pool can be exported in a `participation.Pool` instance.
"""
round = models.ForeignKey(
Round,
on_delete=models.CASCADE,
)
letter = models.PositiveSmallIntegerField(
choices=[
(1, 'A'),
(2, 'B'),
(3, 'C'),
(4, 'D'),
],
verbose_name=_('letter'),
help_text=_("The letter of the pool: A, B, C or D."),
)
size = models.PositiveSmallIntegerField(
verbose_name=_('size'),
validators=[MinValueValidator(3), MaxValueValidator(5)],
help_text=_("The number of teams in this pool, between 3 and 5."),
)
current_team = models.ForeignKey(
'TeamDraw',
on_delete=models.CASCADE,
null=True,
default=None,
related_name='+',
verbose_name=_('current team'),
help_text=_("The current team that is selecting its problem."),
)
associated_pool = models.OneToOneField(
'participation.Pool',
on_delete=models.SET_NULL,
null=True,
default=None,
related_name='draw_pool',
verbose_name=_("associated pool"),
help_text=_("The full pool instance."),
)
def get_absolute_url(self):
return reverse_lazy('draw:index') + f'#{slugify(self.round.draw.tournament.name)}'
@property
def team_draws(self) -> QuerySet["TeamDraw"]:
"""
Returns a query set ordered by passage index of all team draws in this pool.
"""
return self.teamdraw_set.all()
@property
def trigrams(self) -> list[str]:
"""
Returns a list of trigrams of the teams in this pool ordered by passage index.
This property is synchronous.
"""
return [td.participation.team.trigram for td in self.teamdraw_set.order_by('passage_index')
.prefetch_related('participation__team').all()]
async def atrigrams(self) -> list[str]:
"""
Returns a list of trigrams of the teams in this pool ordered by passage index.
This property is asynchronous.
"""
return [td.participation.team.trigram async for td in self.teamdraw_set.order_by('passage_index')
.prefetch_related('participation__team').all()]
async def next_td(self) -> "TeamDraw":
"""
Returns the next team draw after the current one, to know who should draw a new problem.
"""
td = self.current_team
current_index = (td.choose_index + 1) % self.size
td = await self.teamdraw_set.prefetch_related('participation__team').aget(choose_index=current_index)
while td.accepted:
# Ignore if the next team already accepted its problem
current_index += 1
current_index %= self.size
td = await self.teamdraw_set.prefetch_related('participation__team').aget(choose_index=current_index)
return td
@property
def exportable(self) -> bool:
"""
True if this pool is exportable, ie. can be exported to the tournament interface. That means that
each team selected its problem.
This operation is synchronous.
"""
return self.associated_pool_id is None and self.teamdraw_set.exists() \
and all(td.accepted is not None for td in self.teamdraw_set.all())
async def is_exportable(self) -> bool:
"""
True if this pool is exportable, ie. can be exported to the tournament interface. That means that
each team selected its problem.
This operation is asynchronous.
"""
return self.associated_pool_id is None and await self.teamdraw_set.aexists() \
and all([td.accepted is not None async for td in self.teamdraw_set.all()])
async def export(self) -> PPool:
"""
Translates this Pool instance in a :model:`participation.Pool` instance, with the passage orders.
"""
# Create the pool
self.associated_pool, _created = await PPool.objects.aget_or_create(
tournament=self.round.draw.tournament,
round=self.round.number,
letter=self.letter,
)
# Define the participations of the pool
tds = [td async for td in self.team_draws.prefetch_related('participation')]
await self.associated_pool.participations.aset([td.participation async for td in self.team_draws
.prefetch_related('participation')])
await self.asave()
pool2 = None
if self.size == 5:
pool2, _created = await PPool.objects.aget_or_create(
tournament=self.round.draw.tournament,
round=self.round.number,
letter=self.letter,
room=2,
)
await pool2.participations.aset([td.participation async for td in self.team_draws
.prefetch_related('participation')])
# Define the passage matrix according to the number of teams
table = []
if self.size == 3:
table = [
[0, 1, 2],
[1, 2, 0],
[2, 0, 1],
]
elif self.size == 4:
table = [
[0, 1, 2, 3],
[1, 2, 3, 0],
[2, 3, 0, 1],
[3, 0, 1, 2],
]
elif self.size == 5:
table = [
[0, 2, 3, 4],
[1, 3, 4, 0],
[2, 4, 0, 1],
[3, 0, 1, 2],
[4, 1, 2, 3],
]
for i, line in enumerate(table):
passage_pool = self.associated_pool
passage_position = i + 1
if self.size == 5:
# In 5-teams pools, we may create some passages in the second room
if i % 2 == 1:
passage_pool = pool2
passage_position = 1 + i // 2
reporter = tds[line[0]].participation
opponent = tds[line[1]].participation
reviewer = tds[line[2]].participation
observer = tds[line[3]].participation if self.size >= 4 and settings.HAS_OBSERVER else None
# Create the passage
await Passage.objects.acreate(
pool=passage_pool,
position=passage_position,
solution_number=tds[line[0]].accepted,
reporter=reporter,
opponent=opponent,
reviewer=reviewer,
observer=observer,
reporter_penalties=tds[line[0]].penalty_int,
)
# Update Google Sheets
if os.getenv('GOOGLE_PRIVATE_KEY_ID', None):
await sync_to_async(self.associated_pool.update_spreadsheet)()
return self.associated_pool
def __str__(self):
return str(format_lazy(_("Pool {letter}{number}"), letter=self.get_letter_display(), number=self.round.number))
class Meta:
verbose_name = _('pool')
verbose_name_plural = _('pools')
ordering = ('round__draw__tournament__name', 'round__number', 'letter',)
class TeamDraw(models.Model):
"""
This model represents the state of the draw for a given team, including
its accepted problem or their rejected ones.
"""
participation = models.ForeignKey(
Participation,
on_delete=models.CASCADE,
verbose_name=_('participation'),
)
round = models.ForeignKey(
Round,
on_delete=models.CASCADE,
verbose_name=_('round'),
)
pool = models.ForeignKey(
Pool,
on_delete=models.CASCADE,
null=True,
default=None,
verbose_name=_('pool'),
)
passage_index = models.PositiveSmallIntegerField(
choices=zip(range(0, 5), range(0, 5)),
null=True,
default=None,
verbose_name=_('passage index'),
help_text=_("The passage order in the pool, between 0 and the size of the pool minus 1."),
validators=[MinValueValidator(0), MaxValueValidator(4)],
)
choose_index = models.PositiveSmallIntegerField(
choices=zip(range(0, 5), range(0, 5)),
null=True,
default=None,
verbose_name=_('choose index'),
help_text=_("The choice order in the pool, between 0 and the size of the pool minus 1."),
validators=[MinValueValidator(0), MaxValueValidator(4)],
)
accepted = models.PositiveSmallIntegerField(
choices=[
(i, format_lazy(_("Problem #{problem}"), problem=i)) for i in range(1, len(settings.PROBLEMS) + 1)
],
null=True,
default=None,
verbose_name=_("accepted problem"),
)
passage_dice = models.PositiveSmallIntegerField(
choices=zip(range(1, 101), range(1, 101)),
null=True,
default=None,
verbose_name=_("passage dice"),
)
choice_dice = models.PositiveSmallIntegerField(
choices=zip(range(1, 101), range(1, 101)),
null=True,
default=None,
verbose_name=_("choice dice"),
)
purposed = models.PositiveSmallIntegerField(
choices=[
(i, format_lazy(_("Problem #{problem}"), problem=i)) for i in range(1, len(settings.PROBLEMS) + 1)
],
null=True,
default=None,
verbose_name=_("purposed problem"),
)
rejected = models.JSONField(
default=list,
verbose_name=_('rejected problems'),
)
def get_absolute_url(self):
return reverse_lazy('draw:index') + f'#{slugify(self.round.draw.tournament.name)}'
@property
def last_dice(self):
"""
The last dice that was thrown.
"""
return self.passage_dice if self.round.draw.get_state() == 'DICE_SELECT_POULES' else self.choice_dice
@property
def penalty_int(self):
"""
The number of penalties, which is the number of rejected problems after the P - 5 free rejects
(P - 6 for ETEAM), where P is the number of problems.
"""
return max(0, len(self.rejected) - (len(settings.PROBLEMS) - settings.RECOMMENDED_SOLUTIONS_COUNT))
@property
def penalty(self):
"""
The penalty multiplier on the reporter oral, in percentage, which is a malus of 25% for each penalty.
"""
return 25 * self.penalty_int
def __str__(self):
return str(format_lazy(_("Draw of the team {trigram} for the pool {letter}{number}"),
trigram=self.participation.team.trigram,
letter=self.pool.get_letter_display() if self.pool else "",
number=self.round.number))
class Meta:
verbose_name = _('team draw')
verbose_name_plural = _('team draws')
ordering = ('round__draw__tournament__name', 'round__number', 'pool__letter', 'passage_index',
'choice_dice', 'passage_dice',)