Bug: Fix duplicated place numbering in points model
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 14s

This commit is contained in:
2024-10-19 14:01:46 +02:00
parent f38a5f2e6d
commit 61d247508f

View File

@ -14,12 +14,7 @@ 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_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
@ -44,18 +39,9 @@ DRIVER_RACE_POINTS: Dict[int, int] = {
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
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
@ -80,7 +66,7 @@ WDC_STANDING_2023: Dict[str, int] = {
"Daniel Ricciardo": 17,
"Zhou Guanyu": 18,
"Kevin Magnussen": 19,
"Logan Sargeant": 21
"Logan Sargeant": 21,
}
WCC_STANDING_2023: Dict[str, int] = {
"Red Bull": 1,
@ -92,11 +78,14 @@ WCC_STANDING_2023: Dict[str, int] = {
"Williams": 7,
"VCARB": 8,
"Sauber": 9,
"Haas": 10
"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)
guessed_driver_position: int | None = race_result.driver_standing_position(
driver=race_guess.pxx_guess
)
if guessed_driver_position is None:
return 0
@ -106,6 +95,7 @@ def standing_points(race_guess: RaceGuess, race_result: RaceResult) -> int:
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
@ -115,6 +105,7 @@ def dnf_points(race_guess: RaceGuess, race_result: RaceResult) -> int:
return 0
class PointsModel(Model):
"""
This class bundles all data + functionality required to do points calculations.
@ -123,51 +114,76 @@ class PointsModel(Model):
def __init__(self):
Model.__init__(self)
@cache.cached(timeout=None, key_prefix="points_points_per_step") # Clear when adding/updating race results or users
@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
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)
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)
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
@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 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
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
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]]:
"""
@ -175,14 +191,20 @@ class PointsModel(Model):
"""
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
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]
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
@ -206,48 +228,78 @@ class PointsModel(Model):
# Driver stats
#
@cache.cached(timeout=None, key_prefix="points_driver_points_per_step_cumulative") # Cleanup when adding/updating race results
@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():
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]:
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)
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]:
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)
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:
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)
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:
@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]
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():
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
@ -255,24 +307,42 @@ class PointsModel(Model):
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]
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
@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))
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
@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)
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
@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):
for position in range(
1, len(self.all_drivers(include_none=False, include_inactive=True)) + 1
):
standing[position] = list()
position: int = 1
@ -281,14 +351,18 @@ class PointsModel(Model):
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
# If multiple drivers have equal points, a place is shared.
# In this case, the next driver does not occupy the immediate next position.
position += len(standing[position])
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
@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()
@ -298,7 +372,14 @@ class PointsModel(Model):
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
drivers_with_this_position = 0
for _driver, _position in standing.items():
if _position == position:
drivers_with_this_position += 1
# If multiple drivers have equal points, a place is shared.
# In this case, the next driver does not occupy the immediate next position.
position += drivers_with_this_position
standing[driver.name] = position
last_points = points
@ -309,9 +390,13 @@ class PointsModel(Model):
if not driver_name in WDC_STANDING_2023:
return 0
return WDC_STANDING_2023[driver_name] - self.wdc_standing_by_driver()[driver_name]
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
@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
@ -326,7 +411,9 @@ class PointsModel(Model):
return dnf_names
@cache.cached(timeout=None, key_prefix="points_most_gained_names") # Cleanup when adding/updating race results
@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
@ -345,7 +432,9 @@ class PointsModel(Model):
return most_gained_names
@cache.cached(timeout=None, key_prefix="points_most_lost_names") # Cleanup when adding/updating race results
@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
@ -368,7 +457,9 @@ class PointsModel(Model):
# Team points
#
@cache.cached(timeout=None, key_prefix="points_team_points_per_step_cumulative") # Cleanup when adding/updating race results
@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.
@ -379,21 +470,34 @@ class PointsModel(Model):
return points_per_step_cumulative
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
@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)
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
@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)
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
@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):
for position in range(1, len(self.all_teams(include_none=False)) + 1):
standing[position] = list()
position: int = 1
@ -402,14 +506,18 @@ class PointsModel(Model):
for team in self.teams_sorted_by_points():
points: int = self.total_team_points_by(team.name)
if points < last_points:
position += 1
# If multiple teams have equal points, a place is shared.
# In this case, the next team does not occupy the immediate next position.
position += len(standing[position])
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
@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()
@ -419,7 +527,14 @@ class PointsModel(Model):
for team in self.teams_sorted_by_points():
points: int = self.total_team_points_by(team.name)
if points < last_points:
position += 1
teams_with_this_position = 0
for _team, _position in standing.items():
if _position == position:
teams_with_this_position += 1
# If multiple teams have equal points, a place is shared.
# In this case, the next team does not occupy the immediate next position.
position += teams_with_this_position
standing[team.name] = position
last_points = points
@ -464,8 +579,12 @@ class PointsModel(Model):
"""
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:
@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]
@ -498,15 +617,25 @@ class PointsModel(Model):
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
@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
users_with_this_position = 0
for _user, _position in standing.items():
if _position == position:
users_with_this_position += 1
# If multiple users have equal points, a place is shared.
# In this case, the next user does not occupy the immediate next position.
position += users_with_this_position
standing[user.name] = position
@ -518,12 +647,16 @@ class PointsModel(Model):
# 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
@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)
race_result: RaceResult | None = self.race_result_by(
race_name=race_guess.race.name
)
if race_result is None:
continue
@ -545,9 +678,15 @@ class PointsModel(Model):
#
def hot_take_correct(self, user_name: str) -> bool:
season_guess_result: SeasonGuessResult | None = self.season_guess_result_by(user_name=user_name)
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
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)
@ -558,9 +697,15 @@ class PointsModel(Model):
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)
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
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)
@ -586,14 +731,23 @@ class PointsModel(Model):
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
@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)
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]
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
@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)
@ -617,7 +771,7 @@ class PointsModel(Model):
{
"data": self.points_per_step_cumulative()[user.name],
"label": user.name,
"fill": False
"fill": False,
}
for user in self.all_users()
]
@ -635,7 +789,7 @@ class PointsModel(Model):
{
"data": self.driver_points_per_step_cumulative()[driver.name],
"label": driver.abbr,
"fill": False
"fill": False,
}
for driver in self.all_drivers(include_none=False, include_inactive=True)
]
@ -653,7 +807,7 @@ class PointsModel(Model):
{
"data": self.team_points_per_step_cumulative()[team.name],
"label": team.name,
"fill": False
"fill": False,
}
for team in self.all_teams(include_none=False)
]