diff options
| author | Karan Jayachandra <mail@karanjayachandra.com> | 2025-07-06 12:58:07 +0200 |
|---|---|---|
| committer | Karan Jayachandra <mail@karanjayachandra.com> | 2025-07-06 12:58:07 +0200 |
| commit | 0da8dc16fb26e9df5d0efd7c7a463e354d8e00cf (patch) | |
| tree | d016270ebf8e6e9fef499da5767c99b656c75075 /app/model.py | |
| parent | 155e77d22f24d1984476144f48bf173044d089a0 (diff) | |
Transitioned back to a traditional flask app
Diffstat (limited to 'app/model.py')
| -rw-r--r-- | app/model.py | 174 |
1 files changed, 174 insertions, 0 deletions
diff --git a/app/model.py b/app/model.py new file mode 100644 index 0000000..467310b --- /dev/null +++ b/app/model.py @@ -0,0 +1,174 @@ +from math import floor +from typing import Tuple +from os.path import isfile +from pandas import read_csv +from dataclasses import dataclass +from importlib.resources import path +from random import getrandbits, randint +from sqlalchemy.orm import sessionmaker +from sqlalchemy import create_engine, not_ +from names import get_first_name, get_last_name +from match_up.data import ( + MIN_LEVEL, + MAX_LEVEL, + PLAYER_PER_COURT, + BASE, + GUEST_LEVELS, + GUESTS_PER_LEVEL, + Player, + DisplayPlayer, +) + + +@dataclass +class CourtList: + total: int = 12 + + def __post_init__(self) -> None: + self.courts = {i + 1: True for i in range(self.total)} + + def get_court_status(self, court_number: int) -> bool: + return self.courts[court_number] + + def toggle_court_status(self, court_number: int) -> bool: + new_status = not self.courts[court_number] + self.courts[court_number] = new_status + return new_status + + def get_courts(self) -> dict: + return self.courts + + def separate_courts(self) -> Tuple[dict, dict]: + active_courts = {k: v for k, v in self.courts.items() if v} + inactive_courts = {k: v for k, v in self.courts.items() if not v} + return active_courts, inactive_courts + + +def _init_guests(session) -> None: + for level_id, level in enumerate(GUEST_LEVELS): + for player_id in range(GUESTS_PER_LEVEL): + unique_id = (level_id * GUESTS_PER_LEVEL) + player_id + 1 + player = Player( + id=-unique_id, + first="Guest", + last=str(unique_id), + level=level, + status=False, + count=0, + ) + session.add(player) + session.commit() + + +def _init_database_from_csv(session, csv_location: str) -> None: + data = read_csv(csv_location) + session.query(Player).delete() + for _, entry in data.iterrows(): + player = Player( + id=entry["id"], + first=entry["first"], + last=entry["last"], + level=entry["skill"], + team=entry["team"], + status=False, + count=0, + ) + session.add(player) + session.commit() + _init_guests(session) + + +def _init_random_database(session, player_count: int = 80) -> None: + 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 + player = Player( + id=i, + first=get_first_name(), + last=get_last_name(), + level=randint(MIN_LEVEL, MAX_LEVEL), + team=team, + status=bool(getrandbits(1)), + count=0, + ) + session.add(player) + session.commit() + _init_guests(session) + + +class PlayerList: + + def __init__(self, csv_location=None) -> None: + with path("match_up", "players.db") as p: + self.db_location = p + db_file_found = True + if not isfile(self.db_location): + db_file_found = False + self._init_database() + if csv_location is not None: + _init_database_from_csv(self.session, csv_location) + else: + if not db_file_found: + _init_random_database(self.session) + + def _init_database(self): + engine = create_engine("sqlite:///" + str(self.db_location)) + BASE.metadata.create_all(bind=engine) + self.session = sessionmaker(bind=engine)() + + def get_player_status(self, id: int) -> bool: + return self.session.query(Player).filter(Player.id == id).first().status + + def toggle_player_status(self, id: int) -> DisplayPlayer: + player = self.session.query(Player).filter(Player.id == id).first() + player.status = not player.status + self.session.commit() + return DisplayPlayer(player) + + def get_all_players(self) -> Tuple[list[DisplayPlayer], list[DisplayPlayer]]: + all_players = [ + DisplayPlayer(p) + for p in self.session.query(Player) + .order_by(Player.first, Player.last) + .all() + ] + regulars = [player for player in all_players if player.first != "Guest"] + guests = [player for player in all_players if player.first == "Guest"] + guests.sort(key=lambda x: int(x.last)) + return regulars, guests + + def get_possible_game_count(self) -> int: + total_players = len(self.get_players()) + possible_games = floor(total_players / PLAYER_PER_COURT) + return possible_games + + def get_players(self) -> list[Player]: + active_players = ( + self.session.query(Player) + .order_by(Player.first, Player.last) + .filter(not_(Player.status == 0)) + .all() + ) + return active_players + + def increment_game_count(self, player_ids: list[int]): + players = self.session.query(Player).filter(Player.id.in_(player_ids)) + for player in players: + player.count += 1 + self.session.commit() + + def reset_game_count(self): + players = self.session.query(Player) + for player in players: + player.count = 0 + self.session.commit() + + def reset_all_players(self): + players = self.session.query(Player) + for player in players: + player.status = 0 + self.session.commit() |
