diff options
Diffstat (limited to 'app/utilities.py')
| -rw-r--r-- | app/utilities.py | 171 |
1 files changed, 171 insertions, 0 deletions
diff --git a/app/utilities.py b/app/utilities.py new file mode 100644 index 0000000..cd75d2a --- /dev/null +++ b/app/utilities.py @@ -0,0 +1,171 @@ +from math import floor +from app import app, db +from datetime import date +from sqlalchemy import func +from typing import Tuple, Sequence +from random import shuffle, sample +from random import randint, getrandbits +from names import get_first_name, get_last_name +from app.model import Player, Court, Game, DisplayGame + + +def _shuffle_two_lists(a: list, b: list) -> Tuple[list, list]: + indices = list(range(len(a))) + shuffle(indices) + a = [a[index] for index in indices] + b = [b[index] for index in indices] + return a, b + + +def _pick_from_list_after_sorting_other(a: list, b: list[int]) -> list: + indices = [i[0] for i in sorted(enumerate(b), key=lambda x: x[1])] + return [a[index] for index in indices] + + +def sample_list(indices: list, cost: list[int], count=None) -> list: + if count is None: + count = len(indices) + indices, cost = _shuffle_two_lists(indices, cost) + indices = _pick_from_list_after_sorting_other(indices, cost)[:count] + return indices + + +def _normalize_skill(level: int, new_max: int) -> int: + scaling = (new_max - 1) / int(app.config["MAX_PLAYER_LEVEL"]) + new_level = round(scaling * level) + 1 + return new_level + + +def select_players(players: Sequence[Player], count: int) -> list[Player]: + games = db.session.query(Game).filter(func.date(Game.date) == date.today()).all() + player_identifiers = [] + for game in games: + for i in range(4): + player_identifiers.append(getattr(game, f"player_{i+1}")) + identifiers = [player.id for player in players] + play_count = [player_identifiers.count(player.id) for player in players] + ids = sample_list(identifiers, play_count, count) + selected_players = [player for player in players if player.id in ids] + return selected_players + + +def group_by_four(players: list[Player], new_max: int) -> list[list[Player]]: + identifiers = [player.id for player in players] + player_levels = [_normalize_skill(int(player.level), new_max) for player in players] + identifiers = sample_list(identifiers, player_levels) + court_players = [] + for i in range(0, len(players), app.config["PLAYERS_PER_COURT"]): + court_ids = identifiers[i : i + app.config["PLAYERS_PER_COURT"]] + selected_players = [player for player in players if player.id in court_ids] + court_players.append(selected_players) + return court_players + + +def separate_teams_and_players( + players: Sequence[Player], +) -> tuple[dict[int, list[Player]], list[Player]]: + unique_team_id = {player.team for player in players} + grouped_players_by_team = {team_id: [] for team_id in unique_team_id} + for player in players: + team = player.team + grouped_players_by_team[team].append(player) + active_teams = {} + players_without_team = [] + for team_id in grouped_players_by_team: + if team_id == 0: + players_without_team += [ + player for player in grouped_players_by_team[team_id] + ] + else: + if len(grouped_players_by_team[team_id]) >= 4: + active_teams[team_id] = grouped_players_by_team[team_id] + else: + players_without_team += [ + player for player in grouped_players_by_team[team_id] + ] + return active_teams, players_without_team + + +def get_placeholder_games(courts: list[Court], name: str) -> list[DisplayGame]: + p = db.session.query(Player).where(Player.first_name == name).first() + if p is None: + raise ValueError(f"Database isn't initialized.") + player_data = {"round_id": None} | {f"player_{i+1}": p.id for i in range(4)} + games = [ + DisplayGame(Game(**({"court_id": court.id} | player_data))) for court in courts # type: ignore + ] + return games + + +def get_team_games( + courts: list[Court], teams: dict[int, list[Player]] +) -> list[DisplayGame]: + games = [] + for court, (_, players) in zip(courts, teams.items()): + players = sample(players, app.config["PLAYERS_PER_COURT"]) + players.sort(key=lambda x: x.level, reverse=True) + game = Game( + None, court.id, players[0].id, players[3].id, players[1].id, players[2].id + ) + games.append(DisplayGame(game)) + return games + + +def get_shuffle_games( + courts: list[Court], players: Sequence[Player], total_levels: int +) -> list[DisplayGame]: + required_player_count = len(courts) * app.config["PLAYERS_PER_COURT"] + selected_players = select_players(players, required_player_count) + court_players = group_by_four(selected_players, total_levels) + games = [] + for court, players in zip(courts, court_players): + game = Game( + None, court.id, players[0].id, players[3].id, players[1].id, players[2].id + ) + games.append(DisplayGame(game)) + return games + + +def initialize_default_players(): + db.session.query(Player).delete() + for k, v in app.config["DEFAULTS"].items(): + db.session.add(Player(**{"id": v, "first_name": k, "last_name": ""})) + guest_offset = 10000 + for i, (k, v) in enumerate(app.config["GUESTS"].items()): + level_offset = app.config["GUEST_COUNT"] * i + for j in range(app.config["GUEST_COUNT"]): + db.session.add( + Player( + **{ + "id": guest_offset + level_offset + j, + "first_name": k, + "last_name": f"{j + 1}", + "level": v, + } + ) + ) + db.session.commit() + + +def initialize_random_database(): + initialize_default_players() + player_count = 80 + total_teams = 5 + players_per_team = 5 + for i in range(player_count): + if i < total_teams * players_per_team: + team = floor(i / players_per_team) + 1 + else: + team = 0 + args = { + "id": 3 + i, + "first_name": get_first_name(), + "last_name": get_last_name(), + "team": team, + "status": bool(getrandbits(1)), + "level": randint( + app.config["MIN_PLAYER_LEVEL"], app.config["MAX_PLAYER_LEVEL"] + ), + } + db.session.add(Player(**args)) + db.session.commit() |