Compare commits

...

32 Commits

Author SHA1 Message Date
61d247508f Bug: Fix duplicated place numbering in points model
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 14s
2024-10-19 14:01:46 +02:00
f38a5f2e6d Bug: Call delete_memoized with function references
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 33s
2024-10-19 13:41:14 +02:00
95760baebf Flake: Add sqlitebrowser 2024-10-19 13:40:51 +02:00
34434cc7cc Replace memoize with cached for teams_sorted_by_points
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 2m21s
2024-09-24 00:04:36 +02:00
6da5919f86 Add a shitload of caching
Some checks are pending
Build Formula10 Docker Image / build-docker (push) Waiting to run
2024-09-24 00:01:41 +02:00
852f0a04ca Add profiler (currently disabled) 2024-09-24 00:00:41 +02:00
aa38d0138f Update FastF1 frontend button description 2024-09-23 21:23:38 +02:00
a9ab892c2a Add functions to obtain specific sessions to EventDataHelper 2024-09-23 21:23:20 +02:00
e6157893ae Remove deprecated import 2024-09-23 17:13:05 +02:00
b2a01ce453 Use column name string when sorting by column 2024-09-23 17:12:55 +02:00
5234e8a06e Add helper classes for FastF1 dataframes 2024-09-23 16:52:23 +02:00
a965c0007b Prefix OpenF1 functions 2024-09-23 16:52:11 +02:00
9acced3899 Add werkzeug to requirements 2024-09-23 16:52:00 +02:00
ed0c5f51b4 Add __repr__ to Race class 2024-09-23 16:51:18 +02:00
ebb862c9d7 Add Pandas to requirements 2024-09-23 16:50:48 +02:00
4a0eb5329c Set FastF1 cache location in dockerfile 2024-09-23 16:50:35 +02:00
30bf475000 Prefix OpenF1 classes + functions with "openf1" 2024-09-22 23:55:54 +02:00
d66087142d Add fastf1 module 2024-09-22 23:34:16 +02:00
8bdb3b5207 Add fastf1 python dependency to requirements 2024-09-22 23:01:22 +02:00
2e685ed646 Add fastf1 python dependency 2024-09-22 22:00:38 +02:00
b4c459ffe7 Allow specifying drivers as "inactive" and replace Logan Sargeant with Franco Colapinto
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 1m27s
2024-08-30 20:47:06 +02:00
3340b77efe Slight reformatting 2024-08-30 18:48:40 +02:00
0cff53cf42 Fix driver standing points string for NC
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 30s
2024-05-25 14:42:49 +02:00
f15bbb4b96 Update requirements.txt
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 28s
2024-03-10 20:10:49 +01:00
d88ae9568f Add frontend button to fetch from OpenF1
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 24s
2024-03-10 20:08:41 +01:00
1ab92eff9a Implement fetching of certain data from OpenF1 2024-03-10 19:54:28 +01:00
c2838c3332 Update flake with commands 2024-03-10 18:50:29 +01:00
e6a70365f3 Add OpenF1 model classes 2024-03-10 18:50:24 +01:00
6e20275dcd Remove obsolete notes
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 13s
2024-03-10 00:48:42 +01:00
10cf690ee2 Track sprint stats + fastest laps + add to stats
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 15s
2024-03-10 00:32:07 +01:00
628e2cd99a Update frontend for extended stats (sprint)
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 14s
2024-03-09 23:22:18 +01:00
e57109caf4 Add quali_date + has_sprint fields to race
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 15s
2024-03-09 20:56:58 +01:00
37 changed files with 1494 additions and 332 deletions

View File

@ -1,10 +1,17 @@
# syntax=docker/dockerfile:1 # syntax=docker/dockerfile:1
FROM public.ecr.aws/docker/library/python:3.11.8-slim-bookworm FROM public.ecr.aws/docker/library/python:3.11.8-slim-bookworm
RUN apt-get update -y RUN apt-get update -y
WORKDIR /app WORKDIR /app
COPY requirements.txt requirements.txt COPY requirements.txt requirements.txt
RUN pip3 install -r requirements.txt RUN pip3 install -r requirements.txt
COPY . . COPY . .
EXPOSE 5000 EXPOSE 5000
ENV FASTF1_CACHE="/cache"
CMD ["python3", "-u", "-m", "flask", "--app", "formula10", "run", "--host", "0.0.0.0"] CMD ["python3", "-u", "-m", "flask", "--app", "formula10", "run", "--host", "0.0.0.0"]

117
flake.nix
View File

@ -5,26 +5,104 @@
inputs.flake-utils.url = "github:numtide/flake-utils"; inputs.flake-utils.url = "github:numtide/flake-utils";
inputs.devshell.url = "github:numtide/devshell"; inputs.devshell.url = "github:numtide/devshell";
outputs = { self, nixpkgs, flake-utils, devshell }: outputs = {
flake-utils.lib.eachDefaultSystem (system: self,
let nixpkgs,
flake-utils,
devshell,
}:
flake-utils.lib.eachDefaultSystem (system: let
pkgs = import nixpkgs { pkgs = import nixpkgs {
inherit system; inherit system;
config.allowUnfree = true; config.allowUnfree = true;
overlays = [ devshell.overlays.default ]; overlays = [devshell.overlays.default];
}; };
myPython = pkgs.python311.withPackages (p: with p; [ timple = pkgs.python311Packages.buildPythonPackage rec {
pname = "timple";
version = "0.1.8";
src = pkgs.python311Packages.fetchPypi {
inherit pname version;
hash = "sha256-u8EgMA8BA6OpPlSg0ASRxLcIcv5psRIEcBpIicagXw8=";
};
doCheck = false;
pyproject = false;
# Build time deps
nativeBuildInputs = with pkgs.python311Packages; [
setuptools
];
# Run time deps
dependencies = with pkgs.python311Packages; [
matplotlib
numpy
];
};
fastf1 = pkgs.python311Packages.buildPythonPackage rec {
pname = "fastf1";
version = "3.4.0";
src = pkgs.python311Packages.fetchPypi {
inherit pname version;
hash = "sha256-rCjJaM0W2m9Yk3dhHkMOdIqPiKtVqoXuELBasmA9ybA=";
};
doCheck = false;
pyproject = true;
# Build time deps
nativeBuildInputs = with pkgs.python311Packages; [
hatchling
hatch-vcs
];
# Run time deps
dependencies = with pkgs.python311Packages; [
matplotlib
numpy
pandas
python-dateutil
requests
requests-cache
scipy
rapidfuzz
websockets
timple
];
};
myPython = pkgs.python311.withPackages (p:
with p; [
# Basic # Basic
rich rich
numpy
# Web # Web
flask flask
flask-sqlalchemy flask-sqlalchemy
flask-caching
sqlalchemy sqlalchemy
requests
# Test
pytest pytest
fastf1
# TODO: For some reason, listing those under fastf1.dependencies doesn't work???
matplotlib
numpy
pandas
python-dateutil
requests-cache
scipy
rapidfuzz
websockets
timple
]); ]);
in { in {
devShell = pkgs.devshell.mkShell { devShell = pkgs.devshell.mkShell {
@ -33,6 +111,8 @@
packages = with pkgs; [ packages = with pkgs; [
myPython myPython
sqlitebrowser
nodejs_21 nodejs_21
nodePackages.sass nodePackages.sass
nodePackages.postcss-cli nodePackages.postcss-cli
@ -41,11 +121,26 @@
# Use $1 for positional args # Use $1 for positional args
commands = [ commands = [
# { {
# name = ""; name = "vscode";
# help = ""; help = "Launch VSCode";
# command = ""; command = "code . &>/dev/null &";
# } }
{
name = "pycharm";
help = "Launch PyCharm Professional";
command = "pycharm-professional . &>/dev/null &";
}
{
name = "db";
help = "Launch SQLiteBrowser";
command = "sqlitebrowser ./instance/formula10.db &>/dev/null &";
}
{
name = "api";
help = "Launch Hoppscotch in Google Chrome";
command = "google-chrome-stable https://hoppscotch.io &>/dev/null &";
}
]; ];
}; };
}); });

View File

@ -1,6 +1,8 @@
import os import os
from flask import Flask from flask import Flask
from flask_sqlalchemy import SQLAlchemy from flask_sqlalchemy import SQLAlchemy
from flask_caching import Cache
from werkzeug.middleware.profiler import ProfilerMiddleware
# Load local ENV variables (can be set when calling the executable) # Load local ENV variables (can be set when calling the executable)
ENABLE_TIMING: bool = False if os.getenv("DISABLE_TIMING") == "True" else True ENABLE_TIMING: bool = False if os.getenv("DISABLE_TIMING") == "True" else True
@ -9,18 +11,23 @@ if not ENABLE_TIMING:
print("- Disabled timing constraints") print("- Disabled timing constraints")
app: Flask = Flask(__name__) app: Flask = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///formula10.db" app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///formula10.db"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
# Session cookie is used to propagate message to error page # Session cookie is used to propagate message to error page
app.config['SESSION_TYPE'] = 'memcached' app.config["SESSION_TYPE"] = "memcached"
app.config['SECRET_KEY'] = 'ich stinke nach maggi' app.config["SECRET_KEY"] = "ich stinke nach maggi"
app.url_map.strict_slashes = False app.url_map.strict_slashes = False
db: SQLAlchemy = SQLAlchemy() db: SQLAlchemy = SQLAlchemy()
db.init_app(app) db.init_app(app)
cache: Cache = Cache(config={"CACHE_TYPE": "SimpleCache"})
cache.init_app(app)
# app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=("/formula10/*",), sort_by=("cumtime",))
# NOTE: These imports are required to register the routes. They need to be imported after "app" is declared # NOTE: These imports are required to register the routes. They need to be imported after "app" is declared
import formula10.controller.race_controller # type: ignore import formula10.controller.race_controller # type: ignore
import formula10.controller.season_controller import formula10.controller.season_controller

View File

@ -3,10 +3,15 @@ from urllib.parse import unquote
from flask import redirect, render_template, request from flask import redirect, render_template, request
from werkzeug import Response from werkzeug import Response
from formula10.controller.error_controller import error_redirect
from formula10.database.update_queries import update_race_result, update_user from formula10.database.update_queries import update_race_result, update_user
from formula10.domain.cache_invalidator import cache_invalidate_user_updated, cache_invalidate_race_result_updated
from formula10.domain.domain_model import Model from formula10.domain.domain_model import Model
from formula10.domain.template_model import TemplateModel from formula10.domain.template_model import TemplateModel
from formula10 import app from formula10 import app
from formula10.openf1.model.openf1_session import OpenF1Session
from formula10.openf1.openf1_definitions import OPENF1_SESSION_NAME_RACE
from formula10.openf1.openf1_fetcher import openf1_fetch_driver, openf1_fetch_position, openf1_fetch_session
@app.route("/result") @app.route("/result")
@ -31,9 +36,29 @@ def result_enter_post(race_name: str) -> Response:
dnfs: List[str] = request.form.getlist("dnf-drivers") dnfs: List[str] = request.form.getlist("dnf-drivers")
excluded: List[str] = request.form.getlist("excluded-drivers") excluded: List[str] = request.form.getlist("excluded-drivers")
# @todo Ugly # Extra stats for points calculation
fastest_lap: str | None = request.form.get("fastest-lap")
sprint_pxxs: List[str] = request.form.getlist("sprint-pxx-drivers")
sprint_dnf_drivers: List[str] = request.form.getlist("sprint-dnf-drivers")
if fastest_lap is None:
return error_redirect("Data was not saved, because fastest lap was not set.")
cache_invalidate_race_result_updated()
race_id: int = Model().race_by(race_name=race_name).id race_id: int = Model().race_by(race_name=race_name).id
return update_race_result(race_id, pxxs, first_dnfs, dnfs, excluded) return update_race_result(race_id, pxxs, first_dnfs, dnfs, excluded, int(fastest_lap), sprint_pxxs, sprint_dnf_drivers)
@app.route("/result-fetch/<race_name>", methods=["POST"])
def result_fetch_post(race_name: str) -> Response:
session: OpenF1Session = openf1_fetch_session(OPENF1_SESSION_NAME_RACE, "KSA")
openf1_fetch_driver(session.session_key, "VER")
openf1_fetch_position(session.session_key, 1)
# @todo Fetch stuff and build the race_result using update_race_result(...)
cache_invalidate_race_result_updated()
return redirect("/result")
@app.route("/user") @app.route("/user")
@ -46,11 +71,13 @@ def user_root() -> str:
@app.route("/user-add", methods=["POST"]) @app.route("/user-add", methods=["POST"])
def user_add_post() -> Response: def user_add_post() -> Response:
cache_invalidate_user_updated()
username: str | None = request.form.get("select-add-user") username: str | None = request.form.get("select-add-user")
return update_user(username, add=True) return update_user(username, add=True)
@app.route("/user-delete", methods=["POST"]) @app.route("/user-delete", methods=["POST"])
def user_delete_post() -> Response: def user_delete_post() -> Response:
cache_invalidate_user_updated()
username: str | None = request.form.get("select-delete-user") username: str | None = request.form.get("select-delete-user")
return update_user(username, delete=True) return update_user(username, delete=True)

View File

@ -1,8 +1,10 @@
from typing import List
from urllib.parse import unquote from urllib.parse import unquote
from flask import redirect, render_template, request from flask import redirect, render_template, request
from werkzeug import Response from werkzeug import Response
from formula10.database.update_queries import delete_race_guess, update_race_guess from formula10.database.update_queries import delete_race_guess, update_race_guess
from formula10.domain.cache_invalidator import cache_invalidate_race_guess_updated
from formula10.domain.domain_model import Model from formula10.domain.domain_model import Model
from formula10.domain.points_model import PointsModel from formula10.domain.points_model import PointsModel
from formula10.domain.template_model import TemplateModel from formula10.domain.template_model import TemplateModel
@ -37,6 +39,7 @@ def race_guess_post(race_name: str, user_name: str) -> Response:
pxx: str | None = request.form.get("pxxselect") pxx: str | None = request.form.get("pxxselect")
dnf: str | None = request.form.get("dnfselect") dnf: str | None = request.form.get("dnfselect")
cache_invalidate_race_guess_updated()
race_id: int = Model().race_by(race_name=race_name).id race_id: int = Model().race_by(race_name=race_name).id
user_id: int = Model().user_by(user_name=user_name).id user_id: int = Model().user_by(user_name=user_name).id
return update_race_guess(race_id, user_id, return update_race_guess(race_id, user_id,
@ -49,6 +52,7 @@ def race_guess_delete_post(race_name: str, user_name: str) -> Response:
race_name = unquote(race_name) race_name = unquote(race_name)
user_name = unquote(user_name) user_name = unquote(user_name)
cache_invalidate_race_guess_updated()
race_id: int = Model().race_by(race_name=race_name).id race_id: int = Model().race_by(race_name=race_name).id
user_id: int = Model().user_by(user_name=user_name).id user_id: int = Model().user_by(user_name=user_name).id
return delete_race_guess(race_id, user_id) return delete_race_guess(race_id, user_id)

View File

@ -5,6 +5,7 @@ from werkzeug import Response
from formula10.database.model.db_team import DbTeam from formula10.database.model.db_team import DbTeam
from formula10.database.update_queries import update_season_guess from formula10.database.update_queries import update_season_guess
from formula10.domain.cache_invalidator import cache_invalidate_season_guess_updated
from formula10.domain.domain_model import Model from formula10.domain.domain_model import Model
from formula10.domain.model.team import NONE_TEAM from formula10.domain.model.team import NONE_TEAM
from formula10.domain.points_model import PointsModel from formula10.domain.points_model import PointsModel
@ -44,5 +45,6 @@ def season_guess_post(user_name: str) -> Response:
] ]
podium_driver_guesses: List[str] = request.form.getlist("podiumdrivers") podium_driver_guesses: List[str] = request.form.getlist("podiumdrivers")
cache_invalidate_season_guess_updated()
user_id: int = Model().user_by(user_name=user_name).id user_id: int = Model().user_by(user_name=user_name).id
return update_season_guess(user_id, guesses, team_winner_guesses, podium_driver_guesses) return update_season_guess(user_id, guesses, team_winner_guesses, podium_driver_guesses)

View File

@ -1,4 +1,4 @@
from sqlalchemy import Integer, String, ForeignKey from sqlalchemy import Integer, String, ForeignKey, Boolean
from sqlalchemy.orm import mapped_column, Mapped, relationship from sqlalchemy.orm import mapped_column, Mapped, relationship
from formula10.database.model.db_team import DbTeam from formula10.database.model.db_team import DbTeam
@ -20,6 +20,7 @@ class DbDriver(db.Model):
abbr: Mapped[str] = mapped_column(String(4), nullable=False, unique=True) abbr: Mapped[str] = mapped_column(String(4), nullable=False, unique=True)
team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), nullable=False) team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), nullable=False)
country_code: Mapped[str] = mapped_column(String(2), nullable=False) # alpha-2 code country_code: Mapped[str] = mapped_column(String(2), nullable=False) # alpha-2 code
active: Mapped[bool] = mapped_column(Boolean, nullable=False)
# Relationships # Relationships
team: Mapped[DbTeam] = relationship("DbTeam", foreign_keys=[team_id]) team: Mapped[DbTeam] = relationship("DbTeam", foreign_keys=[team_id])

