import json from typing import Any, Callable, Dict, List, overload import numpy as np from formula10 import cache from formula10.domain.domain_model import Model from formula10.domain.model.driver import NONE_DRIVER, Driver from formula10.domain.model.race_guess import RaceGuess from formula10.domain.model.race_result import RaceResult from formula10.domain.model.season_guess import SeasonGuess from formula10.domain.model.season_guess_result import SeasonGuessResult from formula10.domain.model.team import Team from formula10.domain.model.user import User # Guess points RACE_GUESS_OFFSET_POINTS: Dict[int, int] = { 3: 1, 2: 3, 1: 6, 0: 10 } RACE_GUESS_DNF_POINTS: int = 10 SEASON_GUESS_HOT_TAKE_POINTS: int = 10 SEASON_GUESS_P2_POINTS: int = 10 SEASON_GUESS_OVERTAKES_POINTS: int = 10 SEASON_GUESS_DNF_POINTS: int = 10 SEASON_GUESS_GAINED_POINTS: int = 10 SEASON_GUESS_LOST_POINTS: int = 10 SEASON_GUESS_TEAMWINNER_CORRECT_POINTS: int = 3 SEASON_GUESS_TEAMWINNER_FALSE_POINTS: int = -3 SEASON_GUESS_PODIUMS_CORRECT_POINTS: int = 3 SEASON_GUESS_PODIUMS_FALSE_POINTS: int = -2 # Driver points DRIVER_RACE_POINTS: Dict[int, int] = { 1: 25, 2: 18, 3: 15, 4: 12, 5: 10, 6: 8, 7: 6, 8: 4, 9: 2, 10: 1 } DRIVER_SPRINT_POINTS: Dict[int, int] = { 1: 8, 2: 7, 3: 6, 4: 5, 5: 4, 6: 3, 7: 2, 8: 1 } DRIVER_FASTEST_LAP_POINTS: int = 1 # Last season results WDC_STANDING_2023: Dict[str, int] = { "Max Verstappen": 1, "Sergio Perez": 2, "Lewis Hamilton": 3, "Fernando Alonso": 4, "Charles Leclerc": 5, "Lando Norris": 6, "Carlos Sainz": 7, "George Russell": 8, "Oscar Piastri": 9, "Lance Stroll": 10, "Pierre Gasly": 11, "Esteban Ocon": 12, "Alexander Albon": 13, "Yuki Tsunoda": 14, "Valtteri Bottas": 15, "Nico Hulkenberg": 16, "Daniel Ricciardo": 17, "Zhou Guanyu": 18, "Kevin Magnussen": 19, "Logan Sargeant": 21 } WCC_STANDING_2023: Dict[str, int] = { "Red Bull": 1, "Mercedes": 2, "Ferrari": 3, "McLaren": 4, "Aston Martin": 5, "Alpine": 6, "Williams": 7, "VCARB": 8, "Sauber": 9, "Haas": 10 } def standing_points(race_guess: RaceGuess, race_result: RaceResult) -> int: guessed_driver_position: int | None = race_result.driver_standing_position(driver=race_guess.pxx_guess) if guessed_driver_position is None: return 0 position_offset: int = abs(guessed_driver_position - race_guess.race.place_to_guess) if position_offset not in RACE_GUESS_OFFSET_POINTS: return 0 return RACE_GUESS_OFFSET_POINTS[position_offset] def dnf_points(race_guess: RaceGuess, race_result: RaceResult) -> int: if race_guess.dnf_guess in race_result.initial_dnf: return RACE_GUESS_DNF_POINTS if race_guess.dnf_guess == NONE_DRIVER and len(race_result.initial_dnf) == 0: return RACE_GUESS_DNF_POINTS return 0 class PointsModel(Model): """ This class bundles all data + functionality required to do points calculations. """ def __init__(self): Model.__init__(self) @cache.cached(timeout=None, key_prefix="points_points_per_step") # Clear when adding/updating race results or users def points_per_step(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each user. """ points_per_step = dict() for user in self.all_users(): points_per_step[user.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers for race_guess in self.all_race_guesses(): user_name: str = race_guess.user.name race_number: int = race_guess.race.number race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name) if race_result is None: continue points_per_step[user_name][race_number] = standing_points(race_guess, race_result) + dnf_points(race_guess, race_result) return points_per_step @cache.memoize(timeout=None, args_to_ignore=["self"]) # Clear when adding/updating race results def driver_points_per_step(self, *, include_inactive: bool) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each driver. """ driver_points_per_step = dict() for driver in self.all_drivers(include_none=False, include_inactive=include_inactive): driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers for race_result in self.all_race_results(): race_number: int = race_result.race.number for position, driver in race_result.standing.items(): driver_points_per_step[driver.name][race_number] = DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 driver_points_per_step[driver.name][race_number] += DRIVER_FASTEST_LAP_POINTS if race_result.fastest_lap_driver == driver else 0 for position, driver in race_result.sprint_standing.items(): driver_name: str = driver.name driver_points_per_step[driver_name][race_number] += DRIVER_SPRINT_POINTS[int(position)] if int(position) in DRIVER_SPRINT_POINTS else 0 return driver_points_per_step @cache.cached(timeout=None, key_prefix="points_team_points_per_step") def team_points_per_step(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each team. """ team_points_per_step = dict() for team in self.all_teams(include_none=False): team_points_per_step[team.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers for race_result in self.all_race_results(): for driver in race_result.standing.values(): team_name: str = driver.team.name race_number: int = race_result.race.number team_points_per_step[team_name][race_number] += self.driver_points_per_step(include_inactive=True)[driver.name][race_number] return team_points_per_step @cache.cached(timeout=None, key_prefix="points_dnfs") def dnfs(self) -> Dict[str, int]: dnfs = dict() for driver in self.all_drivers(include_none=False, include_inactive=True): dnfs[driver.name] = 0 for race_result in self.all_race_results(): for driver in race_result.all_dnfs: dnfs[driver.name] += 1 for driver in race_result.sprint_dnfs: dnfs[driver.name] += 1 return dnfs # # Driver stats # @cache.cached(timeout=None, key_prefix="points_driver_points_per_step_cumulative") # Cleanup when adding/updating race results def driver_points_per_step_cumulative(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing cumulative points per race for each driver. """ points_per_step_cumulative: Dict[str, List[int]] = dict() for driver_name, points in self.driver_points_per_step(include_inactive=True).items(): points_per_step_cumulative[driver_name] = np.cumsum(points).tolist() return points_per_step_cumulative @overload def driver_points_by(self, *, driver_name: str, include_inactive: bool) -> List[int]: """ Returns a list of points per race for a specific driver. """ return self.driver_points_by(driver_name=driver_name, include_inactive=include_inactive) @overload def driver_points_by(self, *, race_name: str, include_inactive: bool) -> Dict[str, int]: """ Returns a dictionary of points per driver for a specific race. """ return self.driver_points_by(race_name=race_name, include_inactive=include_inactive) @overload def driver_points_by(self, *, driver_name: str, race_name: str, include_inactive: bool) -> int: """ Returns the points for a specific race for a specific driver. """ return self.driver_points_by(driver_name=driver_name, race_name=race_name, include_inactive=include_inactive) @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def driver_points_by(self, *, driver_name: str | None = None, race_name: str | None = None, include_inactive: bool) -> List[int] | Dict[str, int] | int: if driver_name is not None and race_name is None: return self.driver_points_per_step(include_inactive=include_inactive)[driver_name] if driver_name is None and race_name is not None: race_number: int = self.race_by(race_name=race_name).number points_by_race: Dict[str, int] = dict() for _driver_name, points in self.driver_points_per_step(include_inactive=include_inactive).items(): points_by_race[_driver_name] = points[race_number] return points_by_race if driver_name is not None and race_name is not None: race_number: int = self.race_by(race_name=race_name).number return self.driver_points_per_step(include_inactive=include_inactive)[driver_name][race_number] raise Exception("driver_points_by received an illegal combination of arguments") @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def total_driver_points_by(self, driver_name: str) -> int: return sum(self.driver_points_by(driver_name=driver_name, include_inactive=True)) @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def drivers_sorted_by_points(self, *, include_inactive: bool) -> List[Driver]: comparator: Callable[[Driver], int] = lambda driver: self.total_driver_points_by(driver.name) return sorted(self.all_drivers(include_none=False, include_inactive=include_inactive), key=comparator, reverse=True) @cache.cached(timeout=None, key_prefix="points_wdc_standing_by_position") # Cleanup when adding/updating race results def wdc_standing_by_position(self) -> Dict[int, List[str]]: standing: Dict[int, List[str]] = dict() for position in range(1, len(self.all_drivers(include_none=False, include_inactive=True)) + 1): standing[position] = list() position: int = 1 last_points: int = 0 for driver in self.drivers_sorted_by_points(include_inactive=True): points: int = self.total_driver_points_by(driver.name) if points < last_points: position += 1 standing[position].append(driver.name) last_points = points return standing @cache.cached(timeout=None, key_prefix="points_wdc_standing_by_driver") # Cleanup when adding/updating race results def wdc_standing_by_driver(self) -> Dict[str, int]: standing: Dict[str, int] = dict() position: int = 1 last_points: int = 0 for driver in self.drivers_sorted_by_points(include_inactive=True): points: int = self.total_driver_points_by(driver.name) if points < last_points: position += 1 standing[driver.name] = position last_points = points return standing def wdc_diff_2023_by(self, driver_name: str) -> int: if not driver_name in WDC_STANDING_2023: return 0 return WDC_STANDING_2023[driver_name] - self.wdc_standing_by_driver()[driver_name] @cache.cached(timeout=None, key_prefix="points_most_dnf_names") # Cleanup when adding/updating race results def most_dnf_names(self) -> List[str]: dnf_names: List[str] = list() most_dnfs: int = 0 for dnfs in self.dnfs().values(): if dnfs > most_dnfs: most_dnfs = dnfs for driver_name, dnfs in self.dnfs().items(): if dnfs == most_dnfs: dnf_names.append(driver_name) return dnf_names @cache.cached(timeout=None, key_prefix="points_most_gained_names") # Cleanup when adding/updating race results def most_gained_names(self) -> List[str]: most_gained_names: List[str] = list() most_gained: int = 0 for driver in self.all_drivers(include_none=False, include_inactive=True): gained: int = self.wdc_diff_2023_by(driver.name) if gained > most_gained: most_gained = gained for driver in self.all_drivers(include_none=False, include_inactive=True): gained: int = self.wdc_diff_2023_by(driver.name) if gained == most_gained: most_gained_names.append(driver.name) return most_gained_names @cache.cached(timeout=None, key_prefix="points_most_lost_names") # Cleanup when adding/updating race results def most_lost_names(self) -> List[str]: most_lost_names: List[str] = list() most_lost: int = 100 for driver in self.all_drivers(include_none=False, include_inactive=True): lost: int = self.wdc_diff_2023_by(driver.name) if lost < most_lost: most_lost = lost for driver in self.all_drivers(include_none=False, include_inactive=True): lost: int = self.wdc_diff_2023_by(driver.name) if lost == most_lost: most_lost_names.append(driver.name) return most_lost_names # # Team points # @cache.cached(timeout=None, key_prefix="points_team_points_per_step_cumulative") # Cleanup when adding/updating race results def team_points_per_step_cumulative(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing cumulative points per race for each team. """ points_per_step_cumulative: Dict[str, List[int]] = dict() for team_name, points in self.team_points_per_step().items(): points_per_step_cumulative[team_name] = np.cumsum(points).tolist() return points_per_step_cumulative @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def total_team_points_by(self, team_name: str) -> int: teammates: List[Driver] = self.drivers_by(team_name=team_name, include_inactive=True) return sum(sum(self.driver_points_by(driver_name=teammate.name, include_inactive=True)) for teammate in teammates) @cache.cached(timeout=None, key_prefix="points_teams_sorted_by_points") # Cleanup when adding/updating race results def teams_sorted_by_points(self) -> List[Team]: comparator: Callable[[Team], int] = lambda team: self.total_team_points_by(team.name) return sorted(self.all_teams(include_none=False), key=comparator, reverse=True) @cache.cached(timeout=None, key_prefix="points_wcc_standing_by_position") # Cleanup when adding/updating race results def wcc_standing_by_position(self) -> Dict[int, List[str]]: standing: Dict[int, List[str]] = dict() for position in range (1, len(self.all_teams(include_none=False)) + 1): standing[position] = list() position: int = 1 last_points: int = 0 for team in self.teams_sorted_by_points(): points: int = self.total_team_points_by(team.name) if points < last_points: position += 1 standing[position].append(team.name) last_points = points return standing @cache.cached(timeout=None, key_prefix="points_wcc_standing_by_team") # Cleanup when adding/updating race results def wcc_standing_by_team(self) -> Dict[str, int]: standing: Dict[str, int] = dict() position: int = 1 last_points: int = 0 for team in self.teams_sorted_by_points(): points: int = self.total_team_points_by(team.name) if points < last_points: position += 1 standing[team.name] = position last_points = points return standing def wcc_diff_2023_by(self, team_name: str) -> int: return WCC_STANDING_2023[team_name] - self.wcc_standing_by_team()[team_name] # # User stats # def points_per_step_cumulative(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing cumulative points per race for each user. """ points_per_step_cumulative: Dict[str, List[int]] = dict() for user_name, points in self.points_per_step().items(): points_per_step_cumulative[user_name] = np.cumsum(points).tolist() return points_per_step_cumulative @overload def points_by(self, *, user_name: str) -> List[int]: """ Returns a list of points per race for a specific user. """ return self.points_by(user_name=user_name) @overload def points_by(self, *, race_name: str) -> Dict[str, int]: """ Returns a dictionary of points per user for a specific race. """ return self.points_by(race_name=race_name) @overload def points_by(self, *, user_name: str, race_name: str) -> int: """ Returns the points for a specific race for a specific user. """ return self.points_by(user_name=user_name, race_name=race_name) @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results or users def points_by(self, *, user_name: str | None = None, race_name: str | None = None) -> List[int] | Dict[str, int] | int: if user_name is not None and race_name is None: return self.points_per_step()[user_name] if user_name is None and race_name is not None: race_number: int = self.race_by(race_name=race_name).number points_by_race: Dict[str, int] = dict() for _user_name, points in self.points_per_step().items(): points_by_race[_user_name] = points[race_number] return points_by_race if user_name is not None and race_name is not None: race_number: int = self.race_by(race_name=race_name).number return self.points_per_step()[user_name][race_number] raise Exception("points_by received an illegal combination of arguments") def total_points_by(self, user_name: str) -> int: """ Returns the total number of points for a specific user. """ return sum(self.points_by(user_name=user_name)) def users_sorted_by_points(self) -> List[User]: """ Returns the list of users, sorted by their points from race guesses (in descending order). """ comparator: Callable[[User], int] = lambda user: self.total_points_by(user.name) return sorted(self.all_users(), key=comparator, reverse=True) @cache.cached(timeout=None, key_prefix="points_user_standing") # Cleanup when adding/updating race results or users def user_standing(self) -> Dict[str, int]: standing: Dict[str, int] = dict() position: int = 1 last_points: int = 0 for user in self.users_sorted_by_points(): if self.total_points_by(user.name) < last_points: position += 1 standing[user.name] = position last_points = self.total_points_by(user.name) return standing def picks_count(self, user_name: str) -> int: # Treat standing + dnf picks separately return len(self.race_guesses_by(user_name=user_name)) * 2 @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def picks_with_points_count(self, user_name: str) -> int: count: int = 0 for race_guess in self.race_guesses_by(user_name=user_name): race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name) if race_result is None: continue if standing_points(race_guess, race_result) > 0: count += 1 if dnf_points(race_guess, race_result) > 0: count += 1 return count def points_per_pick(self, user_name: str) -> float: if self.picks_count(user_name) == 0: return 0.0 return self.total_points_by(user_name) / self.picks_count(user_name) # # Season guess evaluation # def hot_take_correct(self, user_name: str) -> bool: season_guess_result: SeasonGuessResult | None = self.season_guess_result_by(user_name=user_name) return season_guess_result.hot_take_correct if season_guess_result is not None else False def p2_constructor_correct(self, user_name: str) -> bool: season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) if season_guess is None or season_guess.p2_wcc is None: return False return season_guess.p2_wcc.name in self.wcc_standing_by_position()[2] def overtakes_correct(self, user_name: str) -> bool: season_guess_result: SeasonGuessResult | None = self.season_guess_result_by(user_name=user_name) return season_guess_result.overtakes_correct if season_guess_result is not None else False def dnfs_correct(self, user_name: str) -> bool: season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) if season_guess is None or season_guess.most_dnfs is None: return False return season_guess.most_dnfs.name in self.most_dnf_names() def most_gained_correct(self, user_name: str) -> bool: season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) if season_guess is None or season_guess.most_wdc_gained is None: return False return season_guess.most_wdc_gained.name in self.most_gained_names() def most_lost_correct(self, user_name: str) -> bool: season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) if season_guess is None or season_guess.most_wdc_lost is None: return False return season_guess.most_wdc_lost.name in self.most_lost_names() @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def is_team_winner(self, driver: Driver) -> bool: teammates: List[Driver] = self.drivers_by(team_name=driver.team.name, include_inactive=True) teammate: Driver = teammates[0] if teammates[1] == driver else teammates[1] return self.wdc_standing_by_driver()[driver.name] <= self.wdc_standing_by_driver()[teammate.name] @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def has_podium(self, driver: Driver) -> bool: for race_result in self.all_race_results(): position: int | None = race_result.driver_standing_position(driver) if position is not None and position <= 3: return True return False # # Diagram queries # def cumulative_points_data(self) -> str: data: Dict[Any, Any] = dict() data["labels"] = [0] + [ race.name for race in sorted(self.all_races(), key=lambda race: race.number) ] data["datasets"] = [ { "data": self.points_per_step_cumulative()[user.name], "label": user.name, "fill": False } for user in self.all_users() ] return json.dumps(data) def cumulative_driver_points_data(self) -> str: data: Dict[Any, Any] = dict() data["labels"] = [0] + [ race.name for race in sorted(self.all_races(), key=lambda race: race.number) ] data["datasets"] = [ { "data": self.driver_points_per_step_cumulative()[driver.name], "label": driver.abbr, "fill": False } for driver in self.all_drivers(include_none=False, include_inactive=True) ] return json.dumps(data) def cumulative_team_points_data(self) -> str: data: Dict[Any, Any] = dict() data["labels"] = [0] + [ race.name for race in sorted(self.all_races(), key=lambda race: race.number) ] data["datasets"] = [ { "data": self.team_points_per_step_cumulative()[team.name], "label": team.name, "fill": False } for team in self.all_teams(include_none=False) ] return json.dumps(data)