DataAnalyser.py

import pandas as pandas
from fastf1.core import Session, Lap, Laps

from DataHandler import DataHandler


class DataAnalyser(DataHandler):



    # ===== Overtakes =====

    def analyseRacesForOvertakes(self, races: list[Session]):
        overtakesInRaces: list[list[int]] = [[]]
        for race in races:
            overtakesInRaces.append(self.analyseRaceForOvertakes(race))
        return overtakesInRaces

    def analyseRaceForOvertakes(self, race: Session):
        if not race.session_info["Type"] == "Race":
            raise ValueError(f"Session must be a race session, not a {race.session_info["Type"]} session")
        overtakesInLaps: list[int] = self.countOvertakesPerLap(race)
        return overtakesInLaps

    def countOvertakesPerLap(self, race: Session):
        overtakes: list[int] = []
        for lapNumber in range(1, race.total_laps + 1): # in this context, index 1 = lap 1
            overtakes.append(self.countOvertakesInLap(lapNumber, race))
        return overtakes

    def countOvertakesInLap(self, lapNumber: int, race: Session):
        orderToSort: list[(Lap, int)] = self.prepareWeightedOrderFromLap(lapNumber, race)
        overtakes: int = 0
        i: int = 0

        while i < len(orderToSort) - 1: # do not change to for-loop, seems to not like resetting the index
            weightedDriverAhead: list[(Lap, int)] = orderToSort[i]
            weightedDriverBehind: list[(Lap, int)] = orderToSort[i + 1]
            if weightedDriverAhead[1] > weightedDriverBehind[1]:
                temp: int = orderToSort[i]
                orderToSort[i] = orderToSort[i + 1]
                orderToSort[i + 1] = temp
                i = -1  # reset to first index; -1 because loop will set it to 0

                if not ( # don't count overtake if driver nonexistent or if one of them is on an in-lap
                    weightedDriverAhead[0] is None
                    or weightedDriverBehind[0] is None
                    or self.isAnInLap(weightedDriverAhead[0])
                    or self.isAnInLap(weightedDriverBehind[0])
                ):  overtakes += 1
            i += 1

        return overtakes

    def prepareWeightedOrderFromLap(self, lapNumber: int, race: Session):
        '''Prepare a list from specific lap & race, that can be sorted via bubble sort to determine the number of
        overtakes that occured in that lap.
        :param lapNumber: Which lap to prepare from the given race. Note that value of 1 will return a list ordered by
        starting grid, as there is no previous lap.
        :param race: Race from which to pull the given lap from.
        :return: list[(Lap, int)]: A list with pairs of every driver's lap and their position at the end of the lap. Entries are
        sorted by the driver's positions at the start of the lap. If an invalid lap number (below 1 or above the number
        of laps the race had), all laps in the list will be either None objects or empty Panda Dataframes.
        '''

        previousLaps: Laps = race.laps.pick_laps(lapNumber - 1)
        currentLaps: Laps = race.laps.pick_laps(lapNumber)
        out: int = self.numberOfDrivers  # weighting for drivers that did not complete the lap (they are "out" of the lap/race)
        weightedOrder: list[(Lap, int)] = [(None, out)] * 20  # data from start of lap + weighting for end of lap; initialize to add laps in lap's starting order

        if self.activateDebugOvertakeAnalysis: print(f"Lap: {lapNumber}")

        # Put every driver's laps in a sortable array & apply weighting based on position at end of lap
        for driver in race.drivers:
            if self.activateDebugOvertakeAnalysis: print(f"Driver: {driver}")
            driversPreviousLap: Laps = previousLaps.pick_drivers(
                driver)  # should only get 1 lap, but data type shenanigans
            driversCurrentLap: Laps = currentLaps.pick_drivers(driver)
            startPosition: int = out
            endPosition: int = out

            try:
                if lapNumber == 1:
                    startPosition = self.getGridPositionForDriver(driver, race)
                else: startPosition = int(driversPreviousLap['Position'].iloc[0])
                endPosition = int(driversCurrentLap['Position'].iloc[0])

            except ValueError:
                if self.activateDebugOvertakeAnalysis:
                    print(f"Could not fetch positions from lap; driver %d likely didn't finish lap %d or %d",
                          driver, lapNumber, (lapNumber + 1))
            except IndexError:
                if self.activateDebugOvertakeAnalysis:
                    print("['Position'].iloc[0] was out of bounds; lap was likely empty because driver previously left the race")

            weightedOrder[startPosition - 1] = (driversCurrentLap, endPosition)

        return weightedOrder

    def getGridPositionForDriver(self, driverNumber: int, race: Session):
        sessionResults = race.results
        gridPosition: int = int(sessionResults['GridPosition'].loc[driverNumber])
        return gridPosition

    def isAnInLap(self, lap: Lap):
        try:
            return not pandas.isnull(lap['PitInTime'].iloc[0])
        except: # caused when lap is empty and possibly when lap is None
            return True # like in-laps, empty laps should not be counted for overtakes



    # ===== Weather =====

    def analyseRacesForWeather(self, races: list[Session]):
        weatherInRaces: list[list[bool]] = [[]] # isRaining per lap per race
        for race in races:
            weatherInRace = self.analyseRaceForWeather(race)
            weatherInRaces.append(weatherInRace)
        return weatherInRaces

    def analyseRaceForWeather(self, race: Session):
        x = 0
        return 0



    # ===== Tire Changes =====

    def getEarliestTireChanges(self, races: list[Session]):
        earliestTireChanges: list[int] = [[]]  # isRaining per lap per race
        for race in races:
            earliestTireChange = self.getEarliestTireChange(race)
            earliestTireChanges.append(earliestTireChange)
        return earliestTireChanges

    # Returns -1 if no tire change occured
    def getEarliestTireChange(self, race: Session):
        earliestTireChangeLap: int = -1
        compoundsPerLap: list[list[str]] = self.getCompoundsForRace(race)
        compoundsPerLap[0] = compoundsPerLap[1] # presume grid tires same as 1st lap; races are only picked if weather change after first 10 laps anyways, so its ok
        startingCompound: str = self.getPredominantCompound(compoundsPerLap[0])
        earliestTireChangeLap = self.getFirstLapWithOppositeCompound(compoundsPerLap, startingCompound)

        return earliestTireChangeLap

    def getLatestTireChanges(self, races: list[Session]):
        latestTireChanges: list[int] = [[]]  # isRaining per lap per race
        for race in races:
            latestTireChange = self.getLatestTireChange(race)
            latestTireChanges.append(latestTireChange)
        return latestTireChanges

    # Returns -1 if no tire change occured
    def getLatestTireChange(self, race: Session):
        latestTireChangeLap: int = -1
        compoundsPerLap: list[list[str]] = self.getCompoundsForRace(race)
        compoundsPerLap[0] = compoundsPerLap[1]  # presume grid tires same as 1st lap; races are only picked if weather change after first 10 laps anyways, so its ok
        startingCompound: str = self.getPredominantCompound(compoundsPerLap[0])
        latestTireChangeLap = self.getFirstLapWithoutCompound(compoundsPerLap, startingCompound)

        return latestTireChangeLap

    def getFirstLapWithoutCompound(self, compoundsPerLap: list[list[str]], startingCompound: str):
        currentLap = 0
        filter = self.setFilter(startingCompound)
        for compoundsThisLap in compoundsPerLap:
            noStartingCompoundsLeft = True
            for compound in compoundsThisLap:
                if compound in filter:
                    noStartingCompoundsLeft = False
            if noStartingCompoundsLeft: return currentLap
            currentLap += 1

        return -1 # no lap without compound found; all laps use same compound type

    def getCompoundsForRace(self, race: Session):
        compoundsPerLap: list[list[str]] = [[]]
        allRaceLaps = race.laps

        for raceLapIndex in range(race.total_laps):
            compoundsThisLap: list[str] = []
            for driver in race.drivers:
                raceLap = allRaceLaps.pick_laps(raceLapIndex + 1)  # Lap 0 doesn't exist
                raceLap = raceLap.pick_drivers(driver)
                try:
                    compound = raceLap['Compound'].iloc[0]
                    compoundsThisLap.append(compound)
                except:  # triggered when not all drivers that took part reached lap, probably by crashing or being behind
                    x = 0  # do nothing
            compoundsPerLap.append(compoundsThisLap)

        return compoundsPerLap

    def getPredominantCompound(self, compoundsThisLap: list[str]):
        slickCounter = 0
        interCounter = 0
        wetCounter = 0
        for compound in compoundsThisLap:
            if compound in self.slickCompounds: slickCounter += 1
            if compound == 'INTERMEDIATE': interCounter += 1
            if compound == 'WET': wetCounter += 1
        mostUsed = max(slickCounter, interCounter, wetCounter)
        if slickCounter == mostUsed: return 'SLICK'
        if interCounter == mostUsed: return 'INTERMEDIATE'
        if wetCounter == mostUsed: return 'WET'
        return 'error'

    def getFirstLapWithOppositeCompound(self, compoundsPerLap: list[list[str]], startingCompound: str):
        filter = self.setFilter(startingCompound)
        currentLap = 0
        for compoundsThisLap in compoundsPerLap:
            for compound in compoundsThisLap:
                if compound not in filter:
                    return currentLap
            currentLap += 1
        return -1 # no lap with opposite compound found; all laps use same compound type

    def setFilter(self, startingCompound: str):
        if startingCompound == 'SLICK': return self.slickCompounds
        return startingCompound



    # ===== Crashes =====

    def analyseRacesForCrashes(self, races):
        x = 0



    # ===== Events =====

    def analyseRacesForSafetyCars(self, races):
        x = 0