Add frontend button to fetch from OpenF1

flake.nix

@@ -23,6 +23,7 @@
           flask
           flask-sqlalchemy
           sqlalchemy
+          requests
 
           pytest
         ]);
@@ -41,11 +42,26 @@
 
           # Use $1 for positional args
           commands = [
-            # {
-            #   name = "";
-            #   help = "";
-            #   command = "";
-            # }
+            {
+              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 &";
+            }
           ];
         };
       });

formula10/controller/admin_controller.py

@@ -2,12 +2,15 @@ from typing import List
 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.controller.error_controller import error_redirect
 from formula10.database.update_queries import update_race_result, update_user
 from formula10.domain.domain_model import Model
 from formula10.domain.template_model import TemplateModel
 from formula10 import app
+from formula10.openf1.model.api_session import ApiSession
+from formula10.openf1.openf1_definitions import OPENF1_SESSION_NAME_RACE
+from formula10.openf1.openf1_fetcher import fetch_openf1_driver, fetch_openf1_position, fetch_openf1_session
 
 
 @app.route("/result")
@@ -44,6 +47,17 @@ def result_enter_post(race_name: str) -> Response:
     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: ApiSession = fetch_openf1_session(OPENF1_SESSION_NAME_RACE, "KSA")
+    fetch_openf1_driver(session.session_key, "VER")
+    fetch_openf1_position(session.session_key, 1)
+
+    # @todo Fetch stuff and build the race_result using update_race_result(...)
+
+    return redirect("/result")
+
+
 @app.route("/user")
 def user_root() -> str:
     model = TemplateModel(active_user_name=None,

formula10/openf1/model/api_driver.py

@@ -0,0 +1,55 @@
+from typing import Any, Callable, Dict
+
+
+class ApiDriver():
+    __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("ApiDriver:", 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

formula10/openf1/model/api_position.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from typing import Any, Callable, Dict
+
+
+class ApiPosition():
+    __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("ApiPosition:", 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

formula10/openf1/model/api_session.py

@@ -0,0 +1,58 @@
+from datetime import datetime, time
+from typing import Any, Callable, Dict
+
+
+class ApiSession():
+    __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("ApiSession:", 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

formula10/openf1/openf1_definitions.py

@@ -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"

formula10/openf1/openf1_fetcher.py

@@ -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.api_driver import ApiDriver
+from formula10.openf1.model.api_position import ApiPosition
+from formula10.openf1.model.api_session import ApiSession
+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 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 fetch_openf1_latest_session(session_name: str) -> ApiSession:
+    # ApiSession object only supports integer session_keys
+    response: List[Dict[str, str]] = request_helper(OPENF1_SESSION_ENDPOINT, {
+        "session_key": "latest",
+        "session_type": OPENF1_SESSION_TYPE_RACE,
+        "session_name": session_name
+    })
+
+    return ApiSession(response[0])
+
+
+def fetch_openf1_latest_race_session_key() -> int:
+    return fetch_openf1_latest_session(OPENF1_SESSION_NAME_RACE).session_key
+
+
+def fetch_openf1_latest_sprint_session_key() -> int:
+    return fetch_openf1_latest_session(OPENF1_SESSION_NAME_SPRINT).session_key
+
+
+def fetch_openf1_session(session_name: str, country_code: str) -> ApiSession:
+    _session: ApiSession = ApiSession(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]] = request_helper(OPENF1_SESSION_ENDPOINT, _session.to_params())
+
+    return ApiSession(response[0])
+
+
+def fetch_openf1_driver(session_key: int, name_acronym: str) -> ApiDriver:
+    _driver: ApiDriver = ApiDriver(None)
+    _driver.name_acronym = name_acronym
+    _driver.session_key = session_key
+
+    response: List[Dict[str, str]] = request_helper(OPENF1_DRIVER_ENDPOINT, _driver.to_params())
+
+    return ApiDriver(response[0])
+
+
+def fetch_openf1_position(session_key: int, position: int):
+    _position: ApiPosition = ApiPosition(None)
+    _position.session_key = session_key
+    _position.position = position
+
+    response: List[Dict[str, str]] = 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 ApiPosition(max(response, key=predicate))

formula10/templates/result.jinja

@@ -45,10 +45,25 @@
     {% 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 %}
 
+    <form action="{{ action_fetch_href }}" method="post">
+        <div class="card shadow-sm mb-2 w-100">
+            <div class="card-header">
+                OpenF1
+            </div>
+
+            <div class="card-body">
+                <input type="submit" class="btn btn-danger mt-2 w-100" value="Fetch from OpenF1"
+                       {% if race_result_open == false %}disabled="disabled"{% endif %}>
+            </div>
+        </div>
+    </form>
+
     <form class="grid card-grid" action="{{ action_save_href }}" method="post">
 
         {# Race result #}