from math import floor from app import app, db from datetime import date from operator import attrgetter from sqlalchemy import select, func from app.model import ( Player, Players, Courts, Game, DisplayPlayer, DisplayTeam, DisplayGame, ) from app.utilities import ( get_placeholder_games, get_team_games, get_shuffle_games, separate_teams_and_players, ) class Session: levels: int = 10 teams: bool = False proposal: list[DisplayGame] = [] courts: Courts = Courts() players: Players = Players() def __init__(self): games = db.session.query(Game).filter(Game.date.date() == date.today()).all() if len(games) > 0: return id = self.players.default_id() player_data = {f"player_{i+1}": id for i in range(4)} for court in self.courts.all(): args = {"round_id": 0, "court_id": court.id} | player_data db.session.add(Game(*args)) db.session.commit() def round(self) -> int: query = select(func.max(Game.round_id)).where(Game.date.date() == date.today()) round_id = db.session.scalars(query).first() if round_id is None: raise ValueError(f"Database isn't initialized") return round_id def all(self) -> list[DisplayGame]: games = ( db.session.query(Game) .filter(Game.date.date() == date.today()) .filter(Game.round_id == self.round()) .order_by(Game.court_id) .all() ) dg = [] for game in games: dp = [] for i in range(4): p = db.session.get(Player, getattr(game, f"player_{i+1}")) if p is None: raise ValueError(f"Player id mismatch.") t1 = DisplayTeam(DisplayPlayer(dp[0]), DisplayPlayer(dp[1])) t2 = DisplayTeam(DisplayPlayer(dp[2]), DisplayPlayer(dp[3])) db.session.get(Player, game.player_2) dg.append(DisplayGame(game.court_id, t1, t2)) return dg def propose(self) -> list[DisplayGame]: reserved_courts = [court for court in self.courts.all() if not court.status] reserved_games = get_placeholder_games(reserved_courts, "RESERVED") active_courts = self.courts.active() active_players = self.players.active() game_count = floor(len(active_players) / app.config["PLAYER_PER_COURT"]) game_count = min(len(active_courts), game_count) used_courts = active_courts[:game_count] unused_courts = active_courts[game_count:] blank_games = get_placeholder_games(unused_courts, "---") active_teams = {} if self.teams: active_teams, active_players = separate_teams_and_players(active_players) team_courts = used_courts[: len(active_teams)] shuffle_courts = used_courts[len(active_teams) :] team_games = get_team_games(team_courts, active_teams) shuffle_games = get_shuffle_games(shuffle_courts, active_players, self.levels) self.proposal = shuffle_games + team_games + blank_games + reserved_games self.proposal.sort(key=attrgetter("court_id")) return self.proposal def confirm(self) -> list[DisplayGame]: if len(self.proposal) == 0: return self.all() for game in self.proposal: db.session.add(game) db.session.commit() self.proposal = [] return self.all() def reset(self) -> list[DisplayGame]: db.session.query(Game).filter(Game.date.date() == date.today()).delete() db.session.commit() self.__init__() return self.all()