diff --git a/main.py b/main.py
index 2eadb0310c3f9bc798457da5dc6d23a2e203ba1a..af8b495243a7a6cd354e03cdb7e653abc234e8aa 100644
--- a/main.py
+++ b/main.py
@@ -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()