Skip to content
Snippets Groups Projects
Commit ad527258 authored by maeries's avatar maeries
Browse files

differentiation between combustion and electric

parent 94007bb7
Branches
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 88 additions and 68 deletions
main.py
+ 88
− 68
View file @ ad527258
......@@ -2,38 +2,67 @@ import os
import csv
import math
import glob
from operator import truediv
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
import numpy as np
from datetime import datetime
import multiprocessing as mp
# Constants for Elo calculation
K_FACTOR = 32 # Standard chess K-factor
INITIAL_ELO = 1500 # Starting Elo for all teams
def expected_score(rating_a, rating_b):
"""Calculate the expected score for a player with rating_a against a player with rating_b"""
return 1 / (1 + math.pow(10, (rating_b - rating_a) / 400))
class EloSystem:
def __init__(self, csv_directory, output_directory):
self.csv_directory = csv_directory
def __init__(self, csv_directory, output_directory, competition_type):
self.competition_type = competition_type
self.events_directory = os.path.join(csv_directory, "events")
self.competition_meta_data_directory = csv_directory
self.output_directory = output_directory
self.teams = {} # Dictionary to store team Elo ratings
self.match_history = [] # List to store all match results for history tracking
self.competition_history = {} # Dictionary to store competition results
self.initialize_output_directory()
self.competition_meta_data = self.load_competition_meta_data()
def load_competition_meta_data(self):
if self.competition_type == "combustion":
meta_data = pd.read_csv(os.path.join(self.competition_meta_data_directory, "FSC.csv"))
elif self.competition_type == "electric":
meta_data = pd.read_csv(os.path.join(self.competition_meta_data_directory, "FSE.csv"))
else:
raise
meta_data.sort_values(by=['id'])
return meta_data
def initialize_output_directory(self):
"""Create output directory if it doesn't exist"""
os.makedirs(self.output_directory, exist_ok=True)
os.makedirs(os.path.join(self.output_directory, 'images'), exist_ok=True)
def get_competition_name(self, id):
return self.competition_meta_data[self.competition_meta_data["id"] == int(id)]['abbr'].values[0].strip()
def get_competition_date(self, id):
date_string = self.competition_meta_data[self.competition_meta_data["id"] == int(id)]['date'].values[0]
return datetime.strptime(date_string, "%Y-%m-%d")
def load_and_process_csv_files(self):
"""Load all CSV files and process them to calculate Elo ratings"""
csv_files = glob.glob(os.path.join(self.csv_directory, '*.csv'))
csv_files = self.competition_meta_data
if not csv_files:
print(f"No CSV files found in {self.csv_directory}")
return
csv_files = csv_files.sort_values('date')
csv_files = csv_files['id'].tolist()
csv_files = [os.path.join(self.events_directory, str(x) + ".csv") for x in csv_files]
print(f"Found {len(csv_files)} CSV files to process")
......@@ -41,22 +70,26 @@ class EloSystem:
for csv_file in sorted(csv_files):
print(f"Processing {os.path.basename(csv_file)}...")
competition_name = os.path.splitext(os.path.basename(csv_file))[0]
competition_id = os.path.basename(csv_file).split(".")[0]
teams_in_competition = []
# Read the CSV file
with open(csv_file, 'r', newline='', encoding='utf-8') as file:
reader = csv.reader(file)
# Skip header if exists (assuming first row might be headers)
try:
first_row = next(reader)
# If the first row doesn't start with a number, assume it's a header
if not first_row[0].strip().isdigit():
rank = 1
else:
# It's not a header, it's actual data
rank = int(first_row[0].strip())
team_name = first_row[1].strip() if len(first_row) > 1 else first_row[0].strip()
try:
with open(csv_file, 'r', newline='', encoding='utf-8') as file:
reader = csv.reader(file)
skipped = False
for row in reader:
if not skipped:
skipped = True
continue
if not row: # Skip empty rows
continue
rank = row[3].strip()
team_name = row[0].strip()
# Initialize the team if not seen before
if team_name not in self.teams:
self.teams[team_name] = {
......@@ -65,39 +98,16 @@ class EloSystem:
'history': []
}
teams_in_competition.append((rank, team_name))
except StopIteration:
# File is empty
continue
# Process remaining rows
for row in reader:
if not row: # Skip empty rows
continue
# Check if row starts with a number (ranking)
if row[0].strip().isdigit():
rank = int(row[0].strip())
team_name = row[1].strip() if len(row) > 1 else row[0].strip()
else:
# If no explicit ranking, use row index
rank = len(teams_in_competition) + 1
team_name = row[0].strip()
# Initialize the team if not seen before
if team_name not in self.teams:
self.teams[team_name] = {
'elo': INITIAL_ELO,
'competitions': 0,
'history': []
}
teams_in_competition.append((rank, team_name))
except FileNotFoundError:
print(f"file {csv_file} not found: skipping")
continue
# Sort teams by rank
teams_in_competition.sort(key=lambda x: x[0])
teams_in_competition.sort(key=lambda x: int(x[0]))
# Store competition results
date = datetime.now().strftime("%Y-%m-%d") # Use current date as competition date
self.competition_history[competition_name] = {
date = self.get_competition_date(os.path.basename(csv_file).split(".")[0])
self.competition_history[competition_id] = {
'date': date,
'results': teams_in_competition
}
......@@ -110,12 +120,13 @@ class EloSystem:
self.teams[team_name]['competitions'] += 1
# Update Elo ratings based on this competition's results
self.update_elo_for_competition(team_names, competition_name)
self.update_elo_for_competition(team_names, competition_id)
def update_elo_for_competition(self, teams_in_order, competition_name):
def update_elo_for_competition(self, teams_in_order, competition_id):
"""Update Elo ratings for all teams in a competition"""
# HIER ÄNDERN, UM EV UND CV AUSEINANDERZUSORTIEREN
# For each team, compare with teams ranked below them
date = datetime.now().strftime("%Y-%m-%d") # Use current date as competition date
date = self.get_competition_date(competition_id) # Use current date as competition date
# Store pre-competition Elo ratings
pre_competition_elo = {team: self.teams[team]['elo'] for team in teams_in_order}
......@@ -126,7 +137,7 @@ class EloSystem:
team_b = teams_in_order[j]
# Team A is ranked higher than Team B
self.update_elo_pair(team_a, team_b, 1, 0, competition_name, date)
self.update_elo_pair(team_a, team_b, 1, 0, competition_id, date)
# Store post-competition Elo ratings and changes
for team in teams_in_order:
......@@ -135,23 +146,23 @@ class EloSystem:
# Store in team history
self.teams[team]['history'].append({
'date': date,
'competition': competition_name,
'date': date.strftime("%Y-%m-%d"),
'competition': self.get_competition_name(competition_id),
'rank': rank,
'previous_elo': pre_competition_elo[team],
'new_elo': self.teams[team]['elo'],
'elo_change': elo_change
})
def update_elo_pair(self, team_a, team_b, score_a, score_b, competition_name, date):
def update_elo_pair(self, team_a, team_b, score_a, score_b, competition_id, date):
"""Update Elo ratings for a pair of teams based on their match outcome"""
# Get current Elo ratings
rating_a = self.teams[team_a]['elo']
rating_b = self.teams[team_b]['elo']
# Calculate expected scores
expected_a = self.expected_score(rating_a, rating_b)
expected_b = self.expected_score(rating_b, rating_a)
expected_a = expected_score(rating_a, rating_b)
expected_b = expected_score(rating_b, rating_a)
# Update Elo ratings
new_rating_a = rating_a + K_FACTOR * (score_a - expected_a)
......@@ -160,7 +171,7 @@ class EloSystem:
# Store match in history
match_info = {
'date': date,
'competition': competition_name,
'competition': competition_id,
'team_a': team_a,
'team_b': team_b,
'old_elo_a': rating_a,
......@@ -176,18 +187,22 @@ class EloSystem:
self.teams[team_a]['elo'] = new_rating_a
self.teams[team_b]['elo'] = new_rating_b
def expected_score(self, rating_a, rating_b):
"""Calculate the expected score for a player with rating_a against a player with rating_b"""
return 1 / (1 + math.pow(10, (rating_b - rating_a) / 400))
def generate_website(self):
"""Generate static HTML website with Elo ratings and visualizations"""
# Create main index page
self.create_index_page()
# Create team-specific pages
for team_name in self.teams:
self.create_team_page(team_name)
# # Create team-specific pages
# for team_name in self.teams:
# self.create_team_page(team_name)
# If num_processes not specified, use number of CPU cores
num_processes = int(mp.cpu_count() / 2)
print(f"staring computation on {num_processes} cores")
# Create a pool of workers
pool = mp.Pool(processes=num_processes)
pool.map(self.create_team_page, self.teams)
# Create visualizations
self.create_visualizations()
......@@ -204,7 +219,7 @@ class EloSystem:
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Team Elo Rankings</title>
<title>Formula Student {self.competition_type} Elo Rankings</title>
<style>
body {{
font-family: Arial, sans-serif;
......@@ -256,7 +271,7 @@ class EloSystem:
</style>
</head>
<body>
<h1>Team Elo Rankings</h1>
<h1>Formula Student {self.competition_type} Elo Rankings</h1>
<p>Rankings based on {len(self.competition_history)} competitions.</p>
<div class="chart-container">
......@@ -301,6 +316,7 @@ class EloSystem:
def create_team_page(self, team_name):
"""Create a dedicated page for a specific team"""
print(f"Generating Team site for {team_name}")
team_data = self.teams[team_name]
# Generate Elo history chart for this team
......@@ -582,7 +598,11 @@ def main():
args = parser.parse_args()
# Create and run the Elo system
elo_system = EloSystem(args.input, args.output)
elo_system = EloSystem(args.input, args.output + "_combustion", "combustion")
elo_system.load_and_process_csv_files()
elo_system.generate_website()
elo_system = EloSystem(args.input, args.output + "_electric", "electric")
elo_system.load_and_process_csv_files()
elo_system.generate_website()
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment