Skip to content

Commit

Permalink
ranking sim finished
Browse files Browse the repository at this point in the history
  • Loading branch information
Shom770 committed Mar 7, 2024
1 parent 7a9eea3 commit 4a59124
Show file tree
Hide file tree
Showing 7 changed files with 333 additions and 7 deletions.
1 change: 1 addition & 0 deletions src/page_managers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,5 @@
from .event_manager import EventManager
from .match_manager import MatchManager
from .picklist_manager import PicklistManager
from .ranking_simulator_manager import RankingSimulatorManager
from .team_manager import TeamManager
5 changes: 1 addition & 4 deletions src/page_managers/match_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -494,12 +494,9 @@ def generate_alliance_dashboard(self, team_numbers: list[int], color_gradient: l

# Colored metric displaying the chance of reaching the co-op bonus (1 amp cycle in 45 seconds + auto)
with reaches_coop_col:
coop_by_match = [self.calculated_stats.reaches_coop_bonus_by_match(team) for team in team_numbers]
possible_coop_combos = self.calculated_stats.cartesian_product(*coop_by_match)

colored_metric(
"Chance of Co-Op Bonus",
f"{len([combo for combo in possible_coop_combos if any(combo)]) / len(possible_coop_combos):.0%}",
f"{self.calculated_stats.chance_of_coop_bonus(team_numbers):.0%}",
background_color=color_gradient[3],
opacity=0.4,
border_opacity=0.9
Expand Down
144 changes: 144 additions & 0 deletions src/page_managers/ranking_simulator_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
"""Creates the `RankingSimulatorManager` class used to set up the Ranking Simulator page and its table."""
from collections import defaultdict

import streamlit as st
from numpy import logical_and
from pandas import DataFrame

from .page_manager import PageManager
from utils import (
bar_graph,
box_plot,
CalculatedStats,
colored_metric,
Criteria,
GeneralConstants,
GraphType,
line_graph,
multi_line_graph,
plotly_chart,
Queries,
retrieve_match_data,
retrieve_match_schedule,
retrieve_scouting_data,
retrieve_team_list,
scouting_data_for_team,
stacked_bar_graph,
colored_metric_with_two_values,
populate_missing_data
)


class RankingSimulatorManager(PageManager):
"""The ranking simulator page manager for the `Ranking Simulator` page."""
MATCHES_TO_START_FROM = 12

def __init__(self):
self.calculated_stats = CalculatedStats(
retrieve_scouting_data()
)
self.matches_played = retrieve_match_data()

def generate_input_section(self) -> str:
"""Generates the input section of the `Ranking Simulator` page."""
if (
not self.matches_played.empty
and (last_match_played := self.matches_played["match_number"].max()) >= self.MATCHES_TO_START_FROM
):
return st.slider("Match Number to Simulate From", self.MATCHES_TO_START_FROM, last_match_played)

return st.exception(
ValueError(
f"Come back to the simulator once at least {self.MATCHES_TO_START_FROM} matches have been played!"
)
)

def _generate_rankings(self, to_match: int) -> DataFrame:
"""Generates the rankings for a team given the matches that are specified."""
rankings = []
for team in retrieve_team_list():
red_match_data_for_team = self.matches_played[
logical_and(
self.matches_played["red_alliance"].str.contains(str(team)),
self.matches_played["match_number"] <= to_match
)
]
blue_match_data_for_team = self.matches_played[
logical_and(
self.matches_played["blue_alliance"].str.contains(str(team)),
self.matches_played["match_number"] <= to_match
)
]

matches_played = len(red_match_data_for_team) + len(blue_match_data_for_team)
average_rps = (
red_match_data_for_team["red_alliance_rp"].sum() + blue_match_data_for_team["blue_alliance_rp"].sum()
) / matches_played
average_coop = (
red_match_data_for_team["reached_coop"].sum() + blue_match_data_for_team["reached_coop"].sum()
) / matches_played
average_match_score = (
red_match_data_for_team["red_score"].sum() + blue_match_data_for_team["blue_score"].sum()
) / matches_played
rankings.append((team, average_rps, average_coop, average_match_score, matches_played)) # Sort orders

return DataFrame(
sorted(rankings, key=lambda ranking: ranking[1:-1], reverse=True),
columns=("team", "rp", "coop", "match_score", "matches_played")
)

def generate_simulated_rankings(self, to_match: int) -> None:
"""Generates the simulated rankings up to the match number requested."""
rankings = self._generate_rankings(to_match)
match_schedule = retrieve_match_schedule()
simulated_rankings = defaultdict(lambda: [[], [], [], 0])

teams = retrieve_team_list()
progress_bar = st.progress(0, text="Crunching the simulations...")

for idx, team in enumerate(teams, start=1):
matches_for_team = match_schedule[
match_schedule["red_alliance"]
.apply(lambda alliance: ",".join(map(str, alliance)))
.str.contains(str(team))
| match_schedule["blue_alliance"]
.apply(lambda alliance: ",".join(map(str, alliance)))
.str.contains(str(team))
]
matches_left_for_team = matches_for_team[
matches_for_team["match_key"].apply(
lambda key: int(key.replace("qm", "").replace("sf", "").replace("f", "").replace("m1", "").replace("m2", "").replace("m3", ""))
) >= to_match
][matches_for_team["match_key"].str.contains("qm")]

for _, row in matches_left_for_team.iterrows():
alliance = row["red_alliance"] if team in row["red_alliance"] else row["blue_alliance"]
opposing_alliance = row["blue_alliance"] if team in row["red_alliance"] else row["red_alliance"]

chance_of_coop, chance_of_melody, chance_of_ensemble = self.calculated_stats.chance_of_bonuses(alliance)
chance_of_winning, _, score, __ = self.calculated_stats.chance_of_winning(alliance, opposing_alliance)

total_rps = chance_of_melody + chance_of_ensemble + chance_of_winning * 2
simulated_rankings[team][0].append(total_rps)
simulated_rankings[team][1].append(chance_of_coop)
simulated_rankings[team][2].append(score)
simulated_rankings[team][3] += 1

progress_bar.progress(idx / len(teams), "Crunching the simulations...")

new_rankings = []

# Calculate new rankings with simulated rankings.
for team in teams:
_, rps, avg_coop, avg_match_score, matches_played = rankings[rankings["team"] == team].iloc[0]
total_matches_played = matches_played + simulated_rankings[team][3]
total_avg_rp = (rps * matches_played + sum(simulated_rankings[team][0])) / total_matches_played
total_avg_coop = (avg_coop * matches_played + sum(simulated_rankings[team][1])) / total_matches_played
total_avg_score = (avg_match_score * matches_played + sum(simulated_rankings[team][2])) / total_matches_played
new_rankings.append((team, total_avg_rp, total_avg_coop, total_avg_score))

ranking_df = DataFrame(
sorted(new_rankings, key=lambda ranking: ranking[1:], reverse=True),
columns=("Team", "Average Ranking Points", "Average Coopertition", "Average Match Score")
)
st.table(ranking_df.applymap(lambda value: f"{value:.2f}" if isinstance(value, float) else value))
22 changes: 22 additions & 0 deletions src/pages/6_Ranking_Simulation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
"""Creates the page for creating custom graphs in FalconVis."""

import streamlit as st
from page_managers import RankingSimulatorManager

# Configuration for Streamlit
st.set_page_config(
layout="wide",
page_title="Ranking Simulator",
page_icon="❓",
)
ranking_simulator_manager = RankingSimulatorManager()

if __name__ == '__main__':
# Write the title of the page.
st.write("# Ranking Simulation")

# Add the input section of the page.
match_chosen = ranking_simulator_manager.generate_input_section()

# Display the simulated rankings.
ranking_simulator_manager.generate_simulated_rankings(match_chosen)
127 changes: 125 additions & 2 deletions src/utils/calculated_stats.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
import numpy as np
from numpy import percentile
from pandas import DataFrame, Series, isna
from scipy.integrate import quad
from scipy.stats import norm

from .constants import Criteria, Queries
from .functions import _convert_to_float_from_numpy_type, scouting_data_for_team, retrieve_team_list, retrieve_pit_scouting_data
Expand Down Expand Up @@ -76,6 +78,8 @@ def points_contributed_by_match(self, team_number: int, mode: str = "") -> Serie
return total_auto_points
elif mode == Queries.TELEOP:
return total_teleop_points
elif mode == Queries.ENDGAME:
return total_endgame_points

return (
total_auto_points
Expand Down Expand Up @@ -124,7 +128,7 @@ def average_potential_amplification_periods(self, team_number: int) -> float:
"""
return self.potential_amplification_periods_by_match(team_number).mean()

def cycles_by_match(self, team_number: int, mode: str) -> Series:
def cycles_by_match(self, team_number: int, mode: str = None) -> Series:
"""Returns the cycles for a certain mode (autonomous/teleop) in a match
The following custom graphs are supported with this function:
Expand All @@ -140,8 +144,13 @@ def cycles_by_match(self, team_number: int, mode: str) -> Series:

if mode == Queries.AUTO:
return team_data[Queries.AUTO_SPEAKER] + team_data[Queries.AUTO_AMP]
else:
elif mode == Queries.TELEOP:
return team_data[Queries.TELEOP_SPEAKER] + team_data[Queries.TELEOP_AMP] + team_data[Queries.TELEOP_TRAP]
else:
return (
team_data[Queries.AUTO_SPEAKER] + team_data[Queries.AUTO_AMP]
+ team_data[Queries.TELEOP_SPEAKER] + team_data[Queries.TELEOP_AMP] + team_data[Queries.TELEOP_TRAP]
)

def cycles_by_structure_per_match(self, team_number: int, structure: str | tuple) -> Series:
"""Returns the cycles for a certain structure (auto speaker, auto amp, etc.) in a match
Expand Down Expand Up @@ -368,3 +377,117 @@ def driving_index(self, team_number: int) -> float:
self.average_cycles(team_number, Queries.TELEOP)
* 0 if isna(counter_defense_skill) else counter_defense_skill
)

# Methods for ranking simulation
def chance_of_coop_bonus(self, alliance: list[int]) -> float:
"""Determines the chance of the coop bonus using all possible permutations with an alliance.
:param alliance: The three teams on the alliance.
"""
coop_by_match = [self.reaches_coop_bonus_by_match(team) for team in alliance]
possible_coop_combos = self.cartesian_product(*coop_by_match)
return len([combo for combo in possible_coop_combos if any(combo)]) / len(possible_coop_combos)

def chance_of_bonuses(self, alliance: list[int]) -> tuple[float, float, float]:
"""Determines the chance of the coopertition bonus, the melody bonus and the ensemble bonus using all possible permutations with an alliance.
:param alliance: The three teams on the alliance.
"""
chance_of_coop = self.chance_of_coop_bonus(alliance)
cycles_for_alliance = [self.cycles_by_match(team) for team in alliance]

# Melody RP calculations
possible_cycle_combos = self.cartesian_product(*cycles_for_alliance, reduce_with_sum=True)
chance_of_reaching_15_cycles = (
len([combo for combo in possible_cycle_combos if combo >= 15]) / len(possible_cycle_combos)
)
chance_of_reaching_18_cycles = (
len([combo for combo in possible_cycle_combos if combo >= 18]) / len(possible_cycle_combos)
)

# Ensemble RP calculations
endgame_points_by_team = [self.points_contributed_by_match(team, Queries.ENDGAME) for team in alliance]
possible_endgame_combos = self.cartesian_product(*endgame_points_by_team, reduce_with_sum=True)
chance_of_reaching_10_points = (
len([combo for combo in possible_endgame_combos if combo >= 10]) / len(possible_endgame_combos)
)

ability_to_climb_by_team = [True in self.stat_per_match(team, Queries.CLIMBED_CHAIN) for team in alliance]

if ability_to_climb_by_team.count(True) >= 2: # If teams can climb
chance_of_ensemble_rp = chance_of_reaching_10_points
else:
chance_of_ensemble_rp = 0 # No chance that they can get the RP even if 10 points can be reached.

return (
chance_of_coop,
chance_of_reaching_18_cycles * (1 - chance_of_coop) + chance_of_reaching_15_cycles * chance_of_coop,
chance_of_ensemble_rp
)

def chance_of_winning(self, alliance_one: list[int], alliance_two: list[int]) -> tuple:
"""Returns the chance of winning between two alliances using integrals."""
alliance_one_points = [
self.points_contributed_by_match(team)
for team in alliance_one
]
alliance_two_points = [
self.points_contributed_by_match(team)
for team in alliance_two
]

# Calculate mean and standard deviation of the point distribution of the red alliance.
alliance_one_std = (
sum(
[
np.std(team_distribution) ** 2
for team_distribution in alliance_one_points
]
)
** 0.5
)
alliance_one_mean = sum(
[
np.mean(team_distribution)
for team_distribution in alliance_one_points
]
)

# Calculate mean and standard deviation of the point distribution of the blue alliance.
alliance_two_std = (
sum(
[
np.std(team_distribution) ** 2
for team_distribution in alliance_two_points
]
)
** 0.5
)
alliance_two_mean = sum(
[
np.mean(team_distribution)
for team_distribution in alliance_two_points
]
)

# Calculate mean and standard deviation of the point distribution of red alliance - blue alliance
compared_std = (alliance_one_std ** 2 + alliance_two_std ** 2) ** 0.5
compared_mean = alliance_one_mean - alliance_two_mean

# Use sentinel value if there isn't enough of a distribution yet to determine standard deviation.
if not compared_std and compared_mean:
compared_std = abs(compared_mean)
elif not compared_std:
compared_std = 0.5

compared_distribution = norm(loc=compared_mean, scale=compared_std)

# Calculate odds of red/blue winning using integrals.
odds_of_red_winning = quad(
lambda x: compared_distribution.pdf(x), 0, np.inf
)[0]
odds_of_blue_winning = quad(
lambda x: compared_distribution.pdf(x), -np.inf, 0
)[0]

return odds_of_red_winning, odds_of_blue_winning, alliance_one_mean, alliance_two_mean
3 changes: 2 additions & 1 deletion src/utils/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ class GeneralConstants:
"Average Teleop Cycles",
"Average Auto Cycles"
]
SECONDS_TO_CACHE = 60 * 4
SECONDS_TO_CACHE = 60 * 3
PRIMARY_COLOR = "#EFAE09"
AVERAGE_FOUL_RATE = 1.06

Expand Down Expand Up @@ -110,6 +110,7 @@ class Queries:
# Modes
AUTO = "Auto"
TELEOP = "Teleop"
ENDGAME = "Endgame"

# Custom graph keywords
ONE_TEAM_KEYWORD = "Used for custom graphs with one team."
Expand Down
Loading

0 comments on commit 4a59124

Please sign in to comment.