|
import json |
|
from dataclasses import asdict |
|
import pandas as pd |
|
import streamlit as st |
|
from postgrest.exceptions import APIError |
|
|
|
from config import DEFAULT_ICON |
|
from shared_page import common_page_config |
|
|
|
from data_storage import get_all_users, get_all_rosters |
|
from domain.playoffs import CURRENT_PLAYOFF_WEEK, PLAYOFF_WEEK_TO_NAME |
|
from format_player_html import ( |
|
get_user_html_str, |
|
) |
|
from load_options import get_map_week_player_id_option, PlayerOption |
|
from stats import get_scores_map |
|
|
|
|
|
DEV_DUMP_ROSTER_JSON = False |
|
|
|
|
|
def get_users_df(): |
|
columns = ["user_id", "name"] |
|
all_users = pd.DataFrame(get_all_users(st.session_state["db_client"], columns_included=columns), columns=columns) |
|
return all_users |
|
|
|
|
|
@st.cache_data(ttl=60 * 1) |
|
def load_masked_rosters() -> dict[int, dict[str, PlayerOption]]: |
|
options_map = get_map_week_player_id_option() |
|
roster_user_position_map: dict[int, dict[str, PlayerOption]] = {} |
|
for roster_slot_map in get_all_rosters(st.session_state["db_client"]): |
|
position_id = str(roster_slot_map["position_id"]) |
|
player_id = str(roster_slot_map["player_id"]) |
|
user_id = int(roster_slot_map["user_id"]) |
|
if user_id not in roster_user_position_map: |
|
roster_user_position_map[user_id] = {} |
|
week = int(position_id[0]) |
|
|
|
player = PlayerOption.empty_player(week=week) |
|
if selected_player := options_map[week].get(player_id): |
|
if selected_player.is_locked(): |
|
player = selected_player |
|
else: |
|
player = PlayerOption.hidden_player(week=week, position=selected_player.position) |
|
roster_user_position_map[user_id][position_id] = player |
|
|
|
return roster_user_position_map |
|
|
|
|
|
@st.cache_data(ttl=60 * 1) |
|
def get_roster_multipliers(roster_map: dict[int, dict[str, PlayerOption]]) -> dict[int, dict[int, dict[str, int]]]: |
|
"""Map of user -> week -> player_id -> multiplier""" |
|
multiplier_map: dict[int, dict[int, dict[str, int]]] = {} |
|
for user_id, user_roster_map in roster_map.items(): |
|
user_multipliers: dict[int, dict[str, int]] = {} |
|
|
|
for position_id, player in sorted(user_roster_map.items(), key=lambda x: x[0][0]): |
|
if not player.gsis_id: |
|
|
|
continue |
|
week = int(position_id.split("-", 1)[0]) |
|
if week not in user_multipliers: |
|
user_multipliers[week] = {} |
|
|
|
player_previous_multiplier = user_multipliers.get(week - 1, {}).get(player.gsis_id, 0) |
|
player_multiplier = player_previous_multiplier + 1 |
|
user_multipliers[week][player.gsis_id] = player_multiplier |
|
|
|
multiplier_map[user_id] = user_multipliers |
|
return multiplier_map |
|
|
|
|
|
def assemble_user_scores( |
|
player_scores_map: dict[int, dict[str, float]], |
|
roster_map: dict[int, dict[str, PlayerOption]], |
|
multiplier_map: dict[int, dict[int, dict[str, int]]], |
|
) -> dict[int, dict[int, float]]: |
|
week_user_score_map: dict[int, dict[int, float]] = {w: {} for w in player_scores_map.keys()} |
|
user_totals: dict[int, float] = {} |
|
for user_id, user_roster_map in roster_map.items(): |
|
user_score_map: dict[int, float] = {w: 0.0 for w in player_scores_map.keys()} |
|
for roster_key, player_id in user_roster_map.items(): |
|
week = int(roster_key[0]) |
|
player_score = player_scores_map.get(week, {}).get(player_id.gsis_id, 0.0) |
|
multiplier = float(multiplier_map.get(user_id, {}).get(week, {}).get(player_id.gsis_id, 1)) |
|
user_score_map[week] += round(player_score * multiplier, 0) |
|
for week, week_score in user_score_map.items(): |
|
week_user_score_map[week][user_id] = week_score |
|
if user_id not in user_totals: |
|
user_totals[user_id] = 0.0 |
|
user_totals[user_id] += week_score |
|
week_user_score_map[5] = user_totals |
|
return week_user_score_map |
|
|
|
|
|
def display_masked_rosters(week: int): |
|
rosters = load_masked_rosters() |
|
if DEV_DUMP_ROSTER_JSON: |
|
rosters_serial: dict[int, dict[str, dict]] = {} |
|
for k_user, v in rosters.items(): |
|
rosters_serial[k_user] = {} |
|
for k_pos_id, player in v.items(): |
|
player_dict = asdict(player) |
|
player_dict.pop("gametime") |
|
rosters_serial[k_user][k_pos_id] = player_dict |
|
with open("rosters.json", "w", encoding="utf-8") as f: |
|
json.dump(rosters_serial, f, ensure_ascii=False, indent=4) |
|
multipliers = get_roster_multipliers(rosters) |
|
users = get_users_df() |
|
player_scores = get_scores_map() |
|
user_scores = assemble_user_scores(player_scores, rosters, multipliers) |
|
|
|
scoreboard_str = "" |
|
|
|
scoreboard_str += """<div className="scoreboard">""" |
|
sorted_user_rows = sorted( |
|
users.itertuples(), |
|
key=lambda x: user_scores.get(week, {}).get(x.user_id, 0.0), |
|
reverse=True, |
|
) |
|
for i, row in enumerate(sorted_user_rows): |
|
user_score = user_scores.get(week, {}).get(row.user_id, 0.0) |
|
user_roster_map = rosters.get(row.user_id, {}) |
|
user_place = i + 1 |
|
scoreboard_str += get_user_html_str( |
|
week, row.name, user_roster_map, user_score, user_place, multipliers.get(row.user_id, {}) |
|
) |
|
|
|
scoreboard_str += """</div>""" |
|
|
|
st.markdown(scoreboard_str, unsafe_allow_html=True) |
|
|
|
|
|
def display_rosters(): |
|
st.markdown("<h2>Rosters</h2>", unsafe_allow_html=True) |
|
options = list(PLAYOFF_WEEK_TO_NAME.keys()) |
|
default_selection = options.index(CURRENT_PLAYOFF_WEEK) |
|
week_selected = st.selectbox( |
|
"Week", |
|
options=options, |
|
index=default_selection, |
|
key="roster_week_select", |
|
format_func=lambda x: PLAYOFF_WEEK_TO_NAME[x], |
|
) |
|
|
|
display_masked_rosters(week_selected) |
|
|
|
|
|
def get_page(): |
|
page_title = "Pool Scoreboard" |
|
st.set_page_config(page_title=page_title, page_icon=DEFAULT_ICON, layout="wide", initial_sidebar_state="collapsed") |
|
common_page_config() |
|
|
|
st.title(page_title) |
|
try: |
|
display_rosters() |
|
except APIError: |
|
try: |
|
display_rosters() |
|
except Exception: |
|
st.write("Sorry error occurred loading scoreboard. Please try refreshing page.") |
|
st.stop() |
|
|
|
|
|
if __name__ == "__main__": |
|
get_page() |
|
|
