diff --git a/DataAnalyser.py b/DataAnalyser.py index 6a9d99df4b2b75186afa99948a1cc9a092a25a79..8830ebd4277acaaed402a83bc934d103e4a318dd 100644 --- a/DataAnalyser.py +++ b/DataAnalyser.py @@ -201,6 +201,7 @@ class DataAnalyser(DataHandler): return False + # ===== Tire Changes ===== def getFirstTireChanges(self, races: list[Session]): @@ -245,19 +246,6 @@ class DataAnalyser(DataHandler): return latestTireChangeLap - def __getFirstLapWithoutCompound(self, compoundsPerLap: list[list[str]], startingCompound: str): - currentLap = 0 - compoundFilter = self.__generateFilter(startingCompound) - for compoundsThisLap in compoundsPerLap: - noStartingCompoundsLeft = True - for compound in compoundsThisLap: - if compound in compoundFilter: - 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 @@ -276,6 +264,19 @@ class DataAnalyser(DataHandler): return compoundsPerLap + def __getFirstLapWithoutCompound(self, compoundsPerLap: list[list[str]], startingCompound: str): + currentLap = 0 + compoundFilter = self.__generateFilter(startingCompound) + for compoundsThisLap in compoundsPerLap: + noStartingCompoundsLeft = True + for compound in compoundsThisLap: + if compound in compoundFilter: + noStartingCompoundsLeft = False + if noStartingCompoundsLeft: return currentLap + currentLap += 1 + + return -1 # no lap without compound found; all laps use same compound type + def __getPredominantCompound(self, compoundsThisLap: list[str]): slickCounter = 0 interCounter = 0 @@ -314,11 +315,10 @@ class DataAnalyser(DataHandler): # ===== Events ===== -# def analyseRacesForSafetyCars(self, races): - # ===== Other + # ===== Other ===== def __enforceSessionType(self, session: Session, sessionType: str): if sessionType not in self.validSessionTypes: diff --git a/DataImporter.py b/DataImporter.py index 8c4cefb9ff31ff99b9a3b0f7b5b7da4c65a15f60..65661122f9aa376e33c5ddc30bcea087e0b5f0c3 100644 --- a/DataImporter.py +++ b/DataImporter.py @@ -22,68 +22,7 @@ class DataImporter(DataHandler, ABC): sessions. If the method does not require one of these, it should not be part of this class. """ - def fetchWeather(self, event: Event | SessionIdentifier | list[Event] | list[SessionIdentifier]): - """ - Fetch session weather from the wikipedia entry for the specified event(s). - :param event: The event or identifier for any session of the event to fetch weather for. Alternatively a list of - these. - :return: Weather conditions as string, or a list of such strings if a list is provided. - """ - - # Recursive call if list provided - if isinstance(event, list): - weatherEntries: list[string] = [] - for e in event: - weatherEntries.append(self.fetchWeather(e)) - return weatherEntries - - # Convert from SessionIdentifier to Event if needed - if isinstance(event, SessionIdentifier): - event = self.importEvent(event.year, event.event) - - # Fetch weather - wikiHtml = self.__importWikiHtml(event) - analyser = DataAnalyser() - weather = analyser.extractWeatherFromHtml(wikiHtml) - - return weather - - def __importWikiHtml(self, event: Event): - """ - Fetch the HTML contents of the wikipedia page for the event given. - :param event: Event whose wikipedia page to fetch. - :return: HTML content of the wikipedia page as a string. - """ - apiRootUrl: string = "https://en.wikipedia.org/wiki/" - grandPrixName: string = f"{event.year} {event["EventName"]}" - uri: string = re.sub(" ", "_", grandPrixName) - url: string = f"{apiRootUrl}{uri}" - response = requests.get(url) - if not response.ok: - raise ConnectionError(f"Could not fetch wikipedia article with URL {url}") - return response.text - # URL example: https://en.wikipedia.org/w/api.php?action=query&titles=2024_S%C3%A3o_Paulo_Grand_Prix&prop=extracts&format=json&exintro=1 - - def importEvent(self, year: int, event: int | str): - return fastf1.get_event(year, event) - - def importAllEventsFromYear(self, year: int): - """ - Get all event objects from given year. - :param year: Year from which to retrieve event objects. - :return: List of all event objects from the given year. Returns None if - """ - if self.__isInFuture(year): raise ValueError("Cannot get event data from the future :)") - - events: list[Event] = [] - schedule: EventSchedule = fastf1.get_event_schedule(year, include_testing = False) - if schedule.empty: - print(f"INFO: No races have run in {year} yet. You will receive a None object") - return None - - for raceIndex in schedule['RoundNumber']: - events.append(schedule.get_event_by_round(raceIndex)) - return events + # ===== Sessions ===== def importSessionWeather(self, sessionIdentifier: SessionIdentifier): """ @@ -111,6 +50,10 @@ class DataImporter(DataHandler, ABC): session.load(laps = laps, telemetry = telemetry, weather = weather, messages = messages) return session + + + # ===== Events ===== + def getWetEventsSince(self, firstYear: int): if self.__isInFuture(firstYear): raise ValueError("Cannot get race data from the future :)") @@ -131,10 +74,6 @@ class DataImporter(DataHandler, ABC): return rainRaces - def __isInFuture(self, year: int): - currentYear = datetime.now().year - return year > currentYear - def getRainRacesSince(self, firstYear: int): """ Retrieve all race sessions since and including given year that had at least one minute of rainfall during the race session. @@ -174,3 +113,78 @@ class DataImporter(DataHandler, ABC): return rainRaces + def importEvent(self, year: int, event: int | str): + return fastf1.get_event(year, event) + + def importAllEventsFromYear(self, year: int): + """ + Get all event objects from given year. + :param year: Year from which to retrieve event objects. + :return: List of all event objects from the given year. Returns None if + """ + if self.__isInFuture(year): raise ValueError("Cannot get event data from the future :)") + + events: list[Event] = [] + schedule: EventSchedule = fastf1.get_event_schedule(year, include_testing = False) + if schedule.empty: + print(f"INFO: No races have run in {year} yet. You will receive a None object") + return None + + for raceIndex in schedule['RoundNumber']: + events.append(schedule.get_event_by_round(raceIndex)) + return events + + + + # ===== Weather ===== + + def fetchWeather(self, event: Event | SessionIdentifier | list[Event] | list[SessionIdentifier]): + """ + Fetch session weather from the wikipedia entry for the specified event(s). + :param event: The event or identifier for any session of the event to fetch weather for. Alternatively a list of + these. + :return: Weather conditions as string, or a list of such strings if a list is provided. + """ + + # Recursive call if list provided + if isinstance(event, list): + weatherEntries: list[string] = [] + for e in event: + weatherEntries.append(self.fetchWeather(e)) + return weatherEntries + + # Convert from SessionIdentifier to Event if needed + if isinstance(event, SessionIdentifier): + event = self.importEvent(event.year, event.event) + + # Fetch weather + wikiHtml = self.__importWikiHtml(event) + analyser = DataAnalyser() + weather = analyser.extractWeatherFromHtml(wikiHtml) + + return weather + + def __importWikiHtml(self, event: Event): + """ + Fetch the HTML contents of the wikipedia page for the event given. + :param event: Event whose wikipedia page to fetch. + :return: HTML content of the wikipedia page as a string. + """ + apiRootUrl: string = "https://en.wikipedia.org/wiki/" + grandPrixName: string = f"{event.year} {event["EventName"]}" + uri: string = re.sub(" ", "_", grandPrixName) + url: string = f"{apiRootUrl}{uri}" + response = requests.get(url) + if not response.ok: + raise ConnectionError(f"Could not fetch wikipedia article with URL {url}") + return response.text + # URL example: https://en.wikipedia.org/w/api.php?action=query&titles=2024_S%C3%A3o_Paulo_Grand_Prix&prop=extracts&format=json&exintro=1 + + + + # ===== Other ===== + + def __isInFuture(self, year: int): + currentYear = datetime.now().year + return year > currentYear +