View File

@ -1,5 +1,5 @@
from datetime import datetime from datetime import datetime
from sqlalchemy import DateTime, Integer, String from sqlalchemy import Boolean, DateTime, Integer, String
from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.orm import Mapped, mapped_column
from formula10 import db from formula10 import db
@ -20,3 +20,5 @@ class DbRace(db.Model):
number: Mapped[int] = mapped_column(Integer, nullable=False, unique=True) number: Mapped[int] = mapped_column(Integer, nullable=False, unique=True)
date: Mapped[datetime] = mapped_column(DateTime, nullable=False, unique=True) date: Mapped[datetime] = mapped_column(DateTime, nullable=False, unique=True)
pxx: Mapped[int] = mapped_column(Integer, nullable=False) # This is the place to guess pxx: Mapped[int] = mapped_column(Integer, nullable=False) # This is the place to guess
quali_date: Mapped[datetime] = mapped_column(DateTime, nullable=False, unique=True)
has_sprint: Mapped[bool] = mapped_column(Boolean, nullable=False)

View File

@ -13,7 +13,6 @@ from formula10.database.model.db_season_guess import DbSeasonGuess
from formula10.database.model.db_user import DbUser from formula10.database.model.db_user import DbUser
from formula10.database.validation import any_is_none, positions_are_contiguous, race_has_started from formula10.database.validation import any_is_none, positions_are_contiguous, race_has_started
from formula10 import ENABLE_TIMING, db from formula10 import ENABLE_TIMING, db
from formula10.domain.model.driver import NONE_DRIVER
def find_or_create_race_guess(user_id: int, race_id: int) -> DbRaceGuess: def find_or_create_race_guess(user_id: int, race_id: int) -> DbRaceGuess:
@ -129,8 +128,8 @@ def find_or_create_race_result(race_id: int) -> DbRaceResult:
race_result.excluded_driver_ids_json = json.dumps(["9999"]) race_result.excluded_driver_ids_json = json.dumps(["9999"])
race_result.fastest_lap_id = 9999 race_result.fastest_lap_id = 9999
race_result.sprint_dnf_driver_ids_json = json.dumps(["9999"]) race_result.sprint_dnf_driver_ids_json = json.dumps([])
race_result.sprint_points_json = json.dumps({"9999": "9999"}) race_result.sprint_points_json = json.dumps({})
db.session.add(race_result) db.session.add(race_result)
db.session.commit() db.session.commit()
@ -143,21 +142,22 @@ def find_or_create_race_result(race_id: int) -> DbRaceResult:
return race_result return race_result
def update_race_result(race_id: int, pxx_driver_ids_list: List[str], first_dnf_driver_ids_list: List[str], dnf_driver_ids_list: List[str], excluded_driver_ids_list: List[str]) -> Response: def update_race_result(race_id: int, pxx_driver_ids_list: List[str], first_dnf_driver_ids_list: List[str], dnf_driver_ids_list: List[str], excluded_driver_ids_list: List[str],
fastest_lap_driver_id: int, sprint_pxx_driver_ids_list: List[str], sprint_dnf_driver_ids_list: List[str]) -> Response:
if ENABLE_TIMING and not race_has_started(race_id=race_id): if ENABLE_TIMING and not race_has_started(race_id=race_id):
return error_redirect("No race result can be entered, as the race has not begun!") return error_redirect("No race result can be entered, as the race has not begun!")
# Use strings as keys, as these dicts will be serialized to json # Use strings as keys, as these dicts will be serialized to json
pxx_driver_names: Dict[str, str] = { pxx_driver_ids: Dict[str, str] = {
str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list) str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list)
} }
# Not counted drivers have to be at the end # Not counted drivers have to be at the end
excluded_driver_names: Dict[str, str] = { excluded_driver_ids: Dict[str, str] = {
str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list) str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list)
if driver_id in excluded_driver_ids_list if driver_id in excluded_driver_ids_list
} }
if len(excluded_driver_names) > 0 and (not "20" in excluded_driver_names or not positions_are_contiguous(list(excluded_driver_names.keys()))): if len(excluded_driver_ids) > 0 and (not "20" in excluded_driver_ids or not positions_are_contiguous(list(excluded_driver_ids.keys()))):
return error_redirect("Race result was not saved, as excluded drivers must be contiguous and at the end of the field!") return error_redirect("Race result was not saved, as excluded drivers must be contiguous and at the end of the field!")
# First DNF drivers have to be contained in DNF drivers # First DNF drivers have to be contained in DNF drivers
@ -170,15 +170,19 @@ def update_race_result(race_id: int, pxx_driver_ids_list: List[str], first_dnf_d
return error_redirect("Race result was not saved, as there cannot be DNFs without (an) initial DNF(s)!") return error_redirect("Race result was not saved, as there cannot be DNFs without (an) initial DNF(s)!")
race_result: DbRaceResult = find_or_create_race_result(race_id) race_result: DbRaceResult = find_or_create_race_result(race_id)
race_result.pxx_driver_ids_json = json.dumps(pxx_driver_names) race_result.pxx_driver_ids_json = json.dumps(pxx_driver_ids)
race_result.first_dnf_driver_ids_json = json.dumps(first_dnf_driver_ids_list) race_result.first_dnf_driver_ids_json = json.dumps(first_dnf_driver_ids_list)
race_result.dnf_driver_ids_json = json.dumps(dnf_driver_ids_list) race_result.dnf_driver_ids_json = json.dumps(dnf_driver_ids_list)
race_result.excluded_driver_ids_json = json.dumps(excluded_driver_ids_list) race_result.excluded_driver_ids_json = json.dumps(excluded_driver_ids_list)
# @todo Dummy values # Extra stats for points calculation
race_result.fastest_lap_id = NONE_DRIVER.id sprint_pxx_driver_ids: Dict[str, str] = {
race_result.sprint_dnf_driver_ids_json = json.dumps([NONE_DRIVER.id]) str(position + 1): driver_id for position, driver_id in enumerate(sprint_pxx_driver_ids_list)
race_result.sprint_points_json = json.dumps({NONE_DRIVER.id: 0}) }
race_result.fastest_lap_id = fastest_lap_driver_id
race_result.sprint_dnf_driver_ids_json = json.dumps(sprint_dnf_driver_ids_list)
race_result.sprint_points_json = json.dumps(sprint_pxx_driver_ids)
db.session.commit() db.session.commit()

View File

@ -76,6 +76,19 @@ def find_multiple_strict(predicate: Callable[[_T], bool], iterable: Iterable[_T]
return filtered return filtered
def find_atleast_strict(predicate: Callable[[_T], bool], iterable: Iterable[_T], count: int = 0) -> List[_T]:
"""
Finds at least <count> elements in a sequence matching a predicate.
Throws exception if fewer elements were found than specified.
"""
filtered = list(filter(predicate, iterable))
if len(filtered) < count:
raise Exception(f"find_atleast found {len(filtered)} matching elements but expected at least {count}")
return filtered
def find_single_strict(predicate: Callable[[_T], bool], iterable: Iterable[_T]) -> _T: def find_single_strict(predicate: Callable[[_T], bool], iterable: Iterable[_T]) -> _T:
""" """
Find a single element in a sequence matching a predicate. Find a single element in a sequence matching a predicate.

View File

@ -0,0 +1,98 @@
from typing import Callable, List
from formula10 import cache
from formula10.domain.domain_model import Model
from formula10.domain.points_model import PointsModel
def cache_invalidate_user_updated() -> None:
caches: List[str] = [
"domain_all_users",
"domain_all_race_guesses",
"domain_all_season_guesses",
"points_points_per_step",
"points_user_standing",
]
memoized_caches: List[Callable] = [
PointsModel.points_by,
PointsModel.race_guesses_by,
PointsModel.season_guesses_by,
]
for c in caches:
cache.delete(c)
for c in memoized_caches:
cache.delete_memoized(c)
def cache_invalidate_race_result_updated() -> None:
caches: List[str] = [
"domain_all_race_results",
"points_points_per_step",
"points_team_points_per_step",
"points_dnfs",
"points_driver_points_per_step_cumulative",
"points_wdc_standing_by_position",
"points_wdc_standing_by_driver",
"points_most_dnf_names",
"points_most_gained_names",
"points_most_lost_names",
"points_team_points_per_step_cumulative",
"points_teams_sorted_by_points",
"points_wcc_standing_by_position",
"points_wcc_standing_by_team",
"points_user_standing",
"template_first_race_without_result",
]
memoized_caches: List[Callable] = [
PointsModel.driver_points_per_step,
PointsModel.driver_points_by,
PointsModel.total_driver_points_by,
PointsModel.drivers_sorted_by_points,
PointsModel.total_team_points_by,
PointsModel.points_by,
PointsModel.is_team_winner,
PointsModel.has_podium,
PointsModel.picks_with_points_count,
]
for c in caches:
cache.delete(c)
for c in memoized_caches:
cache.delete_memoized(c)
def cache_invalidate_race_guess_updated() -> None:
caches: List[str] = [
"domain_all_race_guesses",
]
memoized_caches: List[Callable] = [
Model.race_guesses_by,
]
for c in caches:
cache.delete(c)
for c in memoized_caches:
cache.delete_memoized(c)
def cache_invalidate_season_guess_updated() -> None:
caches: List[str] = [
"domain_all_season_guesses",
]
memoized_caches: List[Callable] = [
Model.season_guesses_by,
]
for c in caches:
cache.delete(c)
for c in memoized_caches:
cache.delete_memoized(c)

View File

@ -9,7 +9,7 @@ from formula10.database.model.db_season_guess import DbSeasonGuess
from formula10.database.model.db_season_guess_result import DbSeasonGuessResult from formula10.database.model.db_season_guess_result import DbSeasonGuessResult
from formula10.database.model.db_team import DbTeam from formula10.database.model.db_team import DbTeam
from formula10.database.model.db_user import DbUser from formula10.database.model.db_user import DbUser
from formula10.database.validation import find_multiple_strict, find_single_or_none_strict, find_single_strict from formula10.database.validation import find_multiple_strict, find_single_or_none_strict, find_single_strict, find_atleast_strict
from formula10.domain.model.driver import NONE_DRIVER, Driver from formula10.domain.model.driver import NONE_DRIVER, Driver
from formula10.domain.model.race import Race from formula10.domain.model.race import Race
from formula10.domain.model.race_guess import RaceGuess from formula10.domain.model.race_guess import RaceGuess
@ -18,119 +18,97 @@ from formula10.domain.model.season_guess import SeasonGuess
from formula10.domain.model.season_guess_result import SeasonGuessResult from formula10.domain.model.season_guess_result import SeasonGuessResult
from formula10.domain.model.team import NONE_TEAM, Team from formula10.domain.model.team import NONE_TEAM, Team
from formula10.domain.model.user import User from formula10.domain.model.user import User
from formula10 import db from formula10 import db, cache
class Model(): class Model:
_all_users: List[User] | None = None @staticmethod
_all_race_results: List[RaceResult] | None = None @cache.cached(timeout=None, key_prefix="domain_all_users") # Clear when adding/deleting users
_all_race_guesses: List[RaceGuess] | None = None def all_users() -> List[User]:
_all_season_guesses: List[SeasonGuess] | None = None
_all_season_guess_results: List[SeasonGuessResult] | None = None
_all_races: List[Race] | None = None
_all_drivers: List[Driver] | None = None
_all_teams: List[Team] | None = None
def all_users(self) -> List[User]:
""" """
Returns a list of all enabled users. Returns a list of all enabled users.
""" """
if self._all_users is None: db_users = db.session.query(DbUser).filter_by(enabled=True).all()
self._all_users = [ return [User.from_db_user(db_user) for db_user in db_users]
User.from_db_user(db_user)
for db_user in db.session.query(DbUser).filter_by(enabled=True).all()
]
return self._all_users @staticmethod
@cache.cached(timeout=None, key_prefix="domain_all_race_results") # Clear when adding/updating results
def all_race_results(self) -> List[RaceResult]: def all_race_results() -> List[RaceResult]:
""" """
Returns a list of all race results, in descending order (most recent first). Returns a list of all race results, in descending order (most recent first).
""" """
if self._all_race_results is None: db_race_results = db.session.query(DbRaceResult).join(DbRaceResult.race).order_by(desc("number")).all()
self._all_race_results = [ return [RaceResult.from_db_race_result(db_race_result) for db_race_result in db_race_results]
RaceResult.from_db_race_result(db_race_result)
for db_race_result in db.session.query(DbRaceResult).join(DbRaceResult.race).order_by(desc(DbRace.number)).all()
]
return self._all_race_results @staticmethod
@cache.cached(timeout=None, key_prefix="domain_all_race_guesses") # Clear when adding/updating race guesses or users
def all_race_guesses(self) -> List[RaceGuess]: def all_race_guesses() -> List[RaceGuess]:
""" """
Returns a list of all race guesses (of enabled users). Returns a list of all race guesses (of enabled users).
""" """
if self._all_race_guesses is None: db_race_guesses = db.session.query(DbRaceGuess).join(DbRaceGuess.user).filter_by(enabled=True).all()
self._all_race_guesses = [ return [RaceGuess.from_db_race_guess(db_race_guess) for db_race_guess in db_race_guesses]
RaceGuess.from_db_race_guess(db_race_guess)
for db_race_guess in db.session.query(DbRaceGuess).join(DbRaceGuess.user).filter_by(enabled=True).all() # Ignore disabled users
]
return self._all_race_guesses @staticmethod
@cache.cached(timeout=None, key_prefix="domain_all_season_guesses") # Clear when adding/updating season guesses or users
def all_season_guesses(self) -> List[SeasonGuess]: def all_season_guesses() -> List[SeasonGuess]:
""" """
Returns a list of all season guesses (of enabled users). Returns a list of all season guesses (of enabled users).
""" """
if self._all_season_guesses is None: db_season_guesses = db.session.query(DbSeasonGuess).join(DbSeasonGuess.user).filter_by(enabled=True).all()
self._all_season_guesses = [ return [SeasonGuess.from_db_season_guess(db_season_guess) for db_season_guess in db_season_guesses]
SeasonGuess.from_db_season_guess(db_season_guess)
for db_season_guess in db.session.query(DbSeasonGuess).join(DbSeasonGuess.user).filter_by(enabled=True).all() # Ignore disabled users
]
return self._all_season_guesses @staticmethod
@cache.cached(timeout=None, key_prefix="domain_all_season_guess_results") # No cleanup, bc entered manually
def all_season_guess_results() -> List[SeasonGuessResult]:
"""
Returns a list of all season guess results (of enabled users).
"""
db_season_guess_results = db.session.query(DbSeasonGuessResult).join(DbSeasonGuessResult.user).filter_by(enabled=True).all()
return [SeasonGuessResult.from_db_season_guess_result(db_season_guess_result) for db_season_guess_result in db_season_guess_results]
def all_season_guess_results(self) -> List[SeasonGuessResult]: @staticmethod
if self._all_season_guess_results is None: @cache.cached(timeout=None, key_prefix="domain_all_races") # No cleanup, bc entered manually
self._all_season_guess_results = [ def all_races() -> List[Race]:
SeasonGuessResult.from_db_season_guess_result(db_season_guess_result)
for db_season_guess_result in db.session.query(DbSeasonGuessResult).join(DbSeasonGuessResult.user).filter_by(enabled=True).all() # Ignore disabled users
]
return self._all_season_guess_results
def all_races(self) -> List[Race]:
""" """
Returns a list of all races, in descending order (last race first). Returns a list of all races, in descending order (last race first).
""" """
if self._all_races is None: db_races = db.session.query(DbRace).order_by(desc("number")).all()
self._all_races = [ return [Race.from_db_race(db_race) for db_race in db_races]
Race.from_db_race(db_race)
for db_race in db.session.query(DbRace).order_by(desc(DbRace.number)).all()
]
return self._all_races @staticmethod
@cache.memoize(timeout=None) # No cleanup, bc entered manually
def all_drivers(self, *, include_none: bool) -> List[Driver]: def all_drivers(*, include_none: bool, include_inactive: bool) -> List[Driver]:
""" """
Returns a list of all drivers. Returns a list of all active drivers.
""" """
if self._all_drivers is None: db_drivers = db.session.query(DbDriver).all()
self._all_drivers = [ drivers = [Driver.from_db_driver(db_driver) for db_driver in db_drivers]
Driver.from_db_driver(db_driver)
for db_driver in db.session.query(DbDriver).all()
]
if include_none: if not include_inactive:
return self._all_drivers predicate: Callable[[Driver], bool] = lambda driver: driver.active
else: drivers = find_multiple_strict(predicate, drivers)
if not include_none:
predicate: Callable[[Driver], bool] = lambda driver: driver != NONE_DRIVER predicate: Callable[[Driver], bool] = lambda driver: driver != NONE_DRIVER
return find_multiple_strict(predicate, self._all_drivers) drivers = find_multiple_strict(predicate, drivers)
def all_teams(self, *, include_none: bool) -> List[Team]: return drivers
@staticmethod
@cache.memoize(timeout=None) # No cleanup, bc entered manually
def all_teams(*, include_none: bool) -> List[Team]:
""" """
Returns a list of all teams. Returns a list of all teams.
""" """
if self._all_teams is None: db_teams = db.session.query(DbTeam).all()
self._all_teams = [ teams = [Team.from_db_team(db_team) for db_team in db_teams]
Team.from_db_team(db_team)
for db_team in db.session.query(DbTeam).all()
]
if include_none: if not include_none:
return self._all_teams
else:
predicate: Callable[[Team], bool] = lambda team: team != NONE_TEAM predicate: Callable[[Team], bool] = lambda team: team != NONE_TEAM
return find_multiple_strict(predicate, self._all_teams) return find_multiple_strict(predicate, teams)
return teams
# #
# User queries # User queries
@ -203,6 +181,7 @@ class Model():
""" """
return self.race_guesses_by() return self.race_guesses_by()
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race guesses or users
def race_guesses_by(self, *, user_name: str | None = None, race_name: str | None = None) -> RaceGuess | List[RaceGuess] | Dict[str, Dict[str, RaceGuess]] | None: def race_guesses_by(self, *, user_name: str | None = None, race_name: str | None = None) -> RaceGuess | List[RaceGuess] | Dict[str, Dict[str, RaceGuess]] | None:
# List of all guesses by a single user # List of all guesses by a single user
if user_name is not None and race_name is None: if user_name is not None and race_name is None:
@ -252,6 +231,7 @@ class Model():
""" """
return self.season_guesses_by() return self.season_guesses_by()
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating season guesses or users
def season_guesses_by(self, *, user_name: str | None = None) -> SeasonGuess | Dict[str, SeasonGuess] | None: def season_guesses_by(self, *, user_name: str | None = None) -> SeasonGuess | Dict[str, SeasonGuess] | None:
if user_name is not None: if user_name is not None:
predicate: Callable[[SeasonGuess], bool] = lambda guess: guess.user.name == user_name predicate: Callable[[SeasonGuess], bool] = lambda guess: guess.user.name == user_name
@ -280,34 +260,41 @@ class Model():
# Team queries # Team queries
# #
def none_team(self) -> Team: @staticmethod
def none_team() -> Team:
return NONE_TEAM return NONE_TEAM
# #
# Driver queries # Driver queries
# #
def none_driver(self) -> Driver: @staticmethod
def none_driver() -> Driver:
return NONE_DRIVER return NONE_DRIVER
@overload @overload
def drivers_by(self, *, team_name: str) -> List[Driver]: def drivers_by(self, *, team_name: str, include_inactive: bool) -> List[Driver]:
""" """
Returns a list of all drivers driving for a certain team. Returns a list of all drivers driving for a certain team.
""" """
return self.drivers_by(team_name=team_name) return self.drivers_by(team_name=team_name, include_inactive=include_inactive)
@overload @overload
def drivers_by(self) -> Dict[str, List[Driver]]: def drivers_by(self, *, include_inactive: bool) -> Dict[str, List[Driver]]:
""" """
Returns a dictionary of drivers mapped to team names. Returns a dictionary of drivers mapped to team names.
""" """
return self.drivers_by() return self.drivers_by(include_inactive=include_inactive)
def drivers_by(self, *, team_name: str | None = None) -> List[Driver] | Dict[str, List[Driver]]: @cache.memoize(timeout=None, args_to_ignore=["self"]) # No Cleanup, data added manually
def drivers_by(self, *, team_name: str | None = None, include_inactive: bool) -> List[Driver] | Dict[str, List[Driver]]:
if team_name is not None: if team_name is not None:
predicate: Callable[[Driver], bool] = lambda driver: driver.team.name == team_name predicate: Callable[[Driver], bool] = lambda driver: driver.team.name == team_name
return find_multiple_strict(predicate, self.all_drivers(include_none=False), 2)
if include_inactive:
return find_atleast_strict(predicate, self.all_drivers(include_none=False, include_inactive=True), 2)
else:
return find_multiple_strict(predicate, self.all_drivers(include_none=False, include_inactive=False), 2)
if team_name is None: if team_name is None:
drivers_by: Dict[str, List[Driver]] = dict() drivers_by: Dict[str, List[Driver]] = dict()
@ -316,7 +303,7 @@ class Model():
for team in self.all_teams(include_none=False): for team in self.all_teams(include_none=False):
drivers_by[team.name] = [] drivers_by[team.name] = []
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=include_inactive):
drivers_by[driver.team.name] += [driver] drivers_by[driver.team.name] += [driver]
return drivers_by return drivers_by

View File

@ -4,7 +4,7 @@ from formula10.database.model.db_driver import DbDriver
from formula10.domain.model.team import NONE_TEAM, Team from formula10.domain.model.team import NONE_TEAM, Team
class Driver(): class Driver:
@classmethod @classmethod
def from_db_driver(cls, db_driver: DbDriver): def from_db_driver(cls, db_driver: DbDriver):
driver: Driver = cls() driver: Driver = cls()
@ -13,6 +13,7 @@ class Driver():
driver.abbr = db_driver.abbr driver.abbr = db_driver.abbr
driver.country = db_driver.country_code driver.country = db_driver.country_code
driver.team = Team.from_db_team(db_driver.team) driver.team = Team.from_db_team(db_driver.team)
driver.active = db_driver.active
return driver return driver
def to_db_driver(self) -> DbDriver: def to_db_driver(self) -> DbDriver:
@ -21,6 +22,7 @@ class Driver():
db_driver.abbr = self.abbr db_driver.abbr = self.abbr
db_driver.country_code = self.country db_driver.country_code = self.country
db_driver.team_id = self.team.name db_driver.team_id = self.team.name
db_driver.active = self.active
return db_driver return db_driver
def __eq__(self, __value: object) -> bool: def __eq__(self, __value: object) -> bool:
@ -32,11 +34,17 @@ class Driver():
def __hash__(self) -> int: def __hash__(self) -> int:
return hash(self.id) return hash(self.id)
# This is important to memoize functions getting a Driver as input.
# The repr() will be appended to the cache key.
def __repr__(self) -> str:
return f"Driver(id={self.id}, name={self.name})"
id: int id: int
name: str name: str
abbr: str abbr: str
country: str country: str
team: Team team: Team
active: bool
@property @property
def name_sanitized(self) -> str: def name_sanitized(self) -> str:
@ -49,3 +57,4 @@ NONE_DRIVER.name = "None"
NONE_DRIVER.abbr = "None" NONE_DRIVER.abbr = "None"
NONE_DRIVER.country = "NO" NONE_DRIVER.country = "NO"
NONE_DRIVER.team = NONE_TEAM NONE_DRIVER.team = NONE_TEAM
NONE_DRIVER.active = True

View File

@ -4,7 +4,7 @@ from urllib.parse import quote
from formula10.database.model.db_race import DbRace from formula10.database.model.db_race import DbRace
class Race(): class Race:
@classmethod @classmethod
def from_db_race(cls, db_race: DbRace): def from_db_race(cls, db_race: DbRace):
race: Race = cls() race: Race = cls()
@ -13,6 +13,8 @@ class Race():
race.number = db_race.number race.number = db_race.number
race.date = db_race.date race.date = db_race.date
race.place_to_guess = db_race.pxx race.place_to_guess = db_race.pxx
race.quali_date = db_race.quali_date
race.has_sprint = db_race.has_sprint
return race return race
def to_db_race(self) -> DbRace: def to_db_race(self) -> DbRace:
@ -21,6 +23,8 @@ class Race():
db_race.number = self.number db_race.number = self.number
db_race.date = self.date db_race.date = self.date
db_race.pxx = self.place_to_guess db_race.pxx = self.place_to_guess
db_race.quali_date = self.date
db_race.has_sprint = self.has_sprint
return db_race return db_race
def __eq__(self, __value: object) -> bool: def __eq__(self, __value: object) -> bool:
@ -32,11 +36,16 @@ class Race():
def __hash__(self) -> int: def __hash__(self) -> int:
return hash(self.id) return hash(self.id)
def __repr__(self) -> str:
return f"race(\n\tid={self.id}, name={self.name}, number={self.number},\n\tdate={self.date}, quali_date={self.quali_date},\n\thas_sprint={self.has_sprint}, place_to_guess={self.place_to_guess}\n)"
id: int id: int
name: str name: str
number: int number: int
date: datetime date: datetime
place_to_guess: int place_to_guess: int
quali_date: datetime
has_sprint: bool
@property @property
def name_sanitized(self) -> str: def name_sanitized(self) -> str:

View File

@ -4,7 +4,7 @@ from formula10.domain.model.race import Race
from formula10.domain.model.user import User from formula10.domain.model.user import User
class RaceGuess(): class RaceGuess:
@classmethod @classmethod
def from_db_race_guess(cls, db_race_guess: DbRaceGuess): def from_db_race_guess(cls, db_race_guess: DbRaceGuess):
race_guess: RaceGuess = cls() race_guess: RaceGuess = cls()

View File

@ -154,6 +154,8 @@ class RaceResult:
return points_strings[position_offset] return points_strings[position_offset]
else: else:
return "0 Points" return "0 Points"
elif driver == _driver and driver in self.standing_exclusions:
return "0 Points"
raise Exception(f"Could not get points string for driver {driver.name}") raise Exception(f"Could not get points string for driver {driver.name}")
@ -174,6 +176,11 @@ class RaceResult:
self.standing[str(position)] for position in range(1, 21) self.standing[str(position)] for position in range(1, 21)
] ]
def ordered_sprint_standing_list(self) -> List[Driver]:
return [
self.sprint_standing[str(position)] for position in range(1, 21)
]
def initial_dnf_string(self) -> str: def initial_dnf_string(self) -> str:
if len(self.initial_dnf) == 0: if len(self.initial_dnf) == 0:
return NONE_DRIVER.name return NONE_DRIVER.name

