from functools import reduce
import operator
import sqlite3
import time
from typing import Callable, TypeVar
import pandas as pd
import os
import json
import numpy as np
from src.lib.bs import eval_beakscript
from src.lib.data_config import FANCY_FIL
from src import apputils
from src.apputils import PathUtils
from collections import defaultdict
from collections.abc import Iterable
import logging
from scipy.linalg import svd
import statbotics
import asyncio
logger = logging.getLogger(__name__)
[docs]
class TeamStruct:
""" Represents a dictionary of lists of data for a team across matches """
def __init__(self) -> None:
self.data = {}
[docs]
def extend_data(self, data) -> None:
"""Append a match of data to the team. Everything is procedural because of field-config"""
merged = defaultdict(list)
for d in [self.data, data]:
for k, v in d.items():
try:
merged[k].extend(v)
except:
merged[k].append(v)
self.data = dict(merged)
[docs]
def output_dict(self, config):
"""Get all of the team data as a dictionary, unwrapping DataFields to get averages and such"""
data = {}
# the | operator for dicts in python combines dicts with different entries (OR's them)
for field in config["svd"]:
data |= (
DataField(field["name"], self.data[field["name"]], []).objectify()
| DataField(
f"{field["name"]} Variance",
self.data[f"{field["name"]} Variance"],
[],
).objectify()
)
for field in config["teams"]:
data |= DataField(
field["name"], self.data[field["name"]], field["filters"]
).objectify()
return data
[docs]
class DataField:
""" Represents a list of data for a field """
def __init__(self, name, data, filters) -> None:
self.name = name
self.filters = filters
self.data = data
[docs]
def average(self) -> float | str:
"""returns the average of the dataset"""
return (
# x̄ = Σx/|x|
round(sum(self.data) / len(self.data), 3)
if len(self.data) > 0
else "N/A"
)
[docs]
def max(self):
"""returns the max of the dataset"""
return max(self.data) if len(self.data) > 0 else "N/A"
[docs]
def filter(self) -> float:
"""returns the average of the dataset, filtered by median += 2 * MAD"""
return float(
np.around(np.mean(Processor.mad_filter(np.array(self.data))), 3)
) # around is just round
calc_map = {"avg": average, "max": max, "fil": filter}
[docs]
def objectify(self):
"""calculates the applicable filters to serialize the data in this field into a dict"""
data = {}
if len(self.filters) > 0:
for fil in self.filters:
data[FANCY_FIL[fil] + " " + self.name] = self.calc_map[fil](
self
) # call the cooresponding function to apply the filter
elif (
isinstance(self.data, list) and len(self.data) == 1
): # if no filters, passthrough
data[self.name] = self.data[0]
else:
data[self.name] = self.data # if the data is just a float, use it
return data
[docs]
class MatchStruct:
""" Represents a dictionary of data for teams in a match """
def __init__(self) -> None:
self.teams = {}
[docs]
def add_team_data(self, team, data) -> None:
"""add data from a team for this match"""
self.teams.setdefault(team, data) # use setdefault to create the team if it DNE
[docs]
def output_dict(self, config):
"""serialize the struct into a dictionary"""
data = []
for team in self.teams.keys():
d2 = {"Team": team} # add an entry for what team it is
for field in config["matches"]:
d2 |= DataField(
field["name"], self.teams[team][field["name"]], field["filters"]
).objectify()
data.append(d2)
return data
# Constrained type alias
ET = TypeVar("ET", bound=Callable)
[docs]
class Event[ET]:
""" Represents a FIFO sequence of events to be run asynchrounously from the main thread (but sync with other events) that should be triggered upon an occurance """
event_progress = {}
current_event = ""
current_event_len = 0
current_handle_index_progress: float = 0.0
global_lock = asyncio.Lock()
[docs]
def get_event_progress():
""" Returns a dictionary of events and information about them """
keys = list(Event.event_progress.keys())
for event in keys:
if (
"time_complete" in Event.event_progress[event]
and time.time() - Event.event_progress[event]["time_complete"] > 5
):
Event.event_progress.pop(event)
if event == Event.current_event:
Event.current_event = ""
if Event.current_event != "":
return Event.event_progress | {
Event.current_event: {
"prog": round(
Event.event_progress[Event.current_event]["prog"]
+ Event.current_handle_index_progress / Event.current_event_len,
2,
),
"step": Event.event_progress[Event.current_event]["step"],
"status": Event.event_progress[Event.current_event]["status"],
}
}
return Event.event_progress
def __init__(self, name: str = ""):
self._handlers = []
self.task: asyncio.Task | None = None
self.name = name
def __iadd__(self, f: ET | list[ET]):
""" Append a function or list of functions to the event queue """
if apputils.is_iterable(f):
self._handlers.extend(f)
else:
self._handlers.append(f)
return self
def __isub__(self, f: ET | list[ET]):
""" remove a function or list of functions from the event queue """
if apputils.is_iterable(f):
for f2 in f:
self._handlers.remove(f2)
else:
self._handlers.remove(f)
return self
[docs]
async def fire(self, logging_callback: Callable[[str], None], *args, **kwargs):
""" Fires the event logging to `logging_callback` with the inputted args, which should match the type specification of the Event type parameter """
if self.task and not self.task.done():
logging_callback(
f"Task {self.name} already running. Cancelling and restarting..."
)
self.task.cancel()
try:
await self.task
except asyncio.CancelledError:
pass
self.event_progress[self.name] = {"prog": 0.0, "step": "", "status": "queued"}
async def fire_task():
logging_callback(f"Task for {self.name} queued, waiting for lock.")
async with Event.global_lock:
logging_callback(f"{self.name} has lock.")
Event.current_event = self.name
Event.current_event_len = len(self._handlers)
try:
for i, f in enumerate(self._handlers):
if logging_callback != None:
logging_callback(
f"{self.name} [{i + Event.current_handle_index_progress}/{len(self._handlers)}]: {f.__name__}"
)
Event.event_progress[self.name] = {
"prog": i / len(self._handlers),
"step": f.__name__,
"status": "in-progress",
}
try:
f(*args, **kwargs)
except Warning as w:
logging_callback(
f"Warning in {f.__name__}: {apputils.exception_format(w)}"
)
Event.event_progress[self.name]["prog"] = 1.0
Event.event_progress[self.name]["status"] = "done"
Event.event_progress[self.name]["time_complete"] = time.time()
Event.current_handle_index_progress = 0
except asyncio.CancelledError:
Event.event_progress[self.name]["prog"] = 0.0
Event.event_progress[self.name]["status"] = "cancelled"
logging_callback(f"Event {self.name} cancelled.")
except Exception as e:
logging_callback(
f"Error in {self.name}: {apputils.exception_format(e)}"
)
Event.event_progress[self.name][
"error"
] = f"{apputils.exception_format(e)}"
await asyncio.sleep(0)
self.task = asyncio.create_task(fire_task(), name=self.name)
return None
[docs]
class ObjectHolder[T]:
"""Simple wrapper for pass-by-reference in functions"""
def __init__(self, object: T):
self.obj = object
[docs]
class Processor:
"""Main class of data calculation, handles all calculation basically"""
NUM_TABLES = 5
def __init__(
self, disable_last_opr, tba_key, year, config_data, load_from_cache=True
) -> None:
self.disable_last_opr = disable_last_opr
self.tba_data_static = apputils.TBADataStatic()
self.tba_data_dyn = apputils.TBADataDynamic()
self.event_key = ""
self.tba_key, self.year = tba_key, year
self.__teams: dict[str, TeamStruct] = {}
self.__sb = statbotics.Statbotics()
self.has_sched_data = False
self.__sb_epas = {}
self.__sb_matches = []
self.__sb_rank_points = []
self.__matches = {}
self.config_data = config_data
self.__load_in_event_data = Event[Callable[[], None]]("Load in event data")
self.__periodic_calls = Event[Callable[[], None]]("Periodic Fetch Routine")
self.__data_processing_routine = Event[
Callable[[ObjectHolder[pd.DataFrame]], None]
]("Processing Routine")
self.__post_process_routine = Event[Callable[[], None]](
"Post-processing Routine"
)
self.__load_in_event_data += [
self.__get_year_sb_rankpoint_keys,
self.__create_sql_tables,
self.__load_remote_data_static,
self.__write_match_schedule_file,
self.__write_teams_file,
]
self.__periodic_calls += [
self.__load_remote_data_dyn,
self.__write_statbotics_analytics,
self.__write_statbotics_epa,
self.__write_team_tba_data_file,
]
self.__data_processing_routine += [
self.__pre_filter_chunk,
self.__pre_process_chunk,
self.__drop_duplicates_chunk,
self.__filter_chunk,
self.__compute_fields_chunk,
self.__build_teams_chunk,
self.__svd_data_chunk,
self.__reduce_df,
self.__team_proc_chunk,
self.__match_proc_chunk,
self.__write_output_file,
]
self.__post_process_routine += [
self.__write_match_predictions_file,
self.__write_match_depth_predictions,
self.__write_match_fields,
self.__write_team_fields,
self.__write_other_metrics,
]
self.sql_fields_match_base = [
"Red_1",
"Red_2",
"Red_3",
"Blue_1",
"Blue_2",
"Blue_3",
"Predict_R1_Score",
"Predict_R2_Score",
"Predict_R3_Score",
"Predict_B1_Score",
"Predict_B2_Score",
"Predict_B3_Score",
"Predict_Red_Score",
"Predict_Blue_Score",
"Predict_Winner",
"Statbotics_winner",
"Statbotics_red_win_prob",
"Statbotics_red_Score",
"Statbotics_blue_Score",
]
self.sql_fields = {
"matches": self.sql_fields_match_base,
"matches_depth_predictions": [
"attr_name",
"attr_value",
],
"matches_fields": [
"field_name",
"field_value",
],
"teams": [
"EPA",
"Rank",
"Average_RP",
"OPR",
"Last_OPR",
"Country",
"State",
"City",
"Name",
"School",
"RookieYear",
"PostalCode",
"Website",
],
"teams_fields": [
"field_name",
"field_value",
],
}
tables_exist = False
if self.__check_tables_exist():
logger.info("Static match data preset, loading into memory...")
self.__read_static_tba_info()
logger.info("Static match data preset, loaded")
tables_exist = True
else:
logger.warning(
"Warning: tables do not exist. Until an event is loaded there will be no data"
)
if load_from_cache:
self.__check_load_event_key_cache(tables_exist)
def __check_tables_exist(self) -> bool:
""" Check whether the correct number of tables exists in the db """
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
return len(cursor.fetchall()) == Processor.NUM_TABLES
def __check_load_event_key_cache(self, tables_exist: bool):
""" Checks for and loads a cached event key and loads in table data if there is any """
if os.path.exists(PathUtils.file_set.last_event_cache):
with open(PathUtils.file_set.last_event_cache, "r") as r:
self.event_key = r.read().strip()
if not self.has_sched_data and tables_exist:
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
cursor = conn.cursor()
cursor.execute("SELECT Team FROM teams")
self.has_sched_data = (
len(cursor.fetchall()) > 0
) # if teams are loaded in there's at least scouted data
def __get_year_sb_rankpoint_keys(self):
""" Gets the statbotics names for the rankpoints of a current year """
def make_table_headers(rp):
return [f"Statbotics_red_{rp}", f"Statbotics_blue_{rp}"]
self.__sb_rank_points = [
x
for x in self.__sb.get_year(int(self.year))["percentiles"].keys()
if "rp" in x
]
self.sql_fields["matches"] = self.sql_fields_match_base + [
th for rp in self.__sb_rank_points for th in make_table_headers(rp)
]
def __create_sql_tables(self):
""" Creates sql tables for matches, matches_depth_predictions, matches_fields teams, and teams_fields """
def make_table_headers(rp):
return [f"Statbotics_red_{rp}", f"Statbotics_blue_{rp}"]
# Create Tables
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
conn.executescript(f"""
DROP TABLE IF EXISTS matches;
CREATE TABLE matches (
MatchIdx Int PRIMARY KEY,
Match TEXT,
Red_1 INT,
Red_2 INT,
Red_3 INT,
Blue_1 INT,
Blue_2 INT,
Blue_3 INT,
Predict_R1_Score REAL,
Predict_R2_Score REAL,
Predict_R3_Score REAL,
Predict_B1_Score REAL,
Predict_B2_Score REAL,
Predict_B3_Score REAL,
Predict_Red_Score REAL,
Predict_Blue_Score REAL,
Predict_Winner TEXT,
Statbotics_winner TEXT,
Statbotics_red_win_prob REAL,
Statbotics_red_Score REAL,
Statbotics_blue_Score REAL,
{",\n".join([f"{th} REAL" for rp in self.__sb_rank_points for th in make_table_headers(rp)])}
);
DROP TABLE IF EXISTS matches_depth_predictions;
CREATE TABLE matches_depth_predictions (
match_key TEXT NOT NULL REFERENCES matches(Match),
team_key INT,
attr_name TEXT,
attr_value TEXT,
PRIMARY KEY (match_key, team_key, attr_name)
);
DROP TABLE IF EXISTS matches_fields;
CREATE TABLE matches_fields (
match_key TEXT NOT NULL REFERENCES matches(Match),
team_key INT,
field_name TEXT,
field_value TEXT,
PRIMARY KEY (match_key, team_key, field_name)
);
DROP TABLE IF EXISTS teams;
CREATE TABLE teams (
Team INT PRIMARY KEY,
EPA REAL,
Rank REAL,
Average_RP REAL,
OPR REAL,
Last_OPR REAL,
Country TEXT,
State TEXT,
City TEXT,
Name TEXT,
School TEXT,
RookieYear INT,
PostalCode TEXT,
Website TEXT
);
DROP TABLE IF EXISTS teams_fields;
CREATE TABLE teams_fields (
team_key INT NOT NULL REFERENCES teams(Team),
field_name TEXT,
field_value TEXT,
PRIMARY KEY (team_key, field_name)
);
""")
conn.commit()
def __read_static_tba_info(self):
""" Reads team info, event schedule """
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
# LOAD TEAMS + Last_OPR
cursor = conn.execute(f"SELECT * FROM teams")
rows = cursor.fetchall()
teams = []
team_info = {}
last_oprs = {}
oprs, epas = {}, {}
fields = ["Team"] + self.sql_fields["teams"]
for row in rows:
row_dict = dict(zip(fields, row))
teams.append(row_dict["Team"])
team_info[row_dict["Team"]] = {
"Country": row_dict["Country"],
"State": row_dict["State"],
"City": row_dict["City"],
"Name": row_dict["Name"],
"School": row_dict["School"],
"RookieYear": row_dict["RookieYear"],
"PostalCode": row_dict["PostalCode"],
"Website": row_dict["Website"],
}
last_oprs[row_dict["Team"]] = row_dict["Last_OPR"]
oprs[row_dict["Team"]] = row_dict["OPR"]
epas[row_dict["Team"]] = row_dict["EPA"]
(
self.tba_data_static.teams,
self.tba_data_static.team_info,
self.tba_data_static.oprs,
self.tba_data_dyn.oprs,
self.__sb_epas,
) = (teams, team_info, last_oprs, oprs, epas)
# LOAD SCHEDULE
fields = ["MatchIdx", "Match"] + self.sql_fields["matches"]
cursor = conn.execute(f"SELECT * FROM matches")
rows = cursor.fetchall()
matches = []
for row in rows:
row_dict = dict(zip(fields, row))
if self.event_key.strip() == "":
self.event_key = row_dict["Match"].split("_")[0]
red = [row_dict[f"Red_{i+1}"] for i in range(3)]
blue = [row_dict[f"Blue_{i+1}"] for i in range(3)]
entry = {
"k": row_dict["Match"],
"r": list(map(str, red)),
"b": list(map(str, blue)),
}
matches.append(entry)
self.tba_data_static.schedule = matches
if len(rows) > 0:
self.has_sched_data = True
def __load_remote_data_static(self):
""" Fetches :class:`src.apputils.TBADataStatic` data"""
if self.event_key.strip() == "":
logger.error(
"Error, cannot process. No event is currently loaded in. Please load one in from the settings page."
)
return
try:
logger.info("Loading TBA data...")
self.tba_data_static = apputils.load_tba_data_static(
self.event_key, self.tba_key, self.year, self.disable_last_opr
)
Event.current_handle_index_progress = 0.5
self.has_sched_data = True
logger.info("Loading TBA images...")
try:
apputils.get_tba_images(
self.tba_key, self.year, self.tba_data_static.teams
)
Event.current_handle_index_progress = 1.0
except Exception as e:
logger.warning(f"Error fetching images: {apputils.exception_format(e)}")
except Warning as w:
logger.warning(
f"Warning fetching remote data: {apputils.exception_format(w)}"
)
except Exception as e:
logger.error(f"Error fetching remote data: {apputils.exception_format(e)}")
def __load_remote_data_dyn(self):
""" Fetches :class:`TBADataDynamic` data, loads statbotics epas, and loads statbotics match data """
if self.event_key.strip() == "":
logger.error(
"Error, cannot process. No event is currently loaded in. Please load one in from the settings page."
)
return
try:
logger.info("Loading TBA data...")
self.tba_data_dyn = apputils.load_tba_data_dynamic(
self.event_key,
self.tba_key,
self.config_data,
self.tba_data_static.teams,
)
Event.current_handle_index_progress = 0.33
logger.info("Loading Statbotics EPAs...")
self.__sb_epas = {
s["team"]: round(s["epa"]["total_points"]["mean"], 1)
for s in self.__sb.get_team_events(
event=self.event_key, limit=1000, fields=["team", "epa"]
)
}
Event.current_handle_index_progress = 0.66 # no
logger.info("Loading Statbotics match data...")
self.__sb_matches = self.__sb.get_matches(event=self.event_key)
Event.current_handle_index_progress = 1.0
except Warning as w:
logger.warning(
f"Warning fetching remote data: {apputils.exception_format(w)}"
)
except Exception as e:
logger.error(f"Error fetching remote data: {apputils.exception_format(e)}")
def __write_match_schedule_file(self):
""" Writes match schedule to the matches table """
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
matches = []
for i, match in enumerate(self.tba_data_static.schedule):
matches.append(
{
"MatchIdx": i + 1,
"Match": match["k"],
}
| {f"Red_{i + 1}": v for i, v in enumerate(match["r"])}
| {f"Blue_{i + 1}": v for i, v in enumerate(match["b"])}
)
all_fields = self.sql_fields["matches"]
Event.current_handle_index_progress = 0.5
if len(matches) <= 0:
logger.warning("Warning, no match schedule")
return
fields_not_writing_to = [
x for x in all_fields if x not in matches[0].keys()
]
for match in matches:
conn.execute(
f"""
INSERT OR REPLACE INTO matches (MatchIdx, Match, {", ".join(all_fields)}) VALUES (?, ?, {", ".join(['?'] * len(all_fields))})
""",
[
match["MatchIdx"],
match["Match"],
match["Red_1"],
match["Red_2"],
match["Red_3"],
match["Blue_1"],
match["Blue_2"],
match["Blue_3"],
]
+ [None] * len(fields_not_writing_to),
)
conn.commit()
Event.current_handle_index_progress = 1.0
def __write_teams_file(self):
""" Write teams to the teams table """
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
all_fields = self.sql_fields["teams"]
for i, team in enumerate(self.tba_data_static.teams):
Event.current_handle_index_progress = i / len(
self.tba_data_static.teams
)
data = []
for field in all_fields:
if field in list(self.tba_data_static.team_info.values())[0].keys():
data.append(self.tba_data_static.team_info[team][field])
else:
data.append(None)
conn.execute(
f"""
INSERT OR REPLACE INTO teams (Team, {", ".join(all_fields)}) VALUES (?, {", ".join(['?'] * len(all_fields))})
""",
[team] + data,
)
conn.commit()
def __write_team_tba_data_file(self):
""" write tba statistics (rank/rp/opr/etc.) for each team into the teams table """
df = []
for team in self.tba_data_static.teams:
rank, avg_rp = (
self.tba_data_dyn.ranks[team]
if team in self.tba_data_dyn.ranks
else (None, None)
)
df.append(
{
"Team": team,
"Rank": rank,
"Average_RP": avg_rp,
"OPR": (
self.tba_data_dyn.oprs[team]
if team in self.tba_data_dyn.oprs
else None
),
"Last_OPR": (
self.tba_data_static.oprs[team]
if team in self.tba_data_static.oprs
else None
),
}
)
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
for team in df:
conn.execute(
f"""
UPDATE teams
SET Rank = ?, Average_RP = ?, OPR = ?, Last_OPR = ?
WHERE Team = ?
""",
(
team["Rank"],
team["Average_RP"],
team["OPR"],
team["Last_OPR"],
team["Team"],
),
)
conn.commit()
Event.current_handle_index_progress = 1.0
def __write_team_fields(self):
""" write the teams_fields table """
df = []
def get_coprs_safe(team: int) -> dict:
if team in self.tba_data_dyn.copr:
return self.tba_data_dyn.copr[team]
else:
return (
{c: 0.0 for c in self.config_data["copr"]}
if "copr" in self.config_data
else {}
)
for (
k,
v,
) in (
self.__teams.items()
): # _teams is a dict with team: TeamStruct (ex. {422: TeamData()})
# bind each team to the dict serialization of its cooresponding struct
df.append(
{"Team": k} | v.output_dict(self.config_data) | get_coprs_safe(int(k))
)
if len(df) <= 0:
logger.warning("No team fields")
return
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
fields_to_write = list(df[0].keys())
fields_to_write.remove("Team")
for team in df:
for field in fields_to_write:
try:
conn.execute(
f"""
INSERT OR REPLACE INTO teams_fields (team_key, field_name, field_value) VALUES (?, ?, ?)
""",
(team["Team"], field, str(team[field])),
)
except KeyError as ke:
logger.info(f"TEAM ERROR: {team}")
raise ke from None
Event.current_handle_index_progress = 1.0
def __write_match_predictions_file(self):
""" add score predictions to matches table """
df = []
for match in self.tba_data_static.schedule:
score = [
self.__get_match_pred_score(match, "b"),
self.__get_match_pred_score(match, "r"),
]
colors_pretty = ["Blue", "Red"]
df.append(
{
"Match": match["k"],
}
| reduce(
operator.or_,
[
{
f"Predict_{color.upper()}1_Score": round(score[i][0], 3),
f"Predict_{color.upper()}2_Score": round(score[i][1], 3),
f"Predict_{color.upper()}3_Score": round(score[i][2], 3),
f"Predict_{colors_pretty[i]}_Score": round(
sum(score[i]), 3
),
}
for i, color in enumerate(["b", "r"])
],
) # trust
| {"Predict_Winner": "blue" if sum(score[0]) > sum(score[1]) else "red"}
)
if len(df) <= 0:
logger.warning("No match schedule, skipping match predictions")
return
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
fields_to_write = list(df[0].keys())
fields_to_write.remove("Match")
for match in df:
conn.execute(
f"""
UPDATE matches
SET {", ".join([f"{x} = ?" for x in fields_to_write])}
WHERE Match = ?
""",
[*(list(match.values())[1:]), match["Match"]],
)
conn.commit()
Event.current_handle_index_progress = 1.0
def __write_statbotics_analytics(self):
""" Write statbotics stats for each match to the matches table """
df = []
for match in self.__sb_matches:
df.append(
{
"Match": match["key"],
}
| match["pred"]
)
if len(df) <= 0:
logger.warning("No matches, skipping statbotics analysis")
return
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
fields_to_write = list(df[0].keys())
fields_to_write.remove("Match")
for match in df:
conn.execute(
f"""
UPDATE matches
SET {", ".join([f"Statbotics_{x} = ?" for x in fields_to_write])}
WHERE Match = ?
""",
[*(list(match.values())[1:]), match["Match"]],
)
conn.commit()
Event.current_handle_index_progress = 1.0
def __write_statbotics_epa(self):
""" Write epas to the teams table """
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
for i, team in enumerate(self.tba_data_static.teams):
Event.current_handle_index_progress = i / len(
self.tba_data_static.teams
)
conn.execute(
f"""
UPDATE teams
SET EPA = ?
WHERE Team = ?
""",
(
self.__sb_epas[team] if team in self.__sb_epas else 0,
team,
),
)
conn.commit()
def __write_match_fields(self):
""" write the matches_fieids table """
df = []
for k, v in self.__matches.items():
for matTeam in v.output_dict(self.config_data):
# if int(k) <= len(self.tba_data_static.schedule): TODO: decide to delete or keep this if statement, delete for now :) (probably better to filter in graf. anyway so we have all the data ig.)
df.append(
{
"Match": next(
(
x["k"]
for x in self.tba_data_static.schedule
if str(int(k)) in x["k"]
),
f"practice_{int(k)}",
)
}
| matTeam
)
if len(df) <= 0:
logger.warning("No statbotics analytics fields.")
return
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
fields_to_write = list(df[0].keys())
fields_to_write.remove("Match")
fields_to_write.remove("Team")
for match in df:
for field in fields_to_write:
conn.execute(
f"""
INSERT OR REPLACE INTO matches_fields (match_key, team_key, field_name, field_value) VALUES (?, ?, ?, ?)
""",
(match["Match"], match["Team"], field, str(match[field])),
)
conn.commit()
Event.current_handle_index_progress = 1.0
def __write_match_depth_predictions(self):
""" write matches_depth_predictions table """
df = []
for match in self.tba_data_static.schedule:
for color in ["b", "r"]:
for i in range(3):
team = match[color][i].removeprefix("frc")
dat = {
"Match": match["k"],
"Color": color,
"Team": team,
"OPR": (
self.tba_data_dyn.oprs[int(team)]
if len(self.tba_data_dyn.oprs) > 0
and int(team) in self.tba_data_dyn.oprs
else 0.0
),
"EPA": (
self.__sb_epas[int(team)]
if len(self.__sb_epas) > 0 and int(team) in self.__sb_epas
else 0.0
),
}
for copr in self.config_data["copr"]:
val = 0.0
if (
len(self.tba_data_dyn.copr) > 0
and int(team) in self.tba_data_dyn.copr
and copr in self.tba_data_dyn.copr[int(team)]
):
val = self.tba_data_dyn.copr[int(team)][copr]
dat |= {copr: val}
if int(team) in self.__teams:
teamO = self.__teams[int(team)].output_dict(self.config_data)
for field in self.config_data["deep-predict"]:
if field["source"] in teamO:
dat |= {
field["name"]: teamO[field["source"]]
} # append the relevant datas
else:
for field in self.config_data["deep-predict"]:
dat |= {field["name"]: 0}
df.append(dat)
if len(df) <= 0:
logger.warning("No match schedule, skipping depth predictions")
return
Event.current_handle_index_progress = 0.5
with sqlite3.connect(PathUtils.file_set.data_db) as conn:
fields_to_write = list(df[0].keys())
fields_to_write.remove("Match")
fields_to_write.remove("Team")
for match in df:
for field in fields_to_write:
conn.execute(
f"""
INSERT OR REPLACE INTO matches_depth_predictions (match_key, team_key, attr_name, attr_value) VALUES (?, ?, ?, ?)
""",
(match["Match"], match["Team"], field, str(match[field])),
)
conn.commit()
Event.current_handle_index_progress = 1.0
[docs]
@staticmethod
def mad_filter(
data, c=2
): # https://real-statistics.com/sampling-distributions/identifying-outliers-missing-data
"""Filters the inputted `np.array` by removing all entries that are farther than c * MAD from the median"""
median = np.median(data) # X~
diff = np.abs(data - median) # diffs
mad = np.median(diff)
return data[diff <= (c * mad)]
[docs]
@staticmethod
def round_sigfigs(x, sig=3):
""" Rounds the number x to `sig` sigfigs """
if np.isscalar(x):
if x == 0.0 or np.isnan(np.log10(x)) or np.isinf(x):
return x
return np.around(x, sig - int(np.floor(np.log10(np.abs(x)))) - 1)
else:
return np.array([Processor.round_sigfigs(y) for y in x], dtype=np.float64)
def __get_svd_analysis(
self, stat: pd.DataFrame, compteamname, tn_field
) -> tuple[dict[int, np.float64], dict[int, np.float64], np.float64]:
"""Returns the nomalized rank factors of each team, the variance score of each team, and the stability score"""
# much thanks to pairwise
arrLen = len(self.__teams)
matrix = np.zeros((arrLen, arrLen))
teamkeys = list(self.__teams.keys())
team_index = {team: idx for idx, team in enumerate(teamkeys)}
grouped = (
stat.groupby([tn_field, compteamname])[stat.columns[2]].mean().fillna(0)
)
for (t1, t2), value in grouped.items():
if t1 in team_index and t2 in team_index:
i = team_index[t1]
j = team_index[t2]
matrix[i, j] = value
matrix[j, i] = -value
U, S, _ = svd(matrix)
u_ranks: np.ndarray = U[:, 0]
u_ranks = Processor.round_sigfigs(
(u_ranks - u_ranks.min()) / (u_ranks.max() - u_ranks.min()) * 100
)
stability = S.max() / S.min() if S.min() > 0 else np.inf
variation_score = np.zeros(len(S)) # less = more consistent
for i in range(len(S)):
variation_score[i] = Processor.round_sigfigs(
np.sqrt(sum([(U[i][j] * S[j]) ** 2 for j in range(1, len(S))]))
)
return (
dict(zip(teamkeys, u_ranks)),
dict(zip(teamkeys, variation_score)),
Processor.round_sigfigs(stability),
)
def __get_percent_scouted(self) -> float:
"""gets the percentage of teams scouted in the current comp"""
return round(
len([x for x in self.__teams.keys() if x in self.tba_data_static.teams])
/ len(self.tba_data_static.teams),
2, # use 2 because %
)
[docs]
def get_team_pred_score(self, team):
""" Gets the predicted score of a team based on the p-metric in the config """
score = 0
source_string = self.config_data["p-metric"]["source"]
if source_string == "OPR":
score = self.tba_data_dyn.oprs[int(team.removeprefix("frc"))]
elif source_string == "Last_OPR":
score = self.tba_data_static.oprs[int(team.removeprefix("frc"))]
elif source_string in self.config_data["copr"]:
score = self.tba_data_dyn.copr[int(team.removeprefix("frc"))][source_string]
elif source_string == "EPA":
score = self.__sb_epas[int(team.removeprefix("frc"))]
elif int(team.removeprefix("frc")) in self.__teams:
score = self.__teams[int(team.removeprefix("frc"))].output_dict(
self.config_data
)[source_string]
return score
def __get_match_pred_score(self, match, c):
"""Gets the predicted match score based on the prediction-metric specified in the config yaml"""
score = []
for key in match[c]:
score.append(self.get_team_pred_score(key)) # use prediction metric source
return score
def __write_other_metrics(self) -> None:
"""writes the json other metrics (right now only percent teams scouted) to the outfile"""
with open(PathUtils.file_set.other_metrics, "w") as w:
json.dump(
{"Percent Teams Scouted": self.__get_percent_scouted()}, w, indent=4
)
[docs]
def delete_match_scouter(data_filepath: str, mn: str, si: str) -> None:
""" Deletes a match from a scouter """
df = pd.read_csv(data_filepath)
df[~((df["MN"] == int(mn)) & (df["SI"] == si))].to_csv(
data_filepath, index=False
)
def __pre_filter_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Performs preprocess filtering based on configured metrics """
df.obj["Pre-filter-keep"] = True
for i in range(len(self.config_data["pre-tests"])):
Event.current_handle_index_progress = i / len(self.config_data["pre-tests"])
logger.info(
f"\tPerforming pre-test: {self.config_data["pre-tests"][i]["name"]} [{('x' * (i + 1)) + ('-' * (len(self.config_data["pre-tests"]) - (i + 1)))}]"
)
df.obj["Pre-filter-keep"] = (df.obj["Pre-filter-keep"]) & (
eval_beakscript(
self.config_data["pre-tests"][i]["expr"],
df.obj,
"Data Test " + self.config_data["pre-tests"][i]["name"],
)
)
df.obj = df.obj.loc[
df.obj["Pre-filter-keep"] == True
] # remove values that didn't pass the test (my English grade)
df.obj.drop(columns=["Pre-filter-keep"], inplace=True)
def __pre_process_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" applies preprocessing functions such as data restructure """
if len(self.config_data["preproc"]) > 0:
def apply_preproc_row(row: pd.Series, prep) -> list[pd.Series]:
new_rho = eval_beakscript(
prep["op"], row, "Preprocessor Function " + prep["name"]
)
if isinstance(new_rho, pd.Series):
if len(new_rho) > 0 and isinstance(new_rho.iloc[0], pd.Series):
return list(
new_rho
) # if series of series, return list of series
return [
new_rho
] # else return 1 element list of series (the series)
if isinstance(new_rho, (list, tuple)):
return list(new_rho)
raise TypeError(
f"Error: unsupported return type for preproc function {prep["name"]}: {type(new_rho)}"
)
for i in range(len(self.config_data["preproc"])):
Event.current_handle_index_progress = i / len(
self.config_data["preproc"]
)
last_cols = df.obj.columns
logger.info(
f"\tPerforming preprocess operation: {self.config_data["preproc"][i]["name"]} [{('x' * (i + 1)) + ('-' * (len(self.config_data["preproc"]) - (i + 1)))}]"
)
exp = (
df.obj.apply(
lambda row: apply_preproc_row(
row, self.config_data["preproc"][i]
),
axis=1,
)
.explode()
.reset_index(drop=True)
)
df.obj = pd.DataFrame(exp.tolist())
if "new-headers" in self.config_data["preproc"][i]:
df.obj.columns = self.config_data["preproc"][i]["new-headers"]
else:
df.obj.columns = last_cols
def __drop_duplicates_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Filters out duplicates """
if len(self.config_data["uniques"]) > 0:
dupes = df.obj.duplicated(subset=self.config_data["uniques"], keep=False)
if dupes.any():
logger.warning(
"Warning: duplicate teams for the following matches:\n\t\t"
+ "\n\t\t".join(
str(
df.obj[dupes][self.config_data["uniques"]].drop_duplicates()
).split("\n")
)
)
logger.info("\tFiltering out...")
df.obj = df.obj.drop_duplicates(
subset=self.config_data["uniques"], keep="first"
) # remove the duplicates
def __filter_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Applies configured filters to dataset """
df.obj["filter-keep"] = True
for i in range(len(self.config_data["tests"])):
Event.current_handle_index_progress = i / len(self.config_data["tests"])
logger.info(
f"\tPerforming test: {self.config_data["tests"][i]["name"]} [{('x' * (i + 1)) + ('-' * (len(self.config_data["tests"]) - (i + 1)))}]"
)
df.obj["filter-keep"] = (df.obj["filter-keep"]) & (
eval_beakscript(
self.config_data["tests"][i]["expr"],
df.obj,
"Data Test " + self.config_data["tests"][i]["name"],
)
)
df.obj = df.obj.loc[
df.obj["filter-keep"] == True
] # remove values that didn't pass the test (my English grade)
df.obj.drop(columns=["filter-keep"], inplace=True)
def __compute_fields_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Evaluates intermediate 'computed' fields for each match """
for i, comp in enumerate(self.config_data["compute"]):
Event.current_handle_index_progress = i / len(self.config_data["compute"])
# compute the beakscript formula with the current chunk (for each line), and output it into a new field named comp["name"]
# this works because beakscript fully supports pd.DataFrame's, of which chunk is one
df.obj[comp["name"]] = eval_beakscript(
comp["eq"], df.obj, "Compute Field " + comp["name"]
)
def __build_teams_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" builds the dictionary for teams from the data """
for team in df.obj[self.config_data["tn"]].unique():
self.__teams |= {int(team): TeamStruct()}
def __svd_data_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Evalutes SVD-derived metrics based on comparisons specified in the config """
tn = self.config_data["tn"]
for i, subj in enumerate(self.config_data["svd"]):
Event.current_handle_index_progress = i / len(self.config_data["svd"])
u, v, s = self.__get_svd_analysis(
df.obj[[self.config_data["tn"], subj["comp-team"], subj["source"]]],
subj["comp-team"],
tn,
)
svd_rank = []
svd_var = []
# make sorting
ks = np.array(list(u.keys()))
vs = np.array(list(u.values()))
sorted_i = np.argsort(vs)
sorted_v = vs[sorted_i]
dense_ranks = np.zeros_like(vs, dtype=int)
curr_rank = 0
dense_ranks[sorted_i[0]] = curr_rank
for i in range(1, len(vs)):
if sorted_v[i] != sorted_v[i - 1]:
curr_rank += 1
dense_ranks[sorted_i[i]] = curr_rank
ranks = dict(zip(ks, dense_ranks))
for team in df.obj[tn]:
svd_rank.append(u[team])
svd_var.append(v[team])
df.obj[f"{subj["name"]}"] = svd_rank
if "variance-score" in subj["augs"]:
df.obj[f"{subj["name"]} Variance"] = svd_var
if "stability" in subj["augs"]:
df.obj[f"{subj["name"]} Stabillity"] = [
s for _ in range(len(df.obj[tn]))
]
for team in df.obj[tn].unique():
self.__teams[int(team)].extend_data(
{
f"{subj["name"]}": ranks[team] + 1,
f"{subj["name"]} Variance": v[team],
}
)
def __reduce_df(self, df: ObjectHolder[pd.DataFrame]):
""" Reduces the dataframe based on different unique header specifications to remove post-SVD artifacts """
if "uniques-post" not in self.config_data:
return
df.obj.drop_duplicates(
subset=self.config_data["uniques-post"], inplace=True
) # remove CT dupe, everything same from here
df_numeric = df.obj.apply(lambda col: pd.to_numeric(col, errors="coerce"))
agg_dict = {}
for col in df.obj.columns:
if col in ["A", "B"]:
continue
if df_numeric[col].notna().any():
df.obj[col] = df_numeric[col]
agg_dict[col] = "mean"
else:
agg_dict[col] = "first"
df.obj = df.obj.groupby(
[
x
for x in self.config_data["uniques-post"]
if x != self.config_data["si"]
],
as_index=False,
).agg(agg_dict)
def __team_proc_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Writes team fields """
for i, team in enumerate(
df.obj[self.config_data["tn"]].unique()
): # teams will be in the chunk multiple times, but we just want to loop through all of the DIFFERENT teams there are
Event.current_handle_index_progress = i / len(
df.obj[self.config_data["tn"]].unique()
)
team_data = {}
for field in self.config_data["teams"]:
# call the beakscript functions for the derived fields where the TN == team
val = eval_beakscript(
field["derive"],
df.obj.loc[df.obj[self.config_data["tn"]] == team],
"Team Field " + field["name"],
)
if isinstance(val, pd.Series):
val = val.tolist()
team_data[field["name"]] = val # write the field to the csv
self.__teams[int(team)].extend_data(
team_data
) # add the data to the appropriate team struct
def __match_proc_chunk(self, df: ObjectHolder[pd.DataFrame]):
""" Writes match fields """
tn, mn = self.config_data["tn"], self.config_data["mn"]
for m, match in enumerate(df.obj[mn].unique()): # same thing as teams
Event.current_handle_index_progress = m / len(df.obj[mn].unique())
row = df.obj.loc[
df.obj[mn] == match
] # row is actually 6 rows here to be used for static fields
static_fields = {}
for field in self.config_data[
"matches"
]: # parse static fields before breaking down match by team
if "static" in field:
val = eval_beakscript(
field["derive"],
row,
"Static Match Field" + field["name"],
)
if isinstance(val, pd.Series):
val = val.tolist()
static_fields |= {field["name"]: val}
for i, team in enumerate(
df.obj.loc[df.obj[mn] == match, tn].unique()
): # the .unique is uneccesary but safer
data = {}
for field in self.config_data["matches"]:
row = df.obj.loc[(df.obj[tn] == team) & (df.obj[mn] == match)]
if "static" in field:
continue
# get derived fields for matches
val = eval_beakscript(
field["derive"],
row,
"Match Field" + field["name"],
)
if isinstance(val, pd.Series):
val = val.tolist()
data[field["name"]] = val
for f, fv in static_fields.items():
if isinstance(fv, Iterable) and not isinstance(fv, (str, bytes)):
data[f] = fv[i]
else:
data[f] = fv
self.__matches.setdefault(match, MatchStruct()).add_team_data(
int(team), data
)
def __write_output_file(self, df: ObjectHolder[pd.DataFrame]):
""" writes the output.csv file """
logger.info("Writing chunk...")
df.obj.to_csv(PathUtils.file_set.data_out_file, index=False, header=True)
[docs]
async def proccess_data(self, data_filepath: str) -> None:
"""Reads the input data, performs the calculations specified in field-config.yaml, and outputs all of the output files"""
if not self.has_sched_data:
return asyncio.sleep(0)
if self.tba_data_static.teams == None or self.tba_data_static.schedule == None:
await self.__periodic_calls.fire(logger.info)
if self.tba_data_static.teams == None or self.tba_data_static.schedule == None:
raise Exception("Error: Missing TBA Key: Please add one in settings")
self.__teams.clear()
self.__matches.clear()
df = pd.read_csv(data_filepath)
df_holder = ObjectHolder(df)
await self.__data_processing_routine.fire(logger.info, df_holder)
# write all the other files
await self.__post_process_routine.fire(logger.info)
[docs]
async def load_event_data(self):
""" Fires the load event data event """
await self.__load_in_event_data.fire(logger.info)
[docs]
async def clear_database(self) -> None:
""" Deletes the database """
if os.path.exists(PathUtils.file_set.data_db):
os.remove(PathUtils.file_set.data_db)
self.has_sched_data = False