Implement fetching of certain data from OpenF1

formula10/openf1/model/api_driver.py

@@ -17,7 +17,10 @@ class ApiDriver():
         "team_name": str
     }
 
-    def __init__(self, response: dict[str, 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})")
@@ -29,6 +32,13 @@ class ApiDriver():
 
         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

formula10/openf1/model/api_position.py

@@ -11,7 +11,10 @@ class ApiPosition():
         "position": int
     }
 
-    def __init__(self, response: dict[str, 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})")
@@ -23,6 +26,13 @@ class ApiPosition():
 
         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

formula10/openf1/model/api_session.py

@@ -20,7 +20,10 @@ class ApiSession():
         "year": int
     }
 
-    def __init__(self, response: dict[str, 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})")
@@ -32,6 +35,13 @@ class ApiSession():
 
         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

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,73 @@
+import json
+from typing import Any, 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())
+
+    return ApiPosition(response[0])