View File

@ -7,7 +7,7 @@ from formula10.domain.model.team import Team
from formula10.domain.model.user import User from formula10.domain.model.user import User
class SeasonGuess(): class SeasonGuess:
@classmethod @classmethod
def from_db_season_guess(cls, db_season_guess: DbSeasonGuess): def from_db_season_guess(cls, db_season_guess: DbSeasonGuess):
season_guess: SeasonGuess = cls() season_guess: SeasonGuess = cls()

View File

@ -2,7 +2,7 @@ from formula10.database.model.db_season_guess_result import DbSeasonGuessResult
from formula10.domain.model.user import User from formula10.domain.model.user import User
class SeasonGuessResult(): class SeasonGuessResult:
@classmethod @classmethod
def from_db_season_guess_result(cls, db_season_guess_result: DbSeasonGuessResult): def from_db_season_guess_result(cls, db_season_guess_result: DbSeasonGuessResult):
season_guess_result: SeasonGuessResult = cls() season_guess_result: SeasonGuessResult = cls()

View File

@ -3,7 +3,7 @@ from urllib.parse import quote
from formula10.database.model.db_team import DbTeam from formula10.database.model.db_team import DbTeam
class Team(): class Team:
@classmethod @classmethod
def from_db_team(cls, db_team: DbTeam): def from_db_team(cls, db_team: DbTeam):
team: Team = cls() team: Team = cls()

