Compare commits
31 Commits
openf1
...
f38a5f2e6d
| Author | SHA1 | Date | |
|---|---|---|---|
| f38a5f2e6d | |||
| 95760baebf | |||
| 34434cc7cc | |||
| 6da5919f86 | |||
| 852f0a04ca | |||
| aa38d0138f | |||
| a9ab892c2a | |||
| e6157893ae | |||
| b2a01ce453 | |||
| 5234e8a06e | |||
| a965c0007b | |||
| 9acced3899 | |||
| ed0c5f51b4 | |||
| ebb862c9d7 | |||
| 4a0eb5329c | |||
| 30bf475000 | |||
| d66087142d | |||
| 8bdb3b5207 | |||
| 2e685ed646 | |||
| b4c459ffe7 | |||
| 3340b77efe | |||
| 0cff53cf42 | |||
| f15bbb4b96 | |||
| d88ae9568f | |||
| 1ab92eff9a | |||
| c2838c3332 | |||
| e6a70365f3 | |||
| 6e20275dcd | |||
| 10cf690ee2 | |||
| 628e2cd99a | |||
| e57109caf4 |
@ -1,10 +1,17 @@
|
||||
# syntax=docker/dockerfile:1
|
||||
|
||||
FROM public.ecr.aws/docker/library/python:3.11.8-slim-bookworm
|
||||
|
||||
RUN apt-get update -y
|
||||
|
||||
WORKDIR /app
|
||||
COPY requirements.txt requirements.txt
|
||||
|
||||
RUN pip3 install -r requirements.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
EXPOSE 5000
|
||||
|
||||
ENV FASTF1_CACHE="/cache"
|
||||
CMD ["python3", "-u", "-m", "flask", "--app", "formula10", "run", "--host", "0.0.0.0"]
|
||||
|
||||
153
flake.nix
153
flake.nix
@ -5,48 +5,143 @@
|
||||
inputs.flake-utils.url = "github:numtide/flake-utils";
|
||||
inputs.devshell.url = "github:numtide/devshell";
|
||||
|
||||
outputs = { self, nixpkgs, flake-utils, devshell }:
|
||||
flake-utils.lib.eachDefaultSystem (system:
|
||||
let
|
||||
pkgs = import nixpkgs {
|
||||
inherit system;
|
||||
config.allowUnfree = true;
|
||||
overlays = [ devshell.overlays.default ];
|
||||
outputs = {
|
||||
self,
|
||||
nixpkgs,
|
||||
flake-utils,
|
||||
devshell,
|
||||
}:
|
||||
flake-utils.lib.eachDefaultSystem (system: let
|
||||
pkgs = import nixpkgs {
|
||||
inherit system;
|
||||
config.allowUnfree = true;
|
||||
overlays = [devshell.overlays.default];
|
||||
};
|
||||
|
||||
timple = pkgs.python311Packages.buildPythonPackage rec {
|
||||
pname = "timple";
|
||||
version = "0.1.8";
|
||||
|
||||
src = pkgs.python311Packages.fetchPypi {
|
||||
inherit pname version;
|
||||
hash = "sha256-u8EgMA8BA6OpPlSg0ASRxLcIcv5psRIEcBpIicagXw8=";
|
||||
};
|
||||
|
||||
myPython = pkgs.python311.withPackages (p: with p; [
|
||||
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
|
||||
rich
|
||||
numpy
|
||||
|
||||
# Web
|
||||
flask
|
||||
flask-sqlalchemy
|
||||
flask-caching
|
||||
sqlalchemy
|
||||
requests
|
||||
|
||||
# Test
|
||||
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 {
|
||||
devShell = pkgs.devshell.mkShell {
|
||||
name = "Formula10";
|
||||
in {
|
||||
devShell = pkgs.devshell.mkShell {
|
||||
name = "Formula10";
|
||||
|
||||
packages = with pkgs; [
|
||||
myPython
|
||||
packages = with pkgs; [
|
||||
myPython
|
||||
|
||||
nodejs_21
|
||||
nodePackages.sass
|
||||
nodePackages.postcss-cli
|
||||
nodePackages.autoprefixer
|
||||
];
|
||||
sqlitebrowser
|
||||
|
||||
# Use $1 for positional args
|
||||
commands = [
|
||||
# {
|
||||
# name = "";
|
||||
# help = "";
|
||||
# command = "";
|
||||
# }
|
||||
];
|
||||
};
|
||||
});
|
||||
nodejs_21
|
||||
nodePackages.sass
|
||||
nodePackages.postcss-cli
|
||||
nodePackages.autoprefixer
|
||||
];
|
||||
|
||||
# Use $1 for positional args
|
||||
commands = [
|
||||
{
|
||||
name = "vscode";
|
||||
help = "Launch VSCode";
|
||||
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 &";
|
||||
}
|
||||
];
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
import os
|
||||
from flask import Flask
|
||||
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)
|
||||
ENABLE_TIMING: bool = False if os.getenv("DISABLE_TIMING") == "True" else True
|
||||
@ -9,18 +11,23 @@ if not ENABLE_TIMING:
|
||||
print("- Disabled timing constraints")
|
||||
|
||||
app: Flask = Flask(__name__)
|
||||
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///formula10.db"
|
||||
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///formula10.db"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
|
||||
# Session cookie is used to propagate message to error page
|
||||
app.config['SESSION_TYPE'] = 'memcached'
|
||||
app.config['SECRET_KEY'] = 'ich stinke nach maggi'
|
||||
app.config["SESSION_TYPE"] = "memcached"
|
||||
app.config["SECRET_KEY"] = "ich stinke nach maggi"
|
||||
|
||||
app.url_map.strict_slashes = False
|
||||
|
||||
db: SQLAlchemy = SQLAlchemy()
|
||||
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
|
||||
import formula10.controller.race_controller # type: ignore
|
||||
import formula10.controller.season_controller
|
||||
@ -49,4 +56,4 @@ import formula10.controller.error_controller
|
||||
# General
|
||||
# - Adapt diagram colors to team colors
|
||||
# - Add links to the official F1 stats page (for quali/result), probably best to store entire link in DB (because they are not entirely regular)?
|
||||
# - Unit testing (as much as possible, but especially points calculation)
|
||||
# - Unit testing (as much as possible, but especially points calculation)
|
||||
|
||||
@ -3,10 +3,15 @@ from urllib.parse import unquote
|
||||
from flask import redirect, render_template, request
|
||||
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.domain.cache_invalidator import cache_invalidate_user_updated, cache_invalidate_race_result_updated
|
||||
from formula10.domain.domain_model import Model
|
||||
from formula10.domain.template_model import TemplateModel
|
||||
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")
|
||||
@ -31,9 +36,29 @@ def result_enter_post(race_name: str) -> Response:
|
||||
dnfs: List[str] = request.form.getlist("dnf-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
|
||||
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")
|
||||
@ -46,11 +71,13 @@ def user_root() -> str:
|
||||
|
||||
@app.route("/user-add", methods=["POST"])
|
||||
def user_add_post() -> Response:
|
||||
cache_invalidate_user_updated()
|
||||
username: str | None = request.form.get("select-add-user")
|
||||
return update_user(username, add=True)
|
||||
|
||||
|
||||
@app.route("/user-delete", methods=["POST"])
|
||||
def user_delete_post() -> Response:
|
||||
cache_invalidate_user_updated()
|
||||
username: str | None = request.form.get("select-delete-user")
|
||||
return update_user(username, delete=True)
|
||||
return update_user(username, delete=True)
|
||||
|
||||
@ -1,8 +1,10 @@
|
||||
from typing import List
|
||||
from urllib.parse import unquote
|
||||
from flask import redirect, render_template, request
|
||||
from werkzeug import Response
|
||||
|
||||
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.points_model import PointsModel
|
||||
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")
|
||||
dnf: str | None = request.form.get("dnfselect")
|
||||
|
||||
cache_invalidate_race_guess_updated()
|
||||
race_id: int = Model().race_by(race_name=race_name).id
|
||||
user_id: int = Model().user_by(user_name=user_name).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)
|
||||
user_name = unquote(user_name)
|
||||
|
||||
cache_invalidate_race_guess_updated()
|
||||
race_id: int = Model().race_by(race_name=race_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)
|
||||
|
||||
@ -5,6 +5,7 @@ from werkzeug import Response
|
||||
|
||||
from formula10.database.model.db_team import DbTeam
|
||||
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.model.team import NONE_TEAM
|
||||
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")
|
||||
|
||||
cache_invalidate_season_guess_updated()
|
||||
user_id: int = Model().user_by(user_name=user_name).id
|
||||
return update_season_guess(user_id, guesses, team_winner_guesses, podium_driver_guesses)
|
||||
@ -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 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)
|
||||
team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), nullable=False)
|
||||
country_code: Mapped[str] = mapped_column(String(2), nullable=False) # alpha-2 code
|
||||
active: Mapped[bool] = mapped_column(Boolean, nullable=False)
|
||||
|
||||
# Relationships
|
||||
team: Mapped[DbTeam] = relationship("DbTeam", foreign_keys=[team_id])
|
||||
@ -1,5 +1,5 @@
|
||||
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 formula10 import db
|
||||
@ -19,4 +19,6 @@ class DbRace(db.Model):
|
||||
name: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
|
||||
number: Mapped[int] = mapped_column(Integer, 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)
|
||||
@ -13,7 +13,6 @@ from formula10.database.model.db_season_guess import DbSeasonGuess
|
||||
from formula10.database.model.db_user import DbUser
|
||||
from formula10.database.validation import any_is_none, positions_are_contiguous, race_has_started
|
||||
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:
|
||||
@ -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.fastest_lap_id = 9999
|
||||
race_result.sprint_dnf_driver_ids_json = json.dumps(["9999"])
|
||||
race_result.sprint_points_json = json.dumps({"9999": "9999"})
|
||||
race_result.sprint_dnf_driver_ids_json = json.dumps([])
|
||||
race_result.sprint_points_json = json.dumps({})
|
||||
|
||||
db.session.add(race_result)
|
||||
db.session.commit()
|
||||
@ -143,21 +142,22 @@ def find_or_create_race_result(race_id: int) -> DbRaceResult:
|
||||
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):
|
||||
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
|
||||
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)
|
||||
}
|
||||
|
||||
# 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)
|
||||
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!")
|
||||
|
||||
# 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)!")
|
||||
|
||||
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.dnf_driver_ids_json = json.dumps(dnf_driver_ids_list)
|
||||
race_result.excluded_driver_ids_json = json.dumps(excluded_driver_ids_list)
|
||||
|
||||
# @todo Dummy values
|
||||
race_result.fastest_lap_id = NONE_DRIVER.id
|
||||
race_result.sprint_dnf_driver_ids_json = json.dumps([NONE_DRIVER.id])
|
||||
race_result.sprint_points_json = json.dumps({NONE_DRIVER.id: 0})
|
||||
# Extra stats for points calculation
|
||||
sprint_pxx_driver_ids: Dict[str, str] = {
|
||||
str(position + 1): driver_id for position, driver_id in enumerate(sprint_pxx_driver_ids_list)
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
|
||||
@ -76,6 +76,19 @@ def find_multiple_strict(predicate: Callable[[_T], bool], iterable: Iterable[_T]
|
||||
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:
|
||||
"""
|
||||
Find a single element in a sequence matching a predicate.
|
||||
|
||||
98
formula10/domain/cache_invalidator.py
Normal file
98
formula10/domain/cache_invalidator.py
Normal 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)
|
||||
@ -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_team import DbTeam
|
||||
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.race import Race
|
||||
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.team import NONE_TEAM, Team
|
||||
from formula10.domain.model.user import User
|
||||
from formula10 import db
|
||||
from formula10 import db, cache
|
||||
|
||||
|
||||
class Model():
|
||||
_all_users: List[User] | None = None
|
||||
_all_race_results: List[RaceResult] | None = None
|
||||
_all_race_guesses: List[RaceGuess] | None = None
|
||||
_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]:
|
||||
class Model:
|
||||
@staticmethod
|
||||
@cache.cached(timeout=None, key_prefix="domain_all_users") # Clear when adding/deleting users
|
||||
def all_users() -> List[User]:
|
||||
"""
|
||||
Returns a list of all enabled users.
|
||||
"""
|
||||
if self._all_users is None:
|
||||
self._all_users = [
|
||||
User.from_db_user(db_user)
|
||||
for db_user in db.session.query(DbUser).filter_by(enabled=True).all()
|
||||
]
|
||||
db_users = db.session.query(DbUser).filter_by(enabled=True).all()
|
||||
return [User.from_db_user(db_user) for db_user in db_users]
|
||||
|
||||
return self._all_users
|
||||
|
||||
def all_race_results(self) -> List[RaceResult]:
|
||||
@staticmethod
|
||||
@cache.cached(timeout=None, key_prefix="domain_all_race_results") # Clear when adding/updating results
|
||||
def all_race_results() -> List[RaceResult]:
|
||||
"""
|
||||
Returns a list of all race results, in descending order (most recent first).
|
||||
"""
|
||||
if self._all_race_results is None:
|
||||
self._all_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()
|
||||
]
|
||||
db_race_results = db.session.query(DbRaceResult).join(DbRaceResult.race).order_by(desc("number")).all()
|
||||
return [RaceResult.from_db_race_result(db_race_result) for db_race_result in db_race_results]
|
||||
|
||||
return self._all_race_results
|
||||
|
||||
def all_race_guesses(self) -> List[RaceGuess]:
|
||||
@staticmethod
|
||||
@cache.cached(timeout=None, key_prefix="domain_all_race_guesses") # Clear when adding/updating race guesses or users
|
||||
def all_race_guesses() -> List[RaceGuess]:
|
||||
"""
|
||||
Returns a list of all race guesses (of enabled users).
|
||||
"""
|
||||
if self._all_race_guesses is None:
|
||||
self._all_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
|
||||
]
|
||||
db_race_guesses = db.session.query(DbRaceGuess).join(DbRaceGuess.user).filter_by(enabled=True).all()
|
||||
return [RaceGuess.from_db_race_guess(db_race_guess) for db_race_guess in db_race_guesses]
|
||||
|
||||
return self._all_race_guesses
|
||||
|
||||
def all_season_guesses(self) -> List[SeasonGuess]:
|
||||
@staticmethod
|
||||
@cache.cached(timeout=None, key_prefix="domain_all_season_guesses") # Clear when adding/updating season guesses or users
|
||||
def all_season_guesses() -> List[SeasonGuess]:
|
||||
"""
|
||||
Returns a list of all season guesses (of enabled users).
|
||||
"""
|
||||
if self._all_season_guesses is None:
|
||||
self._all_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
|
||||
]
|
||||
db_season_guesses = db.session.query(DbSeasonGuess).join(DbSeasonGuess.user).filter_by(enabled=True).all()
|
||||
return [SeasonGuess.from_db_season_guess(db_season_guess) for db_season_guess in db_season_guesses]
|
||||
|
||||
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]:
|
||||
if self._all_season_guess_results is None:
|
||||
self._all_season_guess_results = [
|
||||
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]:
|
||||
@staticmethod
|
||||
@cache.cached(timeout=None, key_prefix="domain_all_races") # No cleanup, bc entered manually
|
||||
def all_races() -> List[Race]:
|
||||
"""
|
||||
Returns a list of all races, in descending order (last race first).
|
||||
"""
|
||||
if self._all_races is None:
|
||||
self._all_races = [
|
||||
Race.from_db_race(db_race)
|
||||
for db_race in db.session.query(DbRace).order_by(desc(DbRace.number)).all()
|
||||
]
|
||||
db_races = db.session.query(DbRace).order_by(desc("number")).all()
|
||||
return [Race.from_db_race(db_race) for db_race in db_races]
|
||||
|
||||
return self._all_races
|
||||
|
||||
def all_drivers(self, *, include_none: bool) -> List[Driver]:
|
||||
@staticmethod
|
||||
@cache.memoize(timeout=None) # No cleanup, bc entered manually
|
||||
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:
|
||||
self._all_drivers = [
|
||||
Driver.from_db_driver(db_driver)
|
||||
for db_driver in db.session.query(DbDriver).all()
|
||||
]
|
||||
db_drivers = db.session.query(DbDriver).all()
|
||||
drivers = [Driver.from_db_driver(db_driver) for db_driver in db_drivers]
|
||||
|
||||
if include_none:
|
||||
return self._all_drivers
|
||||
else:
|
||||
if not include_inactive:
|
||||
predicate: Callable[[Driver], bool] = lambda driver: driver.active
|
||||
drivers = find_multiple_strict(predicate, drivers)
|
||||
|
||||
if not include_none:
|
||||
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.
|
||||
"""
|
||||
if self._all_teams is None:
|
||||
self._all_teams = [
|
||||
Team.from_db_team(db_team)
|
||||
for db_team in db.session.query(DbTeam).all()
|
||||
]
|
||||
db_teams = db.session.query(DbTeam).all()
|
||||
teams = [Team.from_db_team(db_team) for db_team in db_teams]
|
||||
|
||||
if include_none:
|
||||
return self._all_teams
|
||||
else:
|
||||
if not include_none:
|
||||
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
|
||||
@ -203,6 +181,7 @@ class Model():
|
||||
"""
|
||||
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:
|
||||
# List of all guesses by a single user
|
||||
if user_name is not None and race_name is None:
|
||||
@ -252,6 +231,7 @@ class Model():
|
||||
"""
|
||||
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:
|
||||
if user_name is not None:
|
||||
predicate: Callable[[SeasonGuess], bool] = lambda guess: guess.user.name == user_name
|
||||
@ -280,34 +260,41 @@ class Model():
|
||||
# Team queries
|
||||
#
|
||||
|
||||
def none_team(self) -> Team:
|
||||
@staticmethod
|
||||
def none_team() -> Team:
|
||||
return NONE_TEAM
|
||||
|
||||
#
|
||||
# Driver queries
|
||||
#
|
||||
|
||||
def none_driver(self) -> Driver:
|
||||
@staticmethod
|
||||
def none_driver() -> Driver:
|
||||
return NONE_DRIVER
|
||||
|
||||
@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.
|
||||
"""
|
||||
return self.drivers_by(team_name=team_name)
|
||||
return self.drivers_by(team_name=team_name, include_inactive=include_inactive)
|
||||
|
||||
@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.
|
||||
"""
|
||||
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:
|
||||
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:
|
||||
drivers_by: Dict[str, List[Driver]] = dict()
|
||||
@ -316,7 +303,7 @@ class Model():
|
||||
|
||||
for team in self.all_teams(include_none=False):
|
||||
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]
|
||||
|
||||
return drivers_by
|
||||
|
||||
@ -4,7 +4,7 @@ from formula10.database.model.db_driver import DbDriver
|
||||
from formula10.domain.model.team import NONE_TEAM, Team
|
||||
|
||||
|
||||
class Driver():
|
||||
class Driver:
|
||||
@classmethod
|
||||
def from_db_driver(cls, db_driver: DbDriver):
|
||||
driver: Driver = cls()
|
||||
@ -13,6 +13,7 @@ class Driver():
|
||||
driver.abbr = db_driver.abbr
|
||||
driver.country = db_driver.country_code
|
||||
driver.team = Team.from_db_team(db_driver.team)
|
||||
driver.active = db_driver.active
|
||||
return driver
|
||||
|
||||
def to_db_driver(self) -> DbDriver:
|
||||
@ -21,6 +22,7 @@ class Driver():
|
||||
db_driver.abbr = self.abbr
|
||||
db_driver.country_code = self.country
|
||||
db_driver.team_id = self.team.name
|
||||
db_driver.active = self.active
|
||||
return db_driver
|
||||
|
||||
def __eq__(self, __value: object) -> bool:
|
||||
@ -32,11 +34,17 @@ class Driver():
|
||||
def __hash__(self) -> int:
|
||||
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
|
||||
name: str
|
||||
abbr: str
|
||||
country: str
|
||||
team: Team
|
||||
active: bool
|
||||
|
||||
@property
|
||||
def name_sanitized(self) -> str:
|
||||
@ -48,4 +56,5 @@ NONE_DRIVER.id = 0
|
||||
NONE_DRIVER.name = "None"
|
||||
NONE_DRIVER.abbr = "None"
|
||||
NONE_DRIVER.country = "NO"
|
||||
NONE_DRIVER.team = NONE_TEAM
|
||||
NONE_DRIVER.team = NONE_TEAM
|
||||
NONE_DRIVER.active = True
|
||||
@ -4,7 +4,7 @@ from urllib.parse import quote
|
||||
from formula10.database.model.db_race import DbRace
|
||||
|
||||
|
||||
class Race():
|
||||
class Race:
|
||||
@classmethod
|
||||
def from_db_race(cls, db_race: DbRace):
|
||||
race: Race = cls()
|
||||
@ -13,6 +13,8 @@ class Race():
|
||||
race.number = db_race.number
|
||||
race.date = db_race.date
|
||||
race.place_to_guess = db_race.pxx
|
||||
race.quali_date = db_race.quali_date
|
||||
race.has_sprint = db_race.has_sprint
|
||||
return race
|
||||
|
||||
def to_db_race(self) -> DbRace:
|
||||
@ -21,6 +23,8 @@ class Race():
|
||||
db_race.number = self.number
|
||||
db_race.date = self.date
|
||||
db_race.pxx = self.place_to_guess
|
||||
db_race.quali_date = self.date
|
||||
db_race.has_sprint = self.has_sprint
|
||||
return db_race
|
||||
|
||||
def __eq__(self, __value: object) -> bool:
|
||||
@ -32,12 +36,17 @@ class Race():
|
||||
def __hash__(self) -> int:
|
||||
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
|
||||
name: str
|
||||
number: int
|
||||
date: datetime
|
||||
place_to_guess: int
|
||||
quali_date: datetime
|
||||
has_sprint: bool
|
||||
|
||||
@property
|
||||
def name_sanitized(self) -> str:
|
||||
return quote(self.name)
|
||||
return quote(self.name)
|
||||
|
||||
@ -4,7 +4,7 @@ from formula10.domain.model.race import Race
|
||||
from formula10.domain.model.user import User
|
||||
|
||||
|
||||
class RaceGuess():
|
||||
class RaceGuess:
|
||||
@classmethod
|
||||
def from_db_race_guess(cls, db_race_guess: DbRaceGuess):
|
||||
race_guess: RaceGuess = cls()
|
||||
|
||||
@ -154,6 +154,8 @@ class RaceResult:
|
||||
return points_strings[position_offset]
|
||||
else:
|
||||
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}")
|
||||
|
||||
@ -174,6 +176,11 @@ class RaceResult:
|
||||
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:
|
||||
if len(self.initial_dnf) == 0:
|
||||
return NONE_DRIVER.name
|
||||
|
||||
@ -7,7 +7,7 @@ from formula10.domain.model.team import Team
|
||||
from formula10.domain.model.user import User
|
||||
|
||||
|
||||
class SeasonGuess():
|
||||
class SeasonGuess:
|
||||
@classmethod
|
||||
def from_db_season_guess(cls, db_season_guess: DbSeasonGuess):
|
||||
season_guess: SeasonGuess = cls()
|
||||
|
||||
@ -2,7 +2,7 @@ from formula10.database.model.db_season_guess_result import DbSeasonGuessResult
|
||||
from formula10.domain.model.user import User
|
||||
|
||||
|
||||
class SeasonGuessResult():
|
||||
class SeasonGuessResult:
|
||||
@classmethod
|
||||
def from_db_season_guess_result(cls, db_season_guess_result: DbSeasonGuessResult):
|
||||
season_guess_result: SeasonGuessResult = cls()
|
||||
|
||||
@ -3,7 +3,7 @@ from urllib.parse import quote
|
||||
from formula10.database.model.db_team import DbTeam
|
||||
|
||||
|
||||
class Team():
|
||||
class Team:
|
||||
@classmethod
|
||||
def from_db_team(cls, db_team: DbTeam):
|
||||
team: Team = cls()
|
||||
|
||||
@ -3,7 +3,7 @@ from urllib.parse import quote
|
||||
from formula10.database.model.db_user import DbUser
|
||||
|
||||
|
||||
class User():
|
||||
class User:
|
||||
@classmethod
|
||||
def from_db_user(cls, db_user: DbUser):
|
||||
user: User = cls()
|
||||
|
||||
@ -2,6 +2,7 @@ import json
|
||||
from typing import Any, Callable, Dict, List, overload
|
||||
import numpy as np
|
||||
|
||||
from formula10 import cache
|
||||
from formula10.domain.domain_model import Model
|
||||
from formula10.domain.model.driver import NONE_DRIVER, Driver
|
||||
from formula10.domain.model.race_guess import RaceGuess
|
||||
@ -11,6 +12,8 @@ from formula10.domain.model.season_guess_result import SeasonGuessResult
|
||||
from formula10.domain.model.team import Team
|
||||
from formula10.domain.model.user import User
|
||||
|
||||
# Guess points
|
||||
|
||||
RACE_GUESS_OFFSET_POINTS: Dict[int, int] = {
|
||||
3: 1,
|
||||
2: 3,
|
||||
@ -18,7 +21,6 @@ RACE_GUESS_OFFSET_POINTS: Dict[int, int] = {
|
||||
0: 10
|
||||
}
|
||||
RACE_GUESS_DNF_POINTS: int = 10
|
||||
|
||||
SEASON_GUESS_HOT_TAKE_POINTS: int = 10
|
||||
SEASON_GUESS_P2_POINTS: int = 10
|
||||
SEASON_GUESS_OVERTAKES_POINTS: int = 10
|
||||
@ -30,6 +32,8 @@ SEASON_GUESS_TEAMWINNER_FALSE_POINTS: int = -3
|
||||
SEASON_GUESS_PODIUMS_CORRECT_POINTS: int = 3
|
||||
SEASON_GUESS_PODIUMS_FALSE_POINTS: int = -2
|
||||
|
||||
# Driver points
|
||||
|
||||
DRIVER_RACE_POINTS: Dict[int, int] = {
|
||||
1: 25,
|
||||
2: 18,
|
||||
@ -42,6 +46,19 @@ DRIVER_RACE_POINTS: Dict[int, int] = {
|
||||
9: 2,
|
||||
10: 1
|
||||
}
|
||||
DRIVER_SPRINT_POINTS: Dict[int, int] = {
|
||||
1: 8,
|
||||
2: 7,
|
||||
3: 6,
|
||||
4: 5,
|
||||
5: 4,
|
||||
6: 3,
|
||||
7: 2,
|
||||
8: 1
|
||||
}
|
||||
DRIVER_FASTEST_LAP_POINTS: int = 1
|
||||
|
||||
# Last season results
|
||||
|
||||
WDC_STANDING_2023: Dict[str, int] = {
|
||||
"Max Verstappen": 1,
|
||||
@ -65,7 +82,6 @@ WDC_STANDING_2023: Dict[str, int] = {
|
||||
"Kevin Magnussen": 19,
|
||||
"Logan Sargeant": 21
|
||||
}
|
||||
|
||||
WCC_STANDING_2023: Dict[str, int] = {
|
||||
"Red Bull": 1,
|
||||
"Mercedes": 2,
|
||||
@ -104,131 +120,134 @@ class PointsModel(Model):
|
||||
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):
|
||||
Model.__init__(self)
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_points_per_step") # Clear when adding/updating race results or users
|
||||
def points_per_step(self) -> Dict[str, List[int]]:
|
||||
"""
|
||||
Returns a dictionary of lists, containing points per race for each user.
|
||||
"""
|
||||
if self._points_per_step is None:
|
||||
self._points_per_step = dict()
|
||||
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 = dict()
|
||||
for user in self.all_users():
|
||||
points_per_step[user.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers
|
||||
|
||||
for race_guess in self.all_race_guesses():
|
||||
user_name: str = race_guess.user.name
|
||||
race_number: int = race_guess.race.number
|
||||
race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name)
|
||||
for race_guess in self.all_race_guesses():
|
||||
user_name: str = race_guess.user.name
|
||||
race_number: int = race_guess.race.number
|
||||
race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name)
|
||||
|
||||
if race_result is None:
|
||||
continue
|
||||
if race_result is None:
|
||||
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
|
||||
def driver_points_per_step(self) -> Dict[str, List[int]]:
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Clear when adding/updating race results
|
||||
def driver_points_per_step(self, *, include_inactive: bool) -> Dict[str, List[int]]:
|
||||
"""
|
||||
Returns a dictionary of lists, containing points per race for each driver.
|
||||
"""
|
||||
if self._driver_points_per_step is None:
|
||||
self._driver_points_per_step = dict()
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
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 = dict()
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=include_inactive):
|
||||
driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers
|
||||
|
||||
for race_result in self.all_race_results():
|
||||
for position, driver in race_result.standing.items():
|
||||
driver_name: str = driver.name
|
||||
race_number: int = race_result.race.number
|
||||
for race_result in self.all_race_results():
|
||||
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]]:
|
||||
"""
|
||||
Returns a dictionary of lists, containing points per race for each team.
|
||||
"""
|
||||
if self._team_points_per_step is None:
|
||||
self._team_points_per_step = dict()
|
||||
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 = dict()
|
||||
for team in self.all_teams(include_none=False):
|
||||
team_points_per_step[team.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers
|
||||
|
||||
for race_result in self.all_race_results():
|
||||
for position, driver in race_result.standing.items():
|
||||
team_name: str = driver.team.name
|
||||
race_number: int = race_result.race.number
|
||||
for race_result in self.all_race_results():
|
||||
for driver in race_result.standing.values():
|
||||
team_name: str = driver.team.name
|
||||
race_number: int = race_result.race.number
|
||||
|
||||
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]:
|
||||
if self._dnfs is None:
|
||||
self._dnfs = dict()
|
||||
dnfs = dict()
|
||||
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
self._dnfs[driver.name] = 0
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=True):
|
||||
dnfs[driver.name] = 0
|
||||
|
||||
for race_result in self.all_race_results():
|
||||
for driver in race_result.all_dnfs:
|
||||
self._dnfs[driver.name] += 1
|
||||
for race_result in self.all_race_results():
|
||||
for driver in race_result.all_dnfs:
|
||||
dnfs[driver.name] += 1
|
||||
|
||||
return self._dnfs
|
||||
for driver in race_result.sprint_dnfs:
|
||||
dnfs[driver.name] += 1
|
||||
|
||||
return dnfs
|
||||
|
||||
#
|
||||
# Driver stats
|
||||
#
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_driver_points_per_step_cumulative") # Cleanup when adding/updating race results
|
||||
def driver_points_per_step_cumulative(self) -> Dict[str, List[int]]:
|
||||
"""
|
||||
Returns a dictionary of lists, containing cumulative points per race for each driver.
|
||||
"""
|
||||
points_per_step_cumulative: Dict[str, List[int]] = dict()
|
||||
for driver_name, points in self.driver_points_per_step().items():
|
||||
for driver_name, points in self.driver_points_per_step(include_inactive=True).items():
|
||||
points_per_step_cumulative[driver_name] = np.cumsum(points).tolist()
|
||||
|
||||
return points_per_step_cumulative
|
||||
|
||||
@overload
|
||||
def driver_points_by(self, *, driver_name: str) -> List[int]:
|
||||
def driver_points_by(self, *, driver_name: str, include_inactive: bool) -> List[int]:
|
||||
"""
|
||||
Returns a list of points per race for a specific driver.
|
||||
"""
|
||||
return self.driver_points_by(driver_name=driver_name)
|
||||
return self.driver_points_by(driver_name=driver_name, include_inactive=include_inactive)
|
||||
|
||||
@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.
|
||||
"""
|
||||
return self.driver_points_by(race_name=race_name)
|
||||
return self.driver_points_by(race_name=race_name, include_inactive=include_inactive)
|
||||
|
||||
@overload
|
||||
def driver_points_by(self, *, driver_name: str, race_name: str) -> int:
|
||||
def driver_points_by(self, *, driver_name: str, race_name: str, include_inactive: bool) -> int:
|
||||
"""
|
||||
Returns the points for a specific race for a specific driver.
|
||||
"""
|
||||
return self.driver_points_by(driver_name=driver_name, race_name=race_name)
|
||||
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:
|
||||
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:
|
||||
race_number: int = self.race_by(race_name=race_name).number
|
||||
points_by_race: Dict[str, int] = dict()
|
||||
|
||||
for _driver_name, points in self.driver_points_per_step().items():
|
||||
for _driver_name, points in self.driver_points_per_step(include_inactive=include_inactive).items():
|
||||
points_by_race[_driver_name] = points[race_number]
|
||||
|
||||
return points_by_race
|
||||
@ -236,27 +255,30 @@ class PointsModel(Model):
|
||||
if driver_name is not None and race_name is not None:
|
||||
race_number: int = self.race_by(race_name=race_name).number
|
||||
|
||||
return self.driver_points_per_step()[driver_name][race_number]
|
||||
return self.driver_points_per_step(include_inactive=include_inactive)[driver_name][race_number]
|
||||
|
||||
raise Exception("driver_points_by received an illegal combination of arguments")
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
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(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
def drivers_sorted_by_points(self, *, include_inactive: bool) -> List[Driver]:
|
||||
comparator: Callable[[Driver], int] = lambda driver: self.total_driver_points_by(driver.name)
|
||||
return sorted(self.all_drivers(include_none=False), key=comparator, reverse=True)
|
||||
return sorted(self.all_drivers(include_none=False, include_inactive=include_inactive), key=comparator, reverse=True)
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_wdc_standing_by_position") # Cleanup when adding/updating race results
|
||||
def wdc_standing_by_position(self) -> Dict[int, List[str]]:
|
||||
standing: Dict[int, List[str]] = dict()
|
||||
|
||||
for position in range(1, len(self.all_drivers(include_none=False)) + 1):
|
||||
for position in range(1, len(self.all_drivers(include_none=False, include_inactive=True)) + 1):
|
||||
standing[position] = list()
|
||||
|
||||
position: int = 1
|
||||
last_points: int = 0
|
||||
|
||||
for driver in self.drivers_sorted_by_points():
|
||||
for driver in self.drivers_sorted_by_points(include_inactive=True):
|
||||
points: int = self.total_driver_points_by(driver.name)
|
||||
if points < last_points:
|
||||
position += 1
|
||||
@ -266,13 +288,14 @@ class PointsModel(Model):
|
||||
|
||||
return standing
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_wdc_standing_by_driver") # Cleanup when adding/updating race results
|
||||
def wdc_standing_by_driver(self) -> Dict[str, int]:
|
||||
standing: Dict[str, int] = dict()
|
||||
|
||||
position: int = 1
|
||||
last_points: int = 0
|
||||
|
||||
for driver in self.drivers_sorted_by_points():
|
||||
for driver in self.drivers_sorted_by_points(include_inactive=True):
|
||||
points: int = self.total_driver_points_by(driver.name)
|
||||
if points < last_points:
|
||||
position += 1
|
||||
@ -283,8 +306,12 @@ class PointsModel(Model):
|
||||
return standing
|
||||
|
||||
def wdc_diff_2023_by(self, driver_name: str) -> int:
|
||||
if not driver_name in WDC_STANDING_2023:
|
||||
return 0
|
||||
|
||||
return WDC_STANDING_2023[driver_name] - self.wdc_standing_by_driver()[driver_name]
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_most_dnf_names") # Cleanup when adding/updating race results
|
||||
def most_dnf_names(self) -> List[str]:
|
||||
dnf_names: List[str] = list()
|
||||
most_dnfs: int = 0
|
||||
@ -299,17 +326,18 @@ class PointsModel(Model):
|
||||
|
||||
return dnf_names
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_most_gained_names") # Cleanup when adding/updating race results
|
||||
def most_gained_names(self) -> List[str]:
|
||||
most_gained_names: List[str] = list()
|
||||
most_gained: int = 0
|
||||
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=True):
|
||||
gained: int = self.wdc_diff_2023_by(driver.name)
|
||||
|
||||
if gained > most_gained:
|
||||
most_gained = gained
|
||||
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=True):
|
||||
gained: int = self.wdc_diff_2023_by(driver.name)
|
||||
|
||||
if gained == most_gained:
|
||||
@ -317,17 +345,18 @@ class PointsModel(Model):
|
||||
|
||||
return most_gained_names
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_most_lost_names") # Cleanup when adding/updating race results
|
||||
def most_lost_names(self) -> List[str]:
|
||||
most_lost_names: List[str] = list()
|
||||
most_lost: int = 100
|
||||
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=True):
|
||||
lost: int = self.wdc_diff_2023_by(driver.name)
|
||||
|
||||
if lost < most_lost:
|
||||
most_lost = lost
|
||||
|
||||
for driver in self.all_drivers(include_none=False):
|
||||
for driver in self.all_drivers(include_none=False, include_inactive=True):
|
||||
lost: int = self.wdc_diff_2023_by(driver.name)
|
||||
|
||||
if lost == most_lost:
|
||||
@ -339,6 +368,7 @@ class PointsModel(Model):
|
||||
# Team points
|
||||
#
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_team_points_per_step_cumulative") # Cleanup when adding/updating race results
|
||||
def team_points_per_step_cumulative(self) -> Dict[str, List[int]]:
|
||||
"""
|
||||
Returns a dictionary of lists, containing cumulative points per race for each team.
|
||||
@ -349,14 +379,17 @@ class PointsModel(Model):
|
||||
|
||||
return points_per_step_cumulative
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
def total_team_points_by(self, team_name: str) -> int:
|
||||
teammates: List[Driver] = self.drivers_by(team_name=team_name)
|
||||
return sum(self.driver_points_by(driver_name=teammates[0].name)) + sum(self.driver_points_by(driver_name=teammates[1].name))
|
||||
teammates: List[Driver] = self.drivers_by(team_name=team_name, include_inactive=True)
|
||||
return sum(sum(self.driver_points_by(driver_name=teammate.name, include_inactive=True)) for teammate in teammates)
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_teams_sorted_by_points") # Cleanup when adding/updating race results
|
||||
def teams_sorted_by_points(self) -> List[Team]:
|
||||
comparator: Callable[[Team], int] = lambda team: self.total_team_points_by(team.name)
|
||||
return sorted(self.all_teams(include_none=False), key=comparator, reverse=True)
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_wcc_standing_by_position") # Cleanup when adding/updating race results
|
||||
def wcc_standing_by_position(self) -> Dict[int, List[str]]:
|
||||
standing: Dict[int, List[str]] = dict()
|
||||
|
||||
@ -376,6 +409,7 @@ class PointsModel(Model):
|
||||
|
||||
return standing
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_wcc_standing_by_team") # Cleanup when adding/updating race results
|
||||
def wcc_standing_by_team(self) -> Dict[str, int]:
|
||||
standing: Dict[str, int] = dict()
|
||||
|
||||
@ -430,6 +464,7 @@ class PointsModel(Model):
|
||||
"""
|
||||
return self.points_by(user_name=user_name, race_name=race_name)
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results or users
|
||||
def points_by(self, *, user_name: str | None = None, race_name: str | None = None) -> List[int] | Dict[str, int] | int:
|
||||
if user_name is not None and race_name is None:
|
||||
return self.points_per_step()[user_name]
|
||||
@ -463,6 +498,7 @@ class PointsModel(Model):
|
||||
comparator: Callable[[User], int] = lambda user: self.total_points_by(user.name)
|
||||
return sorted(self.all_users(), key=comparator, reverse=True)
|
||||
|
||||
@cache.cached(timeout=None, key_prefix="points_user_standing") # Cleanup when adding/updating race results or users
|
||||
def user_standing(self) -> Dict[str, int]:
|
||||
standing: Dict[str, int] = dict()
|
||||
|
||||
@ -482,6 +518,7 @@ class PointsModel(Model):
|
||||
# Treat standing + dnf picks separately
|
||||
return len(self.race_guesses_by(user_name=user_name)) * 2
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
def picks_with_points_count(self, user_name: str) -> int:
|
||||
count: int = 0
|
||||
|
||||
@ -549,14 +586,14 @@ class PointsModel(Model):
|
||||
|
||||
return season_guess.most_wdc_lost.name in self.most_lost_names()
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
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]
|
||||
|
||||
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] <= self.wdc_standing_by_driver()[teammate.name]
|
||||
|
||||
@cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results
|
||||
def has_podium(self, driver: Driver) -> bool:
|
||||
for race_result in self.all_race_results():
|
||||
position: int | None = race_result.driver_standing_position(driver)
|
||||
@ -600,7 +637,7 @@ class PointsModel(Model):
|
||||
"label": driver.abbr,
|
||||
"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)
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
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.model.driver import Driver
|
||||
@ -29,10 +29,12 @@ class TemplateModel(Model):
|
||||
if active_result_race_name is not None:
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
def race_result_open(self, race_name: str) -> bool:
|
||||
@ -52,6 +54,7 @@ class TemplateModel(Model):
|
||||
|
||||
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:
|
||||
"""
|
||||
Returns the first race-object with no associated race result.
|
||||
@ -69,6 +72,14 @@ class TemplateModel(Model):
|
||||
def current_race(self) -> Race | None:
|
||||
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:
|
||||
if self.active_result is not None:
|
||||
return self.active_result.race.name
|
||||
@ -85,9 +96,12 @@ class TemplateModel(Model):
|
||||
else:
|
||||
return self.all_races()[0].name_sanitized
|
||||
|
||||
def all_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)
|
||||
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, 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
|
||||
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))
|
||||
|
||||
0
formula10/fastf1/__init__.py
Normal file
0
formula10/fastf1/__init__.py
Normal file
364
formula10/fastf1/fastf1_definitions.py
Normal file
364
formula10/fastf1/fastf1_definitions.py
Normal 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"]
|
||||
0
formula10/openf1/__init__.py
Normal file
0
formula10/openf1/__init__.py
Normal file
0
formula10/openf1/model/__init__.py
Normal file
0
formula10/openf1/model/__init__.py
Normal file
55
formula10/openf1/model/openf1_driver.py
Normal file
55
formula10/openf1/model/openf1_driver.py
Normal 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
|
||||
40
formula10/openf1/model/openf1_position.py
Normal file
40
formula10/openf1/model/openf1_position.py
Normal 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
|
||||
60
formula10/openf1/model/openf1_session.py
Normal file
60
formula10/openf1/model/openf1_session.py
Normal 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
|
||||
9
formula10/openf1/openf1_definitions.py
Normal file
9
formula10/openf1/openf1_definitions.py
Normal 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"
|
||||
76
formula10/openf1/openf1_fetcher.py
Normal file
76
formula10/openf1/openf1_fetcher.py
Normal 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))
|
||||
@ -24,4 +24,4 @@
|
||||
grid-row-gap: 0;
|
||||
grid-column-gap: 8px;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -30,7 +30,7 @@
|
||||
<option value="" selected disabled hidden></option>
|
||||
|
||||
{% 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 %}
|
||||
|
||||
{% for driver in drivers %}
|
||||
@ -49,7 +49,7 @@
|
||||
{% set user_has_chosen = namespace(driverpre=false) %}
|
||||
|
||||
{% 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 %}
|
||||
|
||||
{% for driver in drivers %}
|
||||
|
||||
@ -42,90 +42,176 @@
|
||||
|
||||
{% 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">
|
||||
{{ model.active_result_race_name_or_current_race_name() }}
|
||||
</div>
|
||||
|
||||
<div class="card-body">
|
||||
<div class="d-inline-block overflow-x-scroll w-100">
|
||||
<div style="width: 410px;">
|
||||
{% 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 %}
|
||||
<div style="width: 460px;">
|
||||
|
||||
{# Place numbers #}
|
||||
<ul class="list-group list-group-flush d-inline-block">
|
||||
{% for driver in model.all_active_drivers_or_active_result_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_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;">
|
||||
{# Fastest lap #}
|
||||
<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"
|
||||
value="{{ driver.id }}"
|
||||
id="first-dnf-{{ driver.id }}" name="first-dnf-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.initial_dnf) %}checked{% endif %}
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
<label for="first-dnf-{{ driver.id }}"
|
||||
class="form-check-label text-muted">1. DNF</label>
|
||||
</div>
|
||||
|
||||
{# 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="dnf-{{ driver.id }}" name="dnf-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.all_dnfs) %}checked{% endif %}
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
<label for="dnf-{{ driver.id }}"
|
||||
class="form-check-label text-muted">DNF</label>
|
||||
</div>
|
||||
|
||||
{# Driver Excluded #}
|
||||
<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="exclude-{{ driver.id }}" name="excluded-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.standing_exclusions) %}checked{% endif %}
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
<label for="exclude-{{ driver.id }}"
|
||||
class="form-check-label text-muted" data-bs-toggle="tooltip"
|
||||
title="Driver is not counted for standing">NC</label>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{# Standing order #}
|
||||
<input type="hidden" name="pxx-drivers" value="{{ driver.id }}">
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
|
||||
<input type="submit" class="btn btn-danger mt-2 w-100" value="Save"
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
</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;">
|
||||
|
||||
<form action="{{ action_save_href }}" method="post">
|
||||
{# Place numbers #}
|
||||
<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_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 #}
|
||||
{# 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_drivers_or_active_result_standing_drivers() %}
|
||||
{% 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 at first #}
|
||||
<div class="form-check form-check-reverse d-inline-block">
|
||||
<input type="checkbox" class="form-check-input"
|
||||
value="{{ driver.id }}"
|
||||
id="first-dnf-{{ driver.id }}" name="first-dnf-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.initial_dnf) %}checked{% endif %}
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
<label for="first-dnf-{{ driver.id }}"
|
||||
class="form-check-label text-muted">1. DNF</label>
|
||||
</div>
|
||||
|
||||
{# 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"
|
||||
value="{{ driver.id }}"
|
||||
id="dnf-{{ driver.id }}" name="dnf-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.all_dnfs) %}checked{% endif %}
|
||||
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="dnf-{{ driver.id }}"
|
||||
<label for="sprint-dnf-{{ driver.id }}"
|
||||
class="form-check-label text-muted">DNF</label>
|
||||
</div>
|
||||
|
||||
{# Driver Excluded #}
|
||||
<div class="form-check form-check-reverse d-inline-block">
|
||||
<input type="checkbox" class="form-check-input"
|
||||
value="{{ driver.id }}"
|
||||
id="exclude-{{ driver.id }}" name="excluded-drivers"
|
||||
{% if (model.active_result is not none) and (driver in model.active_result.standing_exclusions) %}checked{% endif %}
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
<label for="exclude-{{ driver.id }}"
|
||||
class="form-check-label text-muted" data-bs-toggle="tooltip"
|
||||
title="Driver is not counted for standing">NC</label>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{# Standing order #}
|
||||
<input type="hidden" name="pxx-drivers" value="{{ driver.id }}">
|
||||
<input type="hidden" name="sprint-pxx-drivers" value="{{ driver.id }}">
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
|
||||
<input type="submit" class="btn btn-danger mt-2 w-100" value="Save"
|
||||
{% if race_result_open == false %}disabled="disabled"{% endif %}>
|
||||
</form>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
</form>
|
||||
|
||||
{% endblock body %}
|
||||
@ -18,7 +18,6 @@
|
||||
<div class="card-body">
|
||||
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>
|
||||
Points from sprints and fastest laps are not tracked currently.
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@ -82,7 +81,7 @@
|
||||
<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?">
|
||||
{{ 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,
|
||||
border=("border-success" if points.most_gained_correct(user.name) else "")) }}
|
||||
{{ driver_select_with_preselect(driver_match=user_guess.most_wdc_lost, name="lostselect",
|
||||
@ -96,8 +95,8 @@
|
||||
winners:</h6>
|
||||
<div class="grid mt-2 container" style="row-gap: 0;">
|
||||
{% for team in model.all_teams(include_none=false) %}
|
||||
{% set driver_a = model.drivers_by(team_name=team.name)[0] %}
|
||||
{% set driver_b = model.drivers_by(team_name=team.name)[1] %}
|
||||
{% set driver_a = model.drivers_by(team_name=team.name, include_inactive=False)[0] %}
|
||||
{% set driver_b = model.drivers_by(team_name=team.name, include_inactive=False)[1] %}
|
||||
|
||||
<div class="g-col-6">
|
||||
<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>
|
||||
<div class="grid mt-2 container" style="row-gap: 0;">
|
||||
{% for team in model.all_teams(include_none=false) %}
|
||||
{% set driver_a = model.drivers_by(team_name=team.name)[0] %}
|
||||
{% set driver_b = model.drivers_by(team_name=team.name)[1] %}
|
||||
{% set driver_a = model.drivers_by(team_name=team.name, include_inactive=False)[0] %}
|
||||
{% set driver_b = model.drivers_by(team_name=team.name, include_inactive=False)[1] %}
|
||||
|
||||
<div class="g-col-6">
|
||||
<div class="form-check form-check-inline">
|
||||
|
||||
@ -6,16 +6,6 @@
|
||||
|
||||
{% 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="card shadow-sm mb-2">
|
||||
@ -37,7 +27,7 @@
|
||||
</thead>
|
||||
|
||||
<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] %}
|
||||
<tr class="{% if driver_standing == 1 %}table-danger{% endif %}">
|
||||
<td class="text-center text-nowrap">{{ driver_standing }}</td>
|
||||
|
||||
@ -3,6 +3,12 @@ numpy
|
||||
|
||||
flask
|
||||
flask-sqlalchemy
|
||||
flask-caching
|
||||
sqlalchemy
|
||||
requests
|
||||
werkzeug
|
||||
|
||||
fastf1
|
||||
pandas
|
||||
|
||||
pytest
|
||||
|
||||
Reference in New Issue
Block a user