# from git import Repo # import os # GITHUB_PAT = os.environ['GITHUB'] # if not os.path.exists('repo_directory'): # # os.mkdir('repo_directory') # Repo.clone_from(f'https://tracinginsights:{GITHUB_PAT}@github.com/TracingInsights/fastf1api.git', 'repo_directory' ) # from repo_directory.main import * import concurrent.futures import datetime import functools import math import os from io import BytesIO import fastf1 import numpy as np import pandas as pd import requests import streamlit as st from fastapi import Depends, FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, HTMLResponse from fastf1.ergast import Ergast from pydantic import BaseModel, Field from sqlalchemy.orm import Session # from . import accelerations, database, models, utils import accelerations import database import models import utils FASTF1_CACHE_DIR = os.environ["FASTF1_CACHE_DIR"] fastf1.Cache.enable_cache(FASTF1_CACHE_DIR) app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) database.Base.metadata.create_all(bind=database.engine) def get_db(): try: db = database.SessionLocal() yield db finally: db.close() class RacePace(BaseModel): year: int event: str session: str Driver: str LapTime: float Diff: float Team: str fill: str # @functools.cache @app.get("/racepace/{year}/{event}/{session}", response_model=None) async def average_race_pace( year: int, event: str | int, session: str, db: Session = Depends(get_db) ) -> any: race_pace_data = ( db.query(models.RacePace) .filter_by(year=year, event=event, session=session) .all() ) if race_pace_data: print("Fetching from Database") if not race_pace_data: print("Writing to Database") f1session = fastf1.get_session( year, event, session, # backend="fastf1", # force_ergast=False, ) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps laps = laps.loc[laps.LapNumber > 1] laps = laps.pick_track_status( "1", ) laps["LapTime"] = laps.Sector1Time + laps.Sector2Time + laps.Sector3Time # convert LapTime to seconds laps["LapTime"] = laps["LapTime"].apply(lambda x: x.total_seconds()) laps = laps.loc[laps.LapTime < laps.LapTime.min() * 1.07] df = ( laps[["LapTime", "Driver"]].groupby("Driver").mean().reset_index(drop=False) ) df = df.sort_values(by="LapTime").reset_index(drop=True) df["LapTime"] = df["LapTime"].round(3) df["Diff"] = (df["LapTime"] - df["LapTime"].min()).round(3) teams = laps[["Driver", "Team"]].drop_duplicates().reset_index(drop=True) # join teams and df df = df.merge(teams, on="Driver", how="left") car_colors = utils.team_colors(year) df["fill"] = df["Team"].map(car_colors) df_json = df.to_dict("records") # save the data to the database for record in df.to_dict("records"): race_pace = models.RacePace(**record) db.add(race_pace) db.commit() return {"racePace": df_json} return {"racePace": [dict(race_pace) for race_pace in race_pace_data]} @functools.cache @app.get("/topspeed/{year}/{event}/{session}", response_model=None) async def top_speed(year: int, event: str | int, session: str) -> any: f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps team_colors = utils.team_colors(year) fastest_speedtrap = ( laps[["SpeedI1", "SpeedI2", "SpeedST", "SpeedFL"]] .idxmax(axis=1) .value_counts() .index[0] ) speed_df = ( laps[[fastest_speedtrap, "Driver", "Compound", "Team"]] .groupby("Driver") .max() .sort_values(fastest_speedtrap, ascending=False) .reset_index() ) # add team colors to dataframe speed_df["fill"] = speed_df["Team"].apply(lambda x: team_colors[x]) # rename fastest speedtrap column to TopSpeed speed_df.rename(columns={fastest_speedtrap: "TopSpeed"}, inplace=True) # remove nan values in any column speed_df = speed_df.dropna() # Convert to int speed_df["TopSpeed"] = speed_df["TopSpeed"].astype(int) speed_dict = speed_df.to_dict(orient="records") return {"topSpeed": speed_dict} @functools.cache @app.get("/overtakes/{year}/{event}", response_model=None) def get_overtakes(year: int, event: str) -> any: def get_overtakes_df(year, event): if year == 2023: url = "https://docs.google.com/spreadsheets/d/1M4aepPJaIfdqE9oU3L-2CQqKIyubLXG4Q4cqWnyqxp4/export?format=csv" if year == 2022: url = "https://docs.google.com/spreadsheets/d/1cuS3B6hk4iQmMaRQoMTcogIInJpavnV7rKuEsiJnEbU/export?format=csv" if year == 2021: url = "https://docs.google.com/spreadsheets/d/1ANQnPVkefRmvzrmGvEqXoqQ4dBfgcI_R9FPg-0BcM34/export?format=csv" if year == 2020: url = "https://docs.google.com/spreadsheets/d/1eG9WTkXKzFT4NMh-WqHOMs5G0UuPGnb6wP4CnFD8uzY/export?format=csv" if year == 2019: url = "https://docs.google.com/spreadsheets/d/10nHg7BIs5ySh_dE9uuIz2lq-gRWcg02tIMr0EPgPvJs/export?format=csv" if year == 2018: url = "https://docs.google.com/spreadsheets/d/1MyAwQdczccdca_FAIiZKkqZNauNh3ts99JZ278S2OKc/export?format=csv" response = requests.get(url, timeout=10) df = pd.read_csv(BytesIO(response.content)) df = df[["Driver", event]] # replace NaNs with 0s df = df.fillna(0) # convert numbers to ints df[event] = df[event].astype(int) # replace event with "overtakes" df = df.rename(columns={event: "overtakes"}) return df def get_overtaken_df(year, event): if year == 2023: url = "https://docs.google.com/spreadsheets/d/1wszzx694Ot-mvA5YrFCpy3or37xMgnC0XpE8uNnJLWk/export?format=csv" if year == 2022: url = "https://docs.google.com/spreadsheets/d/19_XFDD3BZDIQVkNE4bG6dwuKvMaO4g5HNaUARGaJwhE/export?format=csv" if year == 2021: url = "https://docs.google.com/spreadsheets/d/1dQBHnd3AXEPNH5I75cjbzAAzi9ipqGk3v9eZT9eYKS4/export?format=csv" if year == 2020: url = "https://docs.google.com/spreadsheets/d/1snyntPMxYH4_KHSRI96AwBoJQrPbX6OanJAcqbYyW-Y/export?format=csv" if year == 2019: url = "https://docs.google.com/spreadsheets/d/11FfFkXErJg7F22iVwJo9XfLFAWucMBVlzL1qUGWxM3s/export?format=csv" if year == 2018: url = "https://docs.google.com/spreadsheets/d/1XJXAEyRpRS_UwLHzEtN2PdIaFJYGWSN6ypYN8Ecwp9A/export?format=csv" response = requests.get(url, timeout=10) df = pd.read_csv(BytesIO(response.content)) df = df[["Driver", event]] # replace NaNs with 0s df = df.fillna(0) # convert numbers to ints df[event] = df[event].astype(int) df = df.rename(columns={event: "overtaken"}) return df overtakes = get_overtakes_df(year, event) overtaken = get_overtaken_df(year, event) df = overtakes.merge(overtaken, on="Driver") # remove drivers with 0 overtakes and 0 overtaken df = df[(df["overtakes"] != 0) | (df["overtaken"] != 0)] # sort in the decreasing order of overtakes df = df.sort_values( by=["overtakes", "overtaken"], ascending=[False, True] ).reset_index(drop=True) # convert to dictionary df_dict = df.to_dict(orient="records") return {"overtakes": df_dict} @functools.cache @app.get("/fastest/{year}/{event}/{session}", response_model=None) async def fastest_lap(year: int, event: str | int, session: str) -> any: f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps drivers = pd.unique(laps["Driver"]) list_fastest_laps = list() for drv in drivers: drvs_fastest_lap = laps.pick_driver(drv).pick_fastest() list_fastest_laps.append(drvs_fastest_lap) df = ( fastf1.core.Laps(list_fastest_laps) .sort_values(by="LapTime") .reset_index(drop=True) ) pole_lap = df.pick_fastest() df["Diff"] = df["LapTime"] - pole_lap["LapTime"] car_colors = utils.team_colors(year) df["fill"] = df["Team"].map(car_colors) # convert timedelta to float and round to 3 decimal places df["Diff"] = df["Diff"].dt.total_seconds().round(3) df = df[["Driver", "LapTime", "Diff", "Team", "fill"]] # remove nan values in any column df = df.dropna() df_json = df.to_dict("records") return {"fastest": df_json} # @st.cache_data @app.get("/wdc", response_model=None) async def driver_standings() -> any: YEAR = 2023 # datetime.datetime.now().year df = pd.DataFrame( pd.read_html(f"https://www.formula1.com/en/results.html/{YEAR}/drivers.html")[0] ) df = df[["Driver", "PTS", "Car"]] # reverse the order df = df.sort_values(by="PTS", ascending=True) # in Driver column only keep the last 3 characters df["Driver"] = df["Driver"].str[:-5] # add colors to the dataframe car_colors = utils.team_colors(YEAR) df["fill"] = df["Car"].map(car_colors) # remove rows where points is 0 df = df[df["PTS"] != 0] df.reset_index(inplace=True, drop=True) df.rename(columns={"PTS": "Points"}, inplace=True) return {"WDC": df.to_dict("records")} # @st.cache_data @app.get("/", response_model=None) async def root(): return HTMLResponse( content="""""", status_code=200, ) # @st.cache_data @app.get("/years", response_model=None) async def years_available() -> any: # make a list from 2018 to current year current_year = datetime.datetime.now().year years = list(range(2018, current_year + 1)) # reverse the list to get the latest year first years.reverse() years = [{"label": str(year), "value": year} for year in years] return {"years": years} # format for events {"events":[{"label":"Saudi Arabian Grand Prix","value":2},{"label":"Bahrain Grand Prix","value":1},{"label":"Pre-Season Testing","value":"t1"}]} # @st.cache_data @app.get("/{year}", response_model=None) async def events_available(year: int) -> any: # get events available for a given year data = utils.LatestData(year) events = data.get_events() events = [{"label": event, "value": event} for i, event in enumerate(events)] events.reverse() return {"events": events} # format for sessions {"sessions":[{"label":"FP1","value":"FP1"},{"label":"FP2","value":"FP2"},{"label":"FP3","value":"FP3"},{"label":"Qualifying","value":"Q"},{"label":"Race","value":"R"}]} # @st.cache_data @functools.cache @app.get("/{year}/{event}", response_model=None) async def sessions_available(year: int, event: str | int) -> any: # get sessions available for a given year and event data = utils.LatestData(year) sessions = data.get_sessions(event) sessions = [{"label": session, "value": session} for session in sessions] return {"sessions": sessions} # format for drivers {"drivers":[{"color":"#fff500","label":"RIC","value":"RIC"},{"color":"#ff8700","label":"NOR","value":"NOR"},{"color":"#c00000","label":"VET","value":"VET"},{"color":"#0082fa","label":"LAT","value":"LAT"},{"color":"#787878","label":"GRO","value":"GRO"},{"color":"#ffffff","label":"GAS","value":"GAS"},{"color":"#f596c8","label":"STR","value":"STR"},{"color":"#787878","label":"MAG","value":"MAG"},{"color":"#0600ef","label":"ALB","value":"ALB"},{"color":"#ffffff","label":"KVY","value":"KVY"},{"color":"#fff500","label":"OCO","value":"OCO"},{"color":"#0600ef","label":"VER","value":"VER"},{"color":"#00d2be","label":"HAM","value":"HAM"},{"color":"#ff8700","label":"SAI","value":"SAI"},{"color":"#00d2be","label":"BOT","value":"BOT"},{"color":"#960000","label":"GIO","value":"GIO"}]} # @st.cache_data @functools.cache @app.get("/strategy/{year}/{event}", response_model=None) async def get_strategy(year: int, event: str | int) -> any: f1session = fastf1.get_session(year, event, "R") f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps drivers_list = pd.unique(laps["Driver"]) drivers = pd.DataFrame(drivers_list, columns=["Driver"]) drivers["FinishOrder"] = drivers.index + 1 # Get the LapNumber of the first lap of each stint first_lap = ( laps[["Driver", "Stint", "Compound", "LapNumber"]] .groupby(["Driver", "Stint", "Compound"]) .first() .reset_index() ) # Add FinishOrder to first_lap first_lap = pd.merge(first_lap, drivers, on="Driver") # change LapNumber to LapStart first_lap = first_lap.rename(columns={"LapNumber": "LapStart"}) # reduce the lapstart by 1 first_lap["LapStart"] = first_lap["LapStart"] - 1 # find the last lap of each stint last_lap = ( laps[["Driver", "Stint", "Compound", "LapNumber"]] .groupby(["Driver", "Stint", "Compound"]) .last() .reset_index() ) # change LapNumber to LapEnd last_lap = last_lap.rename(columns={"LapNumber": "LapEnd"}) # combine first_lap and last_lap stint_laps = pd.merge(first_lap, last_lap, on=["Driver", "Stint", "Compound"]) # to cover for outliers stint_laps["fill"] = "white" stint_laps["fill"] = stint_laps["Compound"].map( { "SOFT": "red", "MEDIUM": "yellow", "HARD": "white", "INTERMEDIATE": "blue", "WET": "green", } ) # sort by FinishOrder stint_laps = stint_laps.sort_values(by=["FinishOrder"], ascending=[True]) stint_laps_dict = stint_laps.to_dict("records") return {"strategy": stint_laps_dict} @functools.cache @app.get("/lapchart/{year}/{event}/{session}", response_model=None) async def lap_chart( year: int, event: str | int, session: str, ) -> any: ergast = Ergast() race_names_df = ergast.get_race_schedule(season=year, result_type="pandas") event_number = race_names_df[race_names_df["raceName"] == event]["round"].values[0] drivers_df = ergast.get_driver_info( season=year, round=event_number, result_type="pandas" ) laptimes_df = ergast.get_lap_times( season=year, round=event_number, result_type="pandas", limit=2000 ).content[0] laptimes_df = pd.merge(laptimes_df, drivers_df, how="left", on="driverId") results_df = ergast.get_race_results( season=year, round=event_number, result_type="pandas" ).content[0] results_df = results_df[["driverCode", "constructorName"]] # merge results_df on laptime_df laptimes_df = pd.merge(laptimes_df, results_df, how="left", on="driverCode") team_colors = utils.team_colors(year) # add team_colors to laptimes_df laptimes_df["fill"] = laptimes_df["constructorName"].map(team_colors) # rename number as x and position as y laptimes_df.rename( columns={"number": "x", "position": "y", "driverCode": "id"}, inplace=True ) lap_chart_data = [] for driver in laptimes_df["id"].unique(): data = laptimes_df[laptimes_df["id"] == driver] fill = data["fill"].values[0] data = data[["x", "y"]] data_dict = data.to_dict(orient="records") driver_dict = {"id": driver, "fill": fill, "data": data_dict} # add this to all_data lap_chart_data.append(driver_dict) lap_chart_dict = {"lapChartData": lap_chart_data} return lap_chart_dict @functools.cache @app.get("/{year}/{event}/{session}", response_model=None) async def session_drivers(year: int, event: str | int, session: str) -> any: # get drivers available for a given year, event and session f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps team_colors = utils.team_colors(year) # add team_colors dict to laps on Team column laps["color"] = laps["Team"].map(team_colors) unique_drivers = laps["Driver"].unique() drivers = [ { "color": laps[laps.Driver == driver].color.iloc[0], "label": driver, "value": driver, } for driver in unique_drivers ] return {"drivers": drivers} @functools.cache @app.get("/laps/{year}/{event}/{session}", response_model=None) async def get_driver_laps_data(year: int, event: str | int, session: str) -> any: # get drivers available for a given year, event and session f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps team_colors = utils.team_colors(year) # add team_colors dict to laps on Team column laps["color"] = laps["Team"].map(team_colors) # combine Driver and LapNumber as a new column laps["label"] = ( laps["Driver"] + "-" + laps["LapNumber"].astype(int).astype(str) + "-" + str(year) + "-" + event + "-" + session ) laps["value"] = ( laps["Driver"] + "-" + laps["LapNumber"].astype(int).astype(str) + "-" + str(year) + "-" + event + "-" + session ) laps = laps[["value", "label", "color"]] driver_laps_dict = laps.to_dict("records") return {"laps": driver_laps_dict} # format for chartData {"chartData":[{"lapnumber":1},{ # "VER":91.564, # "VER_compound":"SOFT", # "VER_compound_color":"#FF5733", # "lapnumber":2 # },{"lapnumber":3},{"VER":90.494,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":4},{"lapnumber":5},{"VER":90.062,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":6},{"lapnumber":7},{"VER":89.815,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":8},{"VER":105.248,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":9},{"lapnumber":10},{"VER":89.79,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":11},{"VER":145.101,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":12},{"lapnumber":13},{"VER":89.662,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":14},{"lapnumber":15},{"VER":89.617,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":16},{"lapnumber":17},{"VER":140.717,"VER_compound":"SOFT","VER_compound_color":"#FF5733","lapnumber":18}]} # @st.cache_data @functools.cache @app.get("/{year}/{event}/{session}/{driver}", response_model=None) async def laps_data(year: int, event: str | int, session: str, driver: str) -> any: # get drivers available for a given year, event and session f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps team_colors = utils.team_colors(year) # add team_colors dict to laps on Team column drivers = laps.Driver.unique() # for each driver in drivers, get the Team column from laps and get the color from team_colors dict drivers = [ { "color": team_colors[laps[laps.Driver == driver].Team.iloc[0]], "label": driver, "value": driver, } for driver in drivers ] driver_laps = laps.pick_driver(driver) driver_laps["LapTime"] = driver_laps["LapTime"].dt.total_seconds() # remove rows where LapTime is null driver_laps = driver_laps[driver_laps.LapTime.notnull()] compound_colors = { "SOFT": "#FF0000", "MEDIUM": "#FFFF00", "HARD": "#FFFFFF", "INTERMEDIATE": "#00FF00", "WET": "#088cd0", } driver_laps_data = [] for _, row in driver_laps.iterrows(): if row["LapTime"] > 0: lap = { f"{driver}": row["LapTime"], f"{driver}_compound": row["Compound"], f"{driver}_compound_color": compound_colors[row["Compound"]], "lapnumber": row["LapNumber"], } else: lap = {"lapnumber": row["LapNumber"]} driver_laps_data.append(lap) return {"chartData": driver_laps_data} @functools.cache @app.get("/laptimes/{year}/{event}/{session}/{driver}", response_model=None) async def get_laps_data(year: int, event: str | int, session: str, driver: str) -> any: # get drivers available for a given year, event and session f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=False, weather=False, messages=False) laps = f1session.laps team_colors = utils.team_colors(year) # add team_colors dict to laps on Team column drivers = laps.Driver.unique() # for each driver in drivers, get the Team column from laps and get the color from team_colors dict drivers = [ { "color": team_colors[laps[laps.Driver == driver].Team.iloc[0]], "label": driver, "value": driver, } for driver in drivers ] driver_laps = laps.pick_driver(driver) driver_laps["LapTime"] = driver_laps["LapTime"].dt.total_seconds() driver_laps = driver_laps[["Driver", "LapTime", "LapNumber", "Compound"]] # remove rows where LapTime is null driver_laps = driver_laps[driver_laps.LapTime.notnull()] driver_laps_dict = driver_laps.to_dict("records") return {"chartData": driver_laps_dict} # @st.cache_data @functools.cache @app.get("/{year}/{event}/{session}/{driver}/{lap_number}", response_model=None) async def telemetry_data( year: int, event: str | int, session: str, driver: str, lap_number: int ) -> any: f1session = fastf1.get_session(year, event, session) f1session.load(telemetry=True, weather=False, messages=False) laps = f1session.laps driver_laps = laps.pick_driver(driver) driver_laps["LapTime"] = driver_laps["LapTime"].dt.total_seconds() # get the telemetry for lap_number selected_lap = driver_laps[driver_laps.LapNumber == lap_number] telemetry = selected_lap.get_telemetry() lon_acc, lat_acc = accelerations.compute_accelerations(telemetry) telemetry["lon_acc"] = lon_acc telemetry["lat_acc"] = lat_acc telemetry["Time"] = telemetry["Time"].dt.total_seconds() laptime = selected_lap.LapTime.values[0] data_key = f"{driver} - Lap {int(lap_number)} - {year} {session} [laptime]" telemetry["DRS"] = telemetry["DRS"].apply(lambda x: 1 if x in [10, 12, 14] else 0) brake_tel = [] drs_tel = [] gear_tel = [] rpm_tel = [] speed_tel = [] throttle_tel = [] time_tel = [] track_map = [] lon_acc_tel = [] lat_acc_tel = [] for _, row in telemetry.iterrows(): brake = { "x": row["Distance"], "y": row["Brake"], } brake_tel.append(brake) drs = { "x": row["Distance"], "y": row["DRS"], } drs_tel.append(drs) gear = { "x": row["Distance"], "y": row["nGear"], } gear_tel.append(gear) rpm = { "x": row["Distance"], "y": row["RPM"], } rpm_tel.append(rpm) speed = { "x": row["Distance"], "y": row["Speed"], } speed_tel.append(speed) throttle = { "x": row["Distance"], "y": row["Throttle"], } throttle_tel.append(throttle) time = { "x": row["Distance"], "y": row["Time"], } time_tel.append(time) lon_acc = { "x": row["Distance"], "y": row["lon_acc"], } lon_acc_tel.append(lon_acc) lat_acc = { "x": row["Distance"], "y": row["lat_acc"], } lat_acc_tel.append(lat_acc) track = { "x": row["X"], "y": row["Y"], } track_map.append(track) telemetry_data = { "telemetryData": { "brake": brake_tel, "dataKey": data_key, "drs": drs_tel, "gear": gear_tel, "rpm": rpm_tel, "speed": speed_tel, "throttle": throttle_tel, "time": time_tel, "lon_acc": lon_acc_tel, "lat_acc": lat_acc_tel, "trackMap": track_map, } } return telemetry_data