View File

@ -3,7 +3,7 @@ from urllib.parse import quote
from formula10.database.model.db_user import DbUser from formula10.database.model.db_user import DbUser
class User(): class User:
@classmethod @classmethod
def from_db_user(cls, db_user: DbUser): def from_db_user(cls, db_user: DbUser):
user: User = cls() user: User = cls()

View File

@ -2,6 +2,7 @@ import json
from typing import Any, Callable, Dict, List, overload from typing import Any, Callable, Dict, List, overload
import numpy as np import numpy as np
from formula10 import cache
from formula10.domain.domain_model import Model from formula10.domain.domain_model import Model
from formula10.domain.model.driver import NONE_DRIVER, Driver from formula10.domain.model.driver import NONE_DRIVER, Driver
from formula10.domain.model.race_guess import RaceGuess from formula10.domain.model.race_guess import RaceGuess
@ -11,14 +12,10 @@ from formula10.domain.model.season_guess_result import SeasonGuessResult
from formula10.domain.model.team import Team from formula10.domain.model.team import Team
from formula10.domain.model.user import User from formula10.domain.model.user import User
RACE_GUESS_OFFSET_POINTS: Dict[int, int] = { # Guess points
3: 1,
2: 3,
1: 6,
0: 10
}
RACE_GUESS_DNF_POINTS: int = 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_HOT_TAKE_POINTS: int = 10
SEASON_GUESS_P2_POINTS: int = 10 SEASON_GUESS_P2_POINTS: int = 10
SEASON_GUESS_OVERTAKES_POINTS: int = 10 SEASON_GUESS_OVERTAKES_POINTS: int = 10
@ -30,6 +27,8 @@ SEASON_GUESS_TEAMWINNER_FALSE_POINTS: int = -3
SEASON_GUESS_PODIUMS_CORRECT_POINTS: int = 3 SEASON_GUESS_PODIUMS_CORRECT_POINTS: int = 3
SEASON_GUESS_PODIUMS_FALSE_POINTS: int = -2 SEASON_GUESS_PODIUMS_FALSE_POINTS: int = -2
# Driver points
DRIVER_RACE_POINTS: Dict[int, int] = { DRIVER_RACE_POINTS: Dict[int, int] = {
1: 25, 1: 25,
2: 18, 2: 18,
@ -40,8 +39,12 @@ DRIVER_RACE_POINTS: Dict[int, int] = {
7: 6, 7: 6,
8: 4, 8: 4,
9: 2, 9: 2,
10: 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
WDC_STANDING_2023: Dict[str, int] = { WDC_STANDING_2023: Dict[str, int] = {
"Max Verstappen": 1, "Max Verstappen": 1,
@ -63,9 +66,8 @@ WDC_STANDING_2023: Dict[str, int] = {
"Daniel Ricciardo": 17, "Daniel Ricciardo": 17,
"Zhou Guanyu": 18, "Zhou Guanyu": 18,
"Kevin Magnussen": 19, "Kevin Magnussen": 19,
"Logan Sargeant": 21 "Logan Sargeant": 21,
} }
WCC_STANDING_2023: Dict[str, int] = { WCC_STANDING_2023: Dict[str, int] = {
"Red Bull": 1, "Red Bull": 1,
"Mercedes": 2, "Mercedes": 2,
@ -76,11 +78,14 @@ WCC_STANDING_2023: Dict[str, int] = {
"Williams": 7, "Williams": 7,
"VCARB": 8, "VCARB": 8,
"Sauber": 9, "Sauber": 9,
"Haas": 10 "Haas": 10,
} }
def standing_points(race_guess: RaceGuess, race_result: RaceResult) -> int: 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: if guessed_driver_position is None:
return 0 return 0
@ -90,6 +95,7 @@ def standing_points(race_guess: RaceGuess, race_result: RaceResult) -> int:
return RACE_GUESS_OFFSET_POINTS[position_offset] return RACE_GUESS_OFFSET_POINTS[position_offset]
def dnf_points(race_guess: RaceGuess, race_result: RaceResult) -> int: def dnf_points(race_guess: RaceGuess, race_result: RaceResult) -> int:
if race_guess.dnf_guess in race_result.initial_dnf: if race_guess.dnf_guess in race_result.initial_dnf:
return RACE_GUESS_DNF_POINTS return RACE_GUESS_DNF_POINTS
@ -99,136 +105,201 @@ def dnf_points(race_guess: RaceGuess, race_result: RaceResult) -> int:
return 0 return 0
class PointsModel(Model): class PointsModel(Model):
""" """
This class bundles all data + functionality required to do points calculations. This class bundles all data + functionality required to do points calculations.
""" """
_points_per_step: Dict[str, List[int]] | None = None
_driver_points_per_step: Dict[str, List[int]] | None = None
_team_points_per_step: Dict[str, List[int]] | None = None
_dnfs: Dict[str, int] | None = None
def __init__(self): def __init__(self):
Model.__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]]: def points_per_step(self) -> Dict[str, List[int]]:
""" """
Returns a dictionary of lists, containing points per race for each user. Returns a dictionary of lists, containing points per race for each user.
""" """
if self._points_per_step is None: points_per_step = dict()
self._points_per_step = dict()
for user in self.all_users(): for user in self.all_users():
self._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(): for race_guess in self.all_race_guesses():
user_name: str = race_guess.user.name user_name: str = race_guess.user.name
race_number: int = race_guess.race.number 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: if race_result is None:
continue continue
self._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 self._points_per_step return points_per_step
# @todo Doesn't include fastest lap + sprint points @cache.memoize(
def driver_points_per_step(self) -> Dict[str, List[int]]: 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. Returns a dictionary of lists, containing points per race for each driver.
""" """
if self._driver_points_per_step is None: driver_points_per_step = dict()
self._driver_points_per_step = dict() for driver in self.all_drivers(
for driver in self.all_drivers(include_none=False): include_none=False, include_inactive=include_inactive
self._driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers ):
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(): for race_result in self.all_race_results():
for position, driver in race_result.standing.items():
driver_name: str = driver.name
race_number: int = race_result.race.number race_number: int = race_result.race.number
self._driver_points_per_step[driver_name][race_number] = DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 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
)
return self._driver_points_per_step for position, driver in race_result.sprint_standing.items():
driver_name: str = driver.name
# @todo Doesn't include fastest lap + sprint points 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]]: def team_points_per_step(self) -> Dict[str, List[int]]:
""" """
Returns a dictionary of lists, containing points per race for each team. Returns a dictionary of lists, containing points per race for each team.
""" """
if self._team_points_per_step is None: team_points_per_step = dict()
self._team_points_per_step = dict()
for team in self.all_teams(include_none=False): for team in self.all_teams(include_none=False):
self._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 race_result in self.all_race_results():
for position, driver in race_result.standing.items(): for driver in race_result.standing.values():
team_name: str = driver.team.name team_name: str = driver.team.name
race_number: int = race_result.race.number race_number: int = race_result.race.number
self._team_points_per_step[team_name][race_number] += DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 team_points_per_step[team_name][
race_number
] += self.driver_points_per_step(include_inactive=True)[driver.name][
race_number
]
return self._team_points_per_step return team_points_per_step
# @todo Doesn't include sprint dnfs @cache.cached(timeout=None, key_prefix="points_dnfs")
def dnfs(self) -> Dict[str, int]: def dnfs(self) -> Dict[str, int]:
if self._dnfs is None: dnfs = dict()
self._dnfs = dict()
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=True):
self._dnfs[driver.name] = 0 dnfs[driver.name] = 0
for race_result in self.all_race_results(): for race_result in self.all_race_results():
for driver in race_result.all_dnfs: for driver in race_result.all_dnfs:
self._dnfs[driver.name] += 1 dnfs[driver.name] += 1
return self._dnfs for driver in race_result.sprint_dnfs:
dnfs[driver.name] += 1
return dnfs
# #
# Driver stats # 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]]: def driver_points_per_step_cumulative(self) -> Dict[str, List[int]]:
""" """
Returns a dictionary of lists, containing cumulative points per race for each driver. Returns a dictionary of lists, containing cumulative points per race for each driver.
""" """
points_per_step_cumulative: Dict[str, List[int]] = dict() points_per_step_cumulative: Dict[str, List[int]] = dict()
for driver_name, points in self.driver_points_per_step().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() points_per_step_cumulative[driver_name] = np.cumsum(points).tolist()
return points_per_step_cumulative return points_per_step_cumulative
@overload @overload
def driver_points_by(self, *, driver_name: str) -> 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. Returns a list of points per race for a specific driver.
""" """
return self.driver_points_by(driver_name=driver_name) return self.driver_points_by(
driver_name=driver_name, include_inactive=include_inactive
)
@overload @overload
def driver_points_by(self, *, race_name: str) -> 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. Returns a dictionary of points per driver for a specific race.
""" """
return self.driver_points_by(race_name=race_name) return self.driver_points_by(
race_name=race_name, include_inactive=include_inactive
)
@overload @overload
def driver_points_by(self, *, driver_name: str, race_name: str) -> 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. Returns the points for a specific race for a specific driver.
""" """
return self.driver_points_by(driver_name=driver_name, race_name=race_name) return self.driver_points_by(
driver_name=driver_name,
race_name=race_name,
include_inactive=include_inactive,
)
def driver_points_by(self, *, driver_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
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: if driver_name is not None and race_name is None:
return self.driver_points_per_step()[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: if driver_name is None and race_name is not None:
race_number: int = self.race_by(race_name=race_name).number race_number: int = self.race_by(race_name=race_name).number
points_by_race: Dict[str, int] = dict() points_by_race: Dict[str, int] = dict()
for _driver_name, points in self.driver_points_per_step().items(): for _driver_name, points in self.driver_points_per_step(
include_inactive=include_inactive
).items():
points_by_race[_driver_name] = points[race_number] points_by_race[_driver_name] = points[race_number]
return points_by_race return points_by_race
@ -236,46 +307,79 @@ class PointsModel(Model):
if driver_name is not None and race_name is not None: if driver_name is not None and race_name is not None:
race_number: int = self.race_by(race_name=race_name).number race_number: int = self.race_by(race_name=race_name).number
return self.driver_points_per_step()[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") 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: def total_driver_points_by(self, driver_name: str) -> int:
return sum(self.driver_points_by(driver_name=driver_name)) return sum(
self.driver_points_by(driver_name=driver_name, include_inactive=True)
)
def drivers_sorted_by_points(self) -> List[Driver]: @cache.memoize(
comparator: Callable[[Driver], int] = lambda driver: self.total_driver_points_by(driver.name) timeout=None, args_to_ignore=["self"]
return sorted(self.all_drivers(include_none=False), key=comparator, reverse=True) ) # 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]]: def wdc_standing_by_position(self) -> Dict[int, List[str]]:
standing: Dict[int, List[str]] = dict() standing: Dict[int, List[str]] = dict()
for position in range(1, len(self.all_drivers(include_none=False)) + 1): for position in range(
1, len(self.all_drivers(include_none=False, include_inactive=True)) + 1
):
standing[position] = list() standing[position] = list()
position: int = 1 position: int = 1
last_points: int = 0 last_points: int = 0
for driver in self.drivers_sorted_by_points(): for driver in self.drivers_sorted_by_points(include_inactive=True):
points: int = self.total_driver_points_by(driver.name) points: int = self.total_driver_points_by(driver.name)
if points < last_points: 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) standing[position].append(driver.name)
last_points = points last_points = points
return standing 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]: def wdc_standing_by_driver(self) -> Dict[str, int]:
standing: Dict[str, int] = dict() standing: Dict[str, int] = dict()
position: int = 1 position: int = 1
last_points: int = 0 last_points: int = 0
for driver in self.drivers_sorted_by_points(): for driver in self.drivers_sorted_by_points(include_inactive=True):
points: int = self.total_driver_points_by(driver.name) points: int = self.total_driver_points_by(driver.name)
if points < last_points: 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 standing[driver.name] = position
last_points = points last_points = points
@ -283,8 +387,16 @@ class PointsModel(Model):
return standing return standing
def wdc_diff_2023_by(self, driver_name: str) -> int: def wdc_diff_2023_by(self, driver_name: str) -> int:
return WDC_STANDING_2023[driver_name] - self.wdc_standing_by_driver()[driver_name] 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]: def most_dnf_names(self) -> List[str]:
dnf_names: List[str] = list() dnf_names: List[str] = list()
most_dnfs: int = 0 most_dnfs: int = 0
@ -299,17 +411,20 @@ class PointsModel(Model):
return dnf_names 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]: def most_gained_names(self) -> List[str]:
most_gained_names: List[str] = list() most_gained_names: List[str] = list()
most_gained: int = 0 most_gained: int = 0
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=True):
gained: int = self.wdc_diff_2023_by(driver.name) gained: int = self.wdc_diff_2023_by(driver.name)
if gained > most_gained: if gained > most_gained:
most_gained = gained most_gained = gained
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=True):
gained: int = self.wdc_diff_2023_by(driver.name) gained: int = self.wdc_diff_2023_by(driver.name)
if gained == most_gained: if gained == most_gained:
@ -317,17 +432,20 @@ class PointsModel(Model):
return most_gained_names 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]: def most_lost_names(self) -> List[str]:
most_lost_names: List[str] = list() most_lost_names: List[str] = list()
most_lost: int = 100 most_lost: int = 100
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=True):
lost: int = self.wdc_diff_2023_by(driver.name) lost: int = self.wdc_diff_2023_by(driver.name)
if lost < most_lost: if lost < most_lost:
most_lost = lost most_lost = lost
for driver in self.all_drivers(include_none=False): for driver in self.all_drivers(include_none=False, include_inactive=True):
lost: int = self.wdc_diff_2023_by(driver.name) lost: int = self.wdc_diff_2023_by(driver.name)
if lost == most_lost: if lost == most_lost:
@ -339,6 +457,9 @@ class PointsModel(Model):
# Team points # 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]]: def team_points_per_step_cumulative(self) -> Dict[str, List[int]]:
""" """
Returns a dictionary of lists, containing cumulative points per race for each team. Returns a dictionary of lists, containing cumulative points per race for each team.
@ -349,18 +470,34 @@ class PointsModel(Model):
return points_per_step_cumulative 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: def total_team_points_by(self, team_name: str) -> int:
teammates: List[Driver] = self.drivers_by(team_name=team_name) teammates: List[Driver] = self.drivers_by(
return sum(self.driver_points_by(driver_name=teammates[0].name)) + sum(self.driver_points_by(driver_name=teammates[1].name)) 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]: 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) 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]]: def wcc_standing_by_position(self) -> Dict[int, List[str]]:
standing: Dict[int, List[str]] = dict() 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() standing[position] = list()
position: int = 1 position: int = 1
@ -369,13 +506,18 @@ class PointsModel(Model):
for team in self.teams_sorted_by_points(): for team in self.teams_sorted_by_points():
points: int = self.total_team_points_by(team.name) points: int = self.total_team_points_by(team.name)
if points < last_points: 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) standing[position].append(team.name)
last_points = points last_points = points
return standing 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]: def wcc_standing_by_team(self) -> Dict[str, int]:
standing: Dict[str, int] = dict() standing: Dict[str, int] = dict()
@ -385,7 +527,14 @@ class PointsModel(Model):
for team in self.teams_sorted_by_points(): for team in self.teams_sorted_by_points():
points: int = self.total_team_points_by(team.name) points: int = self.total_team_points_by(team.name)
if points < last_points: 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 standing[team.name] = position
last_points = points last_points = points
@ -430,7 +579,12 @@ class PointsModel(Model):
""" """
return self.points_by(user_name=user_name, race_name=race_name) return self.points_by(user_name=user_name, race_name=race_name)
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: if user_name is not None and race_name is None:
return self.points_per_step()[user_name] return self.points_per_step()[user_name]
@ -463,14 +617,25 @@ class PointsModel(Model):
comparator: Callable[[User], int] = lambda user: self.total_points_by(user.name) comparator: Callable[[User], int] = lambda user: self.total_points_by(user.name)
return sorted(self.all_users(), key=comparator, reverse=True) 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]: def user_standing(self) -> Dict[str, int]:
standing: Dict[str, int] = dict() standing: Dict[str, int] = dict()
position: int = 1 position: int = 1
last_points: int = 0 last_points: int = 0
for user in self.users_sorted_by_points(): for user in self.users_sorted_by_points():
if self.total_points_by(user.name) < last_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 standing[user.name] = position
@ -482,11 +647,16 @@ class PointsModel(Model):
# Treat standing + dnf picks separately # Treat standing + dnf picks separately
return len(self.race_guesses_by(user_name=user_name)) * 2 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: def picks_with_points_count(self, user_name: str) -> int:
count: int = 0 count: int = 0
for race_guess in self.race_guesses_by(user_name=user_name): 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: if race_result is None:
continue continue
@ -508,9 +678,15 @@ class PointsModel(Model):
# #
def hot_take_correct(self, user_name: str) -> bool: 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: def p2_constructor_correct(self, user_name: str) -> bool:
season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name)
@ -521,9 +697,15 @@ class PointsModel(Model):
return season_guess.p2_wcc.name in self.wcc_standing_by_position()[2] return season_guess.p2_wcc.name in self.wcc_standing_by_position()[2]
def overtakes_correct(self, user_name: str) -> bool: 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: def dnfs_correct(self, user_name: str) -> bool:
season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name) season_guess: SeasonGuess | None = self.season_guesses_by(user_name=user_name)
@ -549,14 +731,23 @@ class PointsModel(Model):
return season_guess.most_wdc_lost.name in self.most_lost_names() 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: def is_team_winner(self, driver: Driver) -> bool:
teammates: List[Driver] = self.drivers_by(team_name=driver.team.name) 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] teammate: Driver = teammates[0] if teammates[1] == driver else teammates[1]
print(f"{driver.name} standing: {self.wdc_standing_by_driver()[driver.name]}, {teammate.name} standing: {self.wdc_standing_by_driver()[teammate.name]}") return (
self.wdc_standing_by_driver()[driver.name]
return self.wdc_standing_by_driver()[driver.name] <= self.wdc_standing_by_driver()[teammate.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: def has_podium(self, driver: Driver) -> bool:
for race_result in self.all_race_results(): for race_result in self.all_race_results():
position: int | None = race_result.driver_standing_position(driver) position: int | None = race_result.driver_standing_position(driver)
@ -580,7 +771,7 @@ class PointsModel(Model):
{ {
"data": self.points_per_step_cumulative()[user.name], "data": self.points_per_step_cumulative()[user.name],
"label": user.name, "label": user.name,
"fill": False "fill": False,
} }
for user in self.all_users() for user in self.all_users()
] ]
@ -598,9 +789,9 @@ class PointsModel(Model):
{ {
"data": self.driver_points_per_step_cumulative()[driver.name], "data": self.driver_points_per_step_cumulative()[driver.name],
"label": driver.abbr, "label": driver.abbr,
"fill": False "fill": False,
} }
for driver in self.all_drivers(include_none=False) for driver in self.all_drivers(include_none=False, include_inactive=True)
] ]
return json.dumps(data) return json.dumps(data)
@ -616,7 +807,7 @@ class PointsModel(Model):
{ {
"data": self.team_points_per_step_cumulative()[team.name], "data": self.team_points_per_step_cumulative()[team.name],
"label": team.name, "label": team.name,
"fill": False "fill": False,
} }
for team in self.all_teams(include_none=False) for team in self.all_teams(include_none=False)
] ]

View File

@ -1,5 +1,5 @@
from typing import List, Callable from typing import List, Callable
from formula10 import ENABLE_TIMING from formula10 import ENABLE_TIMING, cache
from formula10.domain.domain_model import Model from formula10.domain.domain_model import Model
from formula10.domain.model.driver import Driver from formula10.domain.model.driver import Driver
@ -29,10 +29,12 @@ class TemplateModel(Model):
if active_result_race_name is not None: if active_result_race_name is not None:
self.active_result = self.race_result_by(race_name=active_result_race_name) self.active_result = self.race_result_by(race_name=active_result_race_name)
def race_guess_open(self, race: Race) -> bool: @staticmethod
def race_guess_open(race: Race) -> bool:
return not race_has_started(race=race) if ENABLE_TIMING else True return not race_has_started(race=race) if ENABLE_TIMING else True
def season_guess_open(self) -> bool: @staticmethod
def season_guess_open() -> bool:
return not race_has_started(race_id=1) if ENABLE_TIMING else True return not race_has_started(race_id=1) if ENABLE_TIMING else True
def race_result_open(self, race_name: str) -> bool: def race_result_open(self, race_name: str) -> bool:
@ -52,6 +54,7 @@ class TemplateModel(Model):
return self.all_users() return self.all_users()
@cache.cached(timeout=None, key_prefix="template_first_race_without_result") # Cleanup when adding/updating race results
def first_race_without_result(self) -> Race | None: def first_race_without_result(self) -> Race | None:
""" """
Returns the first race-object with no associated race result. Returns the first race-object with no associated race result.
@ -69,6 +72,14 @@ class TemplateModel(Model):
def current_race(self) -> Race | None: def current_race(self) -> Race | None:
return self.first_race_without_result() return self.first_race_without_result()
def active_result_race_or_current_race(self) -> Race:
if self.active_result is not None:
return self.active_result.race
elif self.current_race is not None:
return self.current_race
else:
return self.all_races()[0]
def active_result_race_name_or_current_race_name(self) -> str: def active_result_race_name_or_current_race_name(self) -> str:
if self.active_result is not None: if self.active_result is not None:
return self.active_result.race.name return self.active_result.race.name
@ -85,9 +96,12 @@ class TemplateModel(Model):
else: else:
return self.all_races()[0].name_sanitized return self.all_races()[0].name_sanitized
def all_drivers_or_active_result_standing_drivers(self) -> List[Driver]: def all_active_drivers_or_active_result_standing_drivers(self) -> List[Driver]:
return self.active_result.ordered_standing_list() if self.active_result is not None else self.all_drivers(include_none=False) return self.active_result.ordered_standing_list() if self.active_result is not None else self.all_drivers(include_none=False, include_inactive=False)
def drivers_for_wdc_gained(self) -> List[Driver]: def all_active_drivers_or_active_result_sprint_standing_drivers(self) -> List[Driver]:
return self.active_result.ordered_sprint_standing_list() if self.active_result is not None else self.all_drivers(include_none=False, include_inactive=False)
def active_drivers_for_wdc_gained(self) -> List[Driver]:
predicate: Callable[[Driver], bool] = lambda driver: driver.abbr not in self._wdc_gained_excluded_abbrs predicate: Callable[[Driver], bool] = lambda driver: driver.abbr not in self._wdc_gained_excluded_abbrs
return find_multiple_strict(predicate, self.all_drivers(include_none=False)) return find_multiple_strict(predicate, self.all_drivers(include_none=False, include_inactive=False))

View File

View File

@ -0,0 +1,364 @@
# https://docs.fastf1.dev/events.html#event-formats
from typing import Optional
from fastf1.core import Lap, DriverResult, Session
from fastf1.events import Event
from pandas import Timestamp, Timedelta
FASTF1_SESSIONTYPE_NORMAL: str = "conventional"
FASTF1_SESSIONTYPE_SPRINT: str = "sprint_qualifying"
FASTF1_SPRINT_QUALIFYING_SESSION: int = 2
FASTF1_SPRINT_SESSION: int = 3
FASTF1_QUALIFYING_SESSION: int = 4
FASTF1_RACE_SESSION: int = 5
class EventDataHelper:
"""
Helper class that provides easy access to EventDataFrame columns.
"""
def __init__(self, event: Event):
self.__event = event
__event: Event
@property
def round_number(self) -> int:
return self.__event["RoundNumber"]
@property
def country(self) -> str:
return self.__event["Country"]
@property
def location(self) -> str:
return self.__event["Location"]
@property
def official_event_name(self) -> str:
"""
Get the event name including sponsor names etc.
"""
return self.__event["OfficialEventName"]
@property
def event_name(self) -> str:
return self.__event["EventName"]
@property
def event_date(self) -> Timestamp:
return self.__event["EventDate"]
@property
def event_format(self) -> str:
"""
Get the event format.
@return: Either conventional, sprint, sprint_shootout, sprint_qualifying or testing (sprint_qualifying is the 2024 sprint format).
"""
return self.__event["EventFormat"]
def session(self, session_number: int) -> str:
"""
Get the format of a session belonging to this event.
@param session_number: Either 1, 2, 3, 4 or 5
@return: Either Pactice 1, Practice 2, Practice 3, Qualifying, Sprint, Sprint Shootout or Race
"""
return self.__event[f"Session{session_number}"]
def session_date(self, session_number: int) -> Timestamp:
"""
Get the date of a session belonging to this event.
@param session_number: Either 1, 2, 3, 4 or 5
"""
return self.__event[f"Session{session_number}Date"]
def session_date_utc(self, session_number: int) -> Timestamp:
"""
Get the date a session belonging to this event in coordinated universal time.
@param session_number: Either 1, 2, 3, 4 or 5
"""
return self.__event[f"Session{session_number}DateUtc"]
@property
def f1_api_support(self) -> bool:
return self.__event["F1ApiSupport"]
# The following doesn't correspond to DataFrame columns
@property
def sprint_qualifying(self):
return self.__event.get_sprint_qualifying()
@property
def sprint(self):
return self.__event.get_sprint()
@property
def qualifying(self) -> Session:
return self.__event.get_qualifying()
@property
def race(self) -> Session:
return self.__event.get_race()
class LapDataHelper:
"""
Helper class that provides easy access to Lap DataFrame columns.
"""
def __init__(self, lap: Lap):
self.__lap = lap
__lap: Lap
@property
def time(self) -> Timedelta:
return self.__lap["Time"]
@property
def driver(self) -> str:
return self.__lap["Driver"]
@property
def driver_number(self) -> str:
return self.__lap["DriverNumber"]
@property
def lap_time(self) -> Timedelta:
return self.__lap["LapTime"]
@property
def lap_number(self) -> int:
return self.__lap["LapNumber"]
@property
def stint(self) -> int:
return self.__lap["Stint"]
@property
def pit_out_time(self) -> Timedelta:
"""
Get the session time when the car left the pit.
"""
return self.__lap["PitOutTime"]
@property
def pit_in_time(self) -> Timedelta:
"""
Get the session time when the car entered the pit.
"""
return self.__lap["PitInTime"]
def sector_time(self, sector_number: int) -> Timedelta:
"""
Get the sector times set in this lap.
@param sector_number: Either 1, 2 or 3.
"""
return self.__lap[f"Sector{sector_number}Time"]
def sector_session_time(self, sector_number: int) -> Timedelta:
"""
Get the session time when the sector time was set.
@param sector_number: Either 1, 2 or 3.
"""
return self.__lap[f"Sector{sector_number}SessionTime"]
def speed(self, speedtrap: str) -> float:
"""
Get car speed at measure point in this lap.
@param speedtrap: Either I1, I2, FL or ST (sector 1, sector 2, finish line or longest straight)
"""
return self.__lap[f"Speed{speedtrap}"]
@property
def is_personal_best(self) -> bool:
return self.__lap["IsPersonalBest"]
@property
def compound(self) -> str:
"""
Get compound used in this lap.
@return: Either SOFT, MEDIUM, HARD, INTERMEDIATE or WET
"""
return self.__lap["Compound"]
@property
def tyre_life(self) -> int:
"""
Get laps driven on the current tyre (includes laps from other sessions).
"""
return self.__lap["TyreLife"]
@property
def fresh_tyre(self) -> bool:
return self.__lap["FreshTyre"]
@property
def team(self) -> str:
return self.__lap["Team"]
@property
def lap_start_time(self) -> Timedelta:
return self.__lap["LapStartTime"]
@property
def lap_start_date(self) -> Timestamp:
return self.__lap["LapStartDate"]
@property
def track_status(self) -> str:
"""
Get the track status in this lap.
@return: Either 1, 2, 3, 4, 5, 6 or 7 (clear, yellow flag, ?, SC, red flag, VSC and VSC ending).
"""
return self.__lap["TrackStatus"]
@property
def position(self) -> int:
return self.__lap["Position"]
@property
def deleted(self) -> Optional[bool]:
"""
Determine if the lap was deleted (only available if race control messages are loaded).
"""
return self.__lap["Deleted"]
@property
def deleted_reason(self) -> str:
return self.__lap["DeletedReason"]
@property
def fast_f1_generated(self) -> bool:
"""
Determine if the lap was generated by FastF1 (information is interpolated).
"""
return self.__lap["FastF1Generated"]
@property
def is_accurate(self) -> bool:
"""
Determine if lap start and end match with other laps before and after.
"""
return self.__lap["IsAccurate"]
class DriverResultDataHelper:
"""
Helper class that provides easy access to DriverResult DataFrame columns.
"""
def __init__(self, driver_result: DriverResult):
self.__driver_result = driver_result
__driver_result: DriverResult
@property
def driver_number(self) -> str:
return self.__driver_result["DriverNumber"]
@property
def broadcast_name(self) -> str:
"""
Get this driver's broadcast name.
@return: For example P GASLY
"""
return self.__driver_result["BroadcastName"]
@property
def full_name(self) -> str:
return self.__driver_result["FullName"]
@property
def abbreviation(self) -> str:
return self.__driver_result["Abbreviation"]
# @property
# def driver_id(self) -> str:
# """
# Get the driverId used by the Ergast API.
# """
# return self.__driver_result["DriverID"]
@property
def team_name(self) -> str:
return self.__driver_result["TeamName"]
@property
def team_color(self) -> str:
return self.__driver_result["TeamColor"]
# @property
# def team_id(self) -> str:
# """
# Get the constructorId used by the Ergast API.
# @return:
# """
# return self.__driver_result["TeamID"]
@property
def first_name(self) -> str:
return self.__driver_result["FirstName"]
@property
def last_name(self) -> str:
return self.__driver_result["LastName"]
@property
def headshot_url(self) -> str:
return self.__driver_result["HeadshotUrl"]
@property
def country_code(self) -> str:
"""
Get a driver's three-letter country code.
@return: For example FRA
"""
return self.__driver_result["CountryCode"]
@property
def position(self) -> int:
return self.__driver_result["Position"]
@property
def classified_position(self) -> str:
"""
Get the classification result for this driver.
@return: Either R, D, E, W, F, N (retired, disqualified, excluded, withdrawn, failed to qualify and not classified)
"""
return self.__driver_result["ClassifiedPosition"]
@property
def grid_position(self) -> int:
"""
Get the driver's starting position.
"""
return self.__driver_result["GridPosition"]
def qualifying_time(self, qualifying_number: int) -> Timedelta:
"""
Get the driver's best qualifying time.
@param qualifying_number: Either 1, 2 or 3 (for Q1, Q2 and Q3)
"""
return self.__driver_result[f"Q{qualifying_number}"]
@property
def time(self) -> Timedelta:
"""
Get the driver's total race time.
"""
return self.__driver_result["Time"]
@property
def status(self) -> str:
"""
Get the driver's finishing status.
@return: For example Finished, +1 Lap, Crash, Gearbox...
"""
return self.__driver_result["Status"]
@property
def points(self) -> int:
return self.__driver_result["Points"]

View File

View File

View File

@ -0,0 +1,55 @@
from typing import Any, Callable, Dict
class OpenF1Driver:
__type_conversion_map__: Dict[str, Callable[[Any], Any]] = {
"session_key": int,
"meeting_key": int,
"full_name": str,
"first_name": str,
"last_name": str,
"name_acronym": str,
"broadcast_name": str,
"country_code": str,
"headshot_url": str,
"driver_number": int,
"team_colour": str,
"team_name": str
}
def __init__(self, response: dict[str, str] | None):
if response is None:
return
for key in response:
if not hasattr(self, key):
raise Exception(f"Mismatch between response data and {type(self).__name__} (key={key})")
if not key in self.__type_conversion_map__:
raise Exception(f"Mismatch between response data and {type(self).__name__}.__type_map__ (key={key})")
setattr(self, key, self.__type_conversion_map__[key](response[key]))
print("OpenF1Driver:", self.__dict__)
def to_params(self) -> Dict[str, str]:
params: Dict[str, str] = dict()
for key in self.__dict__:
params[str(key)] = str(self.__dict__[key])
return params
# Set all members to None so hasattr works above
session_key: int = None # type: ignore
meeting_key: int = None # type: ignore
full_name: str = None # type: ignore
first_name: str = None # type: ignore
last_name: str = None # type: ignore
name_acronym: str = None # type: ignore
broadcast_name: str = None # type: ignore
country_code: str = None # type: ignore
headshot_url: str = None # type: ignore
driver_number: int = None # type: ignore
team_colour: str = None # type: ignore
team_name: str = None # type: ignore

View File

@ -0,0 +1,40 @@
from datetime import datetime
from typing import Any, Callable, Dict
class OpenF1Position:
__type_conversion_map__: Dict[str, Callable[[Any], Any]] = {
"session_key": int,
"meeting_key": int,
"driver_number": int,
"date": lambda date: datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%f"),
"position": int
}
def __init__(self, response: dict[str, str] | None):
if response is None:
return
for key in response:
if not hasattr(self, key):
raise Exception(f"Mismatch between response data and {type(self).__name__} (key={key})")
if not key in self.__type_conversion_map__:
raise Exception(f"Mismatch between response data and {type(self).__name__}.__type_map__ (key={key})")
setattr(self, key, self.__type_conversion_map__[key](response[key]))
print("OpenF1Position:", self.__dict__)
def to_params(self) -> Dict[str, str]:
params: Dict[str, str] = dict()
for key in self.__dict__:
params[str(key)] = str(self.__dict__[key])
return params
session_key: int = None # type: ignore
meeting_key: int = None # type: ignore
driver_number: int = None # type: ignore
date: datetime = None # type: ignore
position: int = None # type: ignore

View File

@ -0,0 +1,60 @@
from datetime import datetime, time
from typing import Any, Callable, Dict
class OpenF1Session:
__type_conversion_map__: Dict[str, Callable[[Any], Any]] = {
"location": str,
"country_key": int,
"country_code": str,
"country_name": str,
"circuit_key": int,
"circuit_short_name": str,
"session_type": str,
"session_name": str,
"date_start": lambda date: datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"),
"date_end": lambda date: datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"),
"gmt_offset": lambda time: datetime.strptime(time, "%H:%M:%S").time(),
"session_key": int,
"meeting_key": int,
"year": int
}
def __init__(self, response: dict[str, str] | None):
if response is None:
return
for key in response:
if not hasattr(self, key):
raise Exception(f"Mismatch between response data and {type(self).__name__} (key={key})")
if not key in self.__type_conversion_map__:
raise Exception(f"Mismatch between response data and {type(self).__name__}.__type_map__ (key={key})")
setattr(self, key, self.__type_conversion_map__[key](response[key]))
print("OpenF1Session:", self.__dict__)
def to_params(self) -> Dict[str, str]:
params: Dict[str, str] = dict()
for key in self.__dict__:
params[str(key)] = str(self.__dict__[key])
return params
location: str = None # type: ignore
country_key: int = None # type: ignore
country_code: str = None # type: ignore
country_name: str = None # type: ignore
circuit_key: int = None # type: ignore
circuit_short_name: str = None # type: ignore
session_type: str = None # type: ignore
session_name: str = None # type: ignore
date_start: datetime = None # type: ignore
date_end: datetime = None # type: ignore
gmt_offset: time = None # type: ignore
session_key: int = None # type: ignore
meeting_key: int = None # type: ignore
year: int = None # type: ignore

View File

@ -0,0 +1,9 @@
OPENF1_URL: str = "https://api.openf1.org/v1"
OPENF1_SESSION_ENDPOINT: str = f"{OPENF1_URL}/sessions"
OPENF1_POSITION_ENDPOINT: str = f"{OPENF1_URL}/position"
OPENF1_DRIVER_ENDPOINT: str = f"{OPENF1_URL}/drivers"
OPENF1_SESSION_TYPE_RACE: str = "Race"
OPENF1_SESSION_NAME_RACE: str = "Race"
OPENF1_SESSION_NAME_SPRINT: str = "Sprint"

View File

@ -0,0 +1,76 @@
from datetime import datetime
import json
from typing import Any, Callable, Dict, List, cast
from requests import Response, get
from formula10.openf1.model.openf1_driver import OpenF1Driver
from formula10.openf1.model.openf1_position import OpenF1Position
from formula10.openf1.model.openf1_session import OpenF1Session
from formula10.openf1.openf1_definitions import OPENF1_DRIVER_ENDPOINT, OPENF1_POSITION_ENDPOINT, OPENF1_SESSION_ENDPOINT, OPENF1_SESSION_NAME_RACE, OPENF1_SESSION_NAME_SPRINT, OPENF1_SESSION_TYPE_RACE
def openf1_request_helper(endpoint: str, params: Dict[str, str]) -> List[Dict[str, str]]:
response: Response = get(endpoint, params=params)
if not response.ok:
raise Exception(f"OpenF1 request to {response.request.url} failed")
obj: Any = json.loads(response.text)
if isinstance(obj, List):
return cast(List[Dict[str, str]], obj)
elif isinstance(obj, Dict):
return [cast(Dict[str, str], obj)]
else:
# @todo Fail gracefully
raise Exception(f"Unexpected OpenF1 response from {response.request.url}: {obj}")
def openf1_fetch_latest_session(session_name: str) -> OpenF1Session:
# ApiSession object only supports integer session_keys
response: List[Dict[str, str]] = openf1_request_helper(OPENF1_SESSION_ENDPOINT, {
"session_key": "latest",
"session_type": OPENF1_SESSION_TYPE_RACE,
"session_name": session_name
})
return OpenF1Session(response[0])
def openf1_fetch_latest_race_session_key() -> int:
return openf1_fetch_latest_session(OPENF1_SESSION_NAME_RACE).session_key
def openf1_fetch_latest_sprint_session_key() -> int:
return openf1_fetch_latest_session(OPENF1_SESSION_NAME_SPRINT).session_key
def openf1_fetch_session(session_name: str, country_code: str) -> OpenF1Session:
_session: OpenF1Session = OpenF1Session(None)
_session.session_type = OPENF1_SESSION_TYPE_RACE # includes races + sprints
_session.year = 2024
_session.country_code = country_code
_session.session_name = session_name
response: List[Dict[str, str]] = openf1_request_helper(OPENF1_SESSION_ENDPOINT, _session.to_params())
return OpenF1Session(response[0])
def openf1_fetch_driver(session_key: int, name_acronym: str) -> OpenF1Driver:
_driver: OpenF1Driver = OpenF1Driver(None)
_driver.name_acronym = name_acronym
_driver.session_key = session_key
response: List[Dict[str, str]] = openf1_request_helper(OPENF1_DRIVER_ENDPOINT, _driver.to_params())
return OpenF1Driver(response[0])
def openf1_fetch_position(session_key: int, position: int):
_position: OpenF1Position = OpenF1Position(None)
_position.session_key = session_key
_position.position = position
response: List[Dict[str, str]] = openf1_request_helper(OPENF1_POSITION_ENDPOINT, _position.to_params())
# Find the last driver that was on this position at last
predicate: Callable[[Dict[str, str]], datetime] = lambda position: datetime.strptime(position["date"], "%Y-%m-%dT%H:%M:%S.%f")
return OpenF1Position(max(response, key=predicate))

View File

@ -30,7 +30,7 @@
<option value="" selected disabled hidden></option> <option value="" selected disabled hidden></option>
{% if drivers == none %} {% if drivers == none %}
{% set drivers = model.all_drivers(include_none=include_none) %} {% set drivers = model.all_drivers(include_none=include_none, include_inactive=False) %}
{% endif %} {% endif %}
{% for driver in drivers %} {% for driver in drivers %}
@ -49,7 +49,7 @@
{% set user_has_chosen = namespace(driverpre=false) %} {% set user_has_chosen = namespace(driverpre=false) %}
{% if drivers == none %} {% if drivers == none %}
{% set drivers = model.all_drivers(include_none=include_none) %} {% set drivers = model.all_drivers(include_none=include_none, include_inactive=False) %}
{% endif %} {% endif %}
{% for driver in drivers %} {% for driver in drivers %}

View File

@ -42,44 +42,73 @@
{% block body %} {% block body %}
<div class="grid card-grid"> {% set race_result_open=model.race_result_open(model.active_result_race_name_or_current_race_name()) %}
{% if race_result_open == true %}
{% set action_save_href = "/result-enter/" ~ model.active_result_race_name_or_current_race_name_sanitized() %}
{% set action_fetch_href = "/result-fetch/" ~ model.active_result_race_name_or_current_race_name_sanitized() %}
{% else %}
{% set action_save_href = "" %}
{% set action_fetch_href = "" %}
{% endif %}
<div class="card shadow-sm mb-2" style="max-width: 450px;"> <form action="{{ action_fetch_href }}" method="post">
<div class="card shadow-sm mb-2 w-100">
<div class="card-header">
Autofill
</div>
<div class="card-body">
<input type="submit" class="btn btn-danger mt-2 w-100" value="Fetch using FastF1"
{% if race_result_open == false %}disabled="disabled"{% endif %}>
</div>
</div>
</form>
<form class="grid card-grid" action="{{ action_save_href }}" method="post">
{# Race result #}
<div class="card shadow-sm mb-2 w-100">
<div class="card-header"> <div class="card-header">
{{ model.active_result_race_name_or_current_race_name() }} {{ model.active_result_race_name_or_current_race_name() }}
</div> </div>
<div class="card-body"> <div class="card-body">
<div class="d-inline-block overflow-x-scroll w-100"> <div class="d-inline-block overflow-x-scroll w-100">
<div style="width: 410px;"> <div style="width: 460px;">
{% set race_result_open=model.race_result_open(model.active_result_race_name_or_current_race_name()) %}
{% if race_result_open == true %}
{% set action_save_href = "/result-enter/" ~ model.active_result_race_name_or_current_race_name_sanitized() %}
{% else %}
{% set action_save_href = "" %}
{% endif %}
<form action="{{ action_save_href }}" method="post">
{# Place numbers #} {# Place numbers #}
<ul class="list-group list-group-flush d-inline-block"> <ul class="list-group list-group-flush d-inline-block">
{% for driver in model.all_drivers_or_active_result_standing_drivers() %} {% for driver in model.all_active_drivers_or_active_result_standing_drivers() %}
<li class="list-group-item p-1"><span id="place_number" <li class="list-group-item p-1"><span id="place_number"
class="fw-bold">P{{ "%02d" % loop.index }}</span>: class="fw-bold">P{{ "%02d" % loop.index }}</span>:
</li> </li>
{% endfor %} {% endfor %}
</ul> </ul>
{# Drag and drop #} {# Drag and drop, "#columns .column" is the selector for the JS #}
<ul id="columns" class="list-group list-group-flush d-inline-block float-end"> <ul id="columns" class="list-group list-group-flush d-inline-block float-end">
{% for driver in model.all_drivers_or_active_result_standing_drivers() %} {% for driver in model.all_active_drivers_or_active_result_standing_drivers() %}
<li class="list-group-item {% if race_result_open == true %}column{% endif %} p-1" <li class="list-group-item {% if race_result_open == true %}column{% endif %} p-1"
{% if race_result_open == true %}draggable="true"{% endif %}> {% if race_result_open == true %}draggable="true"{% endif %}>
{{ driver.name }} {{ driver.name }}
<div class="d-inline-block float-end" style="margin-left: 30px;"> <div class="d-inline-block float-end" style="margin-left: 30px;">
{# Driver DNFed at first #} {# Fastest lap #}
<div class="form-check form-check-reverse d-inline-block"> <div class="form-check form-check-reverse d-inline-block">
<input type="radio" class="form-check-input"
value="{{ driver.id }}"
id="fastest-lap-{{ driver.id }}" name="fastest-lap"
{% if (model.active_result is not none) and (driver.id == model.active_result.fastest_lap_driver.id) %}checked{% endif %}
{% if race_result_open == false %}disabled="disabled"{% endif %}>
<label for="fastest-lap-{{ driver.id }}"
class="form-check-label text-muted" data-bs-toggle="tooltip"
title="Fastest lap">Lap</label>
</div>
{# Driver DNFed at first #}
<div class="form-check form-check-reverse d-inline-block"
style="margin-left: 2px;">
<input type="checkbox" class="form-check-input" <input type="checkbox" class="form-check-input"
value="{{ driver.id }}" value="{{ driver.id }}"
id="first-dnf-{{ driver.id }}" name="first-dnf-drivers" id="first-dnf-{{ driver.id }}" name="first-dnf-drivers"
@ -90,7 +119,8 @@
</div> </div>
{# Driver DNFed #} {# Driver DNFed #}
<div class="form-check form-check-reverse d-inline-block mx-2"> <div class="form-check form-check-reverse d-inline-block"
style="margin-left: 2px;">
<input type="checkbox" class="form-check-input" <input type="checkbox" class="form-check-input"
value="{{ driver.id }}" value="{{ driver.id }}"
id="dnf-{{ driver.id }}" name="dnf-drivers" id="dnf-{{ driver.id }}" name="dnf-drivers"
@ -101,7 +131,8 @@
</div> </div>
{# Driver Excluded #} {# Driver Excluded #}
<div class="form-check form-check-reverse d-inline-block"> <div class="form-check form-check-reverse d-inline-block"
style="margin-left: 2px;">
<input type="checkbox" class="form-check-input" <input type="checkbox" class="form-check-input"
value="{{ driver.id }}" value="{{ driver.id }}"
id="exclude-{{ driver.id }}" name="excluded-drivers" id="exclude-{{ driver.id }}" name="excluded-drivers"
@ -121,11 +152,66 @@
<input type="submit" class="btn btn-danger mt-2 w-100" value="Save" <input type="submit" class="btn btn-danger mt-2 w-100" value="Save"
{% if race_result_open == false %}disabled="disabled"{% endif %}> {% if race_result_open == false %}disabled="disabled"{% endif %}>
</form>
</div>
</div> </div>
</div> </div>
</div> </div>
</div> </div>
{# Sprint result #}
{% if model.active_result_race_or_current_race().has_sprint == true %}
<div class="card shadow-sm mb-2 w-100">
<div class="card-header">
Sprint
</div>
<div class="card-body">
<div class="d-inline-block overflow-x-scroll w-100">
<div style="width: 275px;">
{# Place numbers #}
<ul class="list-group list-group-flush d-inline-block">
{% for driver in model.all_active_drivers_or_active_result_sprint_standing_drivers() %}
<li class="list-group-item p-1"><span id="place_number"
class="fw-bold">P{{ "%02d" % loop.index }}</span>:
</li>
{% endfor %}
</ul>
{# Drag and drop, "#columns .column" is the selector for the JS #}
<ul id="columns" class="list-group list-group-flush d-inline-block float-end">
{% for driver in model.all_active_drivers_or_active_result_sprint_standing_drivers() %}
<li class="list-group-item {% if race_result_open == true %}column{% endif %} p-1"
{% if race_result_open == true %}draggable="true"{% endif %}>
{{ driver.name }}
<div class="d-inline-block float-end" style="margin-left: 30px;">
{# Driver DNFed #}
<div class="form-check form-check-reverse d-inline-block"
style="margin-left: 2px;">
<input type="checkbox" class="form-check-input"
value="{{ driver.id }}"
id="sprint-dnf-{{ driver.id }}" name="sprint-dnf-drivers"
{% if (model.active_result is not none) and (driver in model.active_result.sprint_dnfs) %}checked{% endif %}
{% if race_result_open == false %}disabled="disabled"{% endif %}>
<label for="sprint-dnf-{{ driver.id }}"
class="form-check-label text-muted">DNF</label>
</div>
</div>
{# Standing order #}
<input type="hidden" name="sprint-pxx-drivers" value="{{ driver.id }}">
</li>
{% endfor %}
</ul>
</div>
</div>
</div>
</div>
{% endif %}
</form>
{% endblock body %} {% endblock body %}

View File

@ -18,7 +18,6 @@
<div class="card-body"> <div class="card-body">
Picks that match the current standings are marked in green, except for the hot-take and overtake picks, as Picks that match the current standings are marked in green, except for the hot-take and overtake picks, as
those are not evaluated automatically.<br> those are not evaluated automatically.<br>
Points from sprints and fastest laps are not tracked currently.
</div> </div>
</div> </div>
@ -82,7 +81,7 @@
<div class="input-group mt-2" data-bs-toggle="tooltip" <div class="input-group mt-2" data-bs-toggle="tooltip"
title="Which driver will gain/lose the most places in comparison to last season's results?"> title="Which driver will gain/lose the most places in comparison to last season's results?">
{{ driver_select_with_preselect(driver_match=user_guess.most_wdc_gained, name="gainedselect", {{ driver_select_with_preselect(driver_match=user_guess.most_wdc_gained, name="gainedselect",
label="Most WDC pl. gained:", include_none=false, drivers=model.drivers_for_wdc_gained(), label="Most WDC pl. gained:", include_none=false, drivers=model.active_drivers_for_wdc_gained(),
disabled=not season_guess_open, disabled=not season_guess_open,
border=("border-success" if points.most_gained_correct(user.name) else "")) }} border=("border-success" if points.most_gained_correct(user.name) else "")) }}
{{ driver_select_with_preselect(driver_match=user_guess.most_wdc_lost, name="lostselect", {{ driver_select_with_preselect(driver_match=user_guess.most_wdc_lost, name="lostselect",
@ -96,8 +95,8 @@
winners:</h6> winners:</h6>
<div class="grid mt-2 container" style="row-gap: 0;"> <div class="grid mt-2 container" style="row-gap: 0;">
{% for team in model.all_teams(include_none=false) %} {% for team in model.all_teams(include_none=false) %}
{% set driver_a = model.drivers_by(team_name=team.name)[0] %} {% set driver_a = model.drivers_by(team_name=team.name, include_inactive=False)[0] %}
{% set driver_b = model.drivers_by(team_name=team.name)[1] %} {% set driver_b = model.drivers_by(team_name=team.name, include_inactive=False)[1] %}
<div class="g-col-6"> <div class="g-col-6">
<div class="form-check form-check-inline"> <div class="form-check form-check-inline">
@ -132,8 +131,8 @@
title="Which driver will reach at least a single podium?">Drivers with podium(s):</h6> title="Which driver will reach at least a single podium?">Drivers with podium(s):</h6>
<div class="grid mt-2 container" style="row-gap: 0;"> <div class="grid mt-2 container" style="row-gap: 0;">
{% for team in model.all_teams(include_none=false) %} {% for team in model.all_teams(include_none=false) %}
{% set driver_a = model.drivers_by(team_name=team.name)[0] %} {% set driver_a = model.drivers_by(team_name=team.name, include_inactive=False)[0] %}
{% set driver_b = model.drivers_by(team_name=team.name)[1] %} {% set driver_b = model.drivers_by(team_name=team.name, include_inactive=False)[1] %}
<div class="g-col-6"> <div class="g-col-6">
<div class="form-check form-check-inline"> <div class="form-check form-check-inline">

View File

@ -6,16 +6,6 @@
{% block body %} {% block body %}
<div class="card shadow-sm mb-2">
<div class="card-header">
Note
</div>
<div class="card-body">
Points from sprints and fastest laps are not tracked currently.
</div>
</div>
<div class="grid card-grid-2"> <div class="grid card-grid-2">
<div class="card shadow-sm mb-2"> <div class="card shadow-sm mb-2">
@ -37,7 +27,7 @@
</thead> </thead>
<tbody> <tbody>
{% for driver in points.drivers_sorted_by_points() %} {% for driver in points.drivers_sorted_by_points(include_inactive=True) %}
{% set driver_standing = points.wdc_standing_by_driver()[driver.name] %} {% set driver_standing = points.wdc_standing_by_driver()[driver.name] %}
<tr class="{% if driver_standing == 1 %}table-danger{% endif %}"> <tr class="{% if driver_standing == 1 %}table-danger{% endif %}">
<td class="text-center text-nowrap">{{ driver_standing }}</td> <td class="text-center text-nowrap">{{ driver_standing }}</td>

View File

@ -3,6 +3,12 @@ numpy
flask flask
flask-sqlalchemy flask-sqlalchemy
flask-caching
sqlalchemy sqlalchemy
requests
werkzeug
fastf1
pandas
pytest pytest