aboutsummaryrefslogtreecommitdiff
path: root/app/model.py
diff options
context:
space:
mode:
authorKaran Jayachandra <mail@karanjayachandra.com>2025-07-06 12:58:07 +0200
committerKaran Jayachandra <mail@karanjayachandra.com>2025-07-06 12:58:07 +0200
commit0da8dc16fb26e9df5d0efd7c7a463e354d8e00cf (patch)
treed016270ebf8e6e9fef499da5767c99b656c75075 /app/model.py
parent155e77d22f24d1984476144f48bf173044d089a0 (diff)
Transitioned back to a traditional flask app
Diffstat (limited to 'app/model.py')
-rw-r--r--app/model.py174